jayzhan211 commented on code in PR #26816:
URL: https://github.com/apache/datafusion/pull/26816#discussion_r4454919133
##########
datafusion/common/src/hash_utils.rs:
##########
@@ -28,6 +32,134 @@ use std::hash::{BuildHasher, Hash, Hasher};
pub type RandomState = FixedState;
pub type QualityRandomState = foldhash::quality::FixedState;
+type DictionaryHashEntry = (Weak<dyn Array>, Vec<u64>);
+
+/// Hashes arrays, reusing dictionary value hashes by column when possible.
+#[derive(Debug)]
+pub struct CachingArrayHasher {
+ /// Stored with the cache so value hashes cannot cross hash seeds.
+ random_state: RandomState,
+ /// One entry per column. Weak identity avoids retaining dictionary values.
+ dictionaries: Vec<Option<DictionaryHashEntry>>,
+}
+
+impl CachingArrayHasher {
+ /// Creates one cache slot per column in `schema`.
+ ///
+ /// For each dictionary column, the cache tracks the values array with a
+ /// weak pointer, so it does not keep the values alive, and owns a reusable
+ /// value-hash buffer. [`Self::create_hashes`] reuses those hashes while
the
+ /// values `Arc` is unchanged, recomputes them when it changes, and
scatters
+ /// them through the dictionary keys into the row hashes. Non-dictionary
+ /// columns use the regular array hashing path.
+ pub fn new(schema: &Schema, random_state: RandomState) -> Self {
Review Comment:
Optional cleanup: `new` only uses the schema's length, which makes
`BatchPartitioner` build a placeholder `Schema` behind a lazy `Option`. Taking
the column count lets the partitioner create the hasher up front from
`exprs.len()`, and #26826 can pass `schema.fields().len()`. I tried this
locally and clippy and the repartition/hash_utils tests pass:
```diff
- pub fn new(schema: &Schema, random_state: RandomState) -> Self {
+ pub fn new(num_columns: usize, random_state: RandomState) -> Self {
Self {
random_state,
- dictionaries: schema.fields().iter().map(|_| None).collect(),
+ dictionaries: vec![None; num_columns],
}
}
```
```diff
- hasher: Option<CachingArrayHasher>,
+ hasher: CachingArrayHasher,
...
+ let hasher = CachingArrayHasher::new(
+ exprs.len(),
+ REPARTITION_RANDOM_STATE.random_state().clone(),
+ );
Ok(Self {
state: BatchPartitionerState::Hash {
...
- hasher: None,
+ hasher,
...
- let hasher = hasher.get_or_insert_with(|| {
- let fields = arrays
- ...
- });
hasher.create_hashes(&arrays, hash_buffer)?;
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
