You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/09/15 12:02:55 UTC
(datafusion) branch gh-readonly-queue/main/pr-25185-7917a9a65a0343a6b5536299de5482a899d77c51 created (now 4a3cc9d94d)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-25185-7917a9a65a0343a6b5536299de5482a899d77c51
in repository https://gitbox.apache.org/repos/asf/datafusion.git
at 4a3cc9d94d perf: reuse cached dictionary value hashes in vectorized_append (#25185)
This branch includes the following new commits:
new 4a3cc9d94d perf: reuse cached dictionary value hashes in vectorized_append (#25185)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion) 01/01: perf: reuse cached dictionary value hashes in vectorized_append (#25185)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-25185-7917a9a65a0343a6b5536299de5482a899d77c51
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 4a3cc9d94d58e13885b701a6e9cd378ac5460c73
Author: Lía Adriana <li...@datadoghq.com>
AuthorDate: Tue Sep 15 12:02:42 2026 +0000
perf: reuse cached dictionary value hashes in vectorized_append (#25185)
## Which issue does this PR close?
- Closes #.
## Rationale for this change
`vectorized_append` is the path taken when we have non streaming hash
aggregations (when there is no ordering in the `group by`), right now it
re hashed the entire dictionary values on every batch unlike append_val
which that only does this if the values array was already seen before.
## What changes are included in this PR?
`vectorized_append` now reuses the cached value hashes when a batch
carries the same dictionary values array as the previous one (this is
the common case of a repartition or filter), instead of re hashing the
whole dictionary and re-initialising the map on every batch. This
extends the existing `Arc::ptr_eq` cache from #24418, which only covered
the scalar `append_val` path.
## What is the testing strategy for this PR?
All dict tests pass, I have a PR for a benchmark that exercises this
path -> https://github.com/apache/datafusion/pull/25198
## Are there any user-facing changes?
no, this is a pure perf PR
---
.../group_values/multi_group_by/dictionary.rs | 194 ++++++++++++++++++---
1 file changed, 168 insertions(+), 26 deletions(-)
diff --git a/datafusion/physical-plan/src/aggregates/group_values/multi_group_by/dictionary.rs b/datafusion/physical-plan/src/aggregates/group_values/multi_group_by/dictionary.rs
index b190c17297..45a7c15bfd 100644
--- a/datafusion/physical-plan/src/aggregates/group_values/multi_group_by/dictionary.rs
+++ b/datafusion/physical-plan/src/aggregates/group_values/multi_group_by/dictionary.rs
@@ -54,12 +54,18 @@ pub struct DictionaryGroupValuesColumn<K: ArrowDictionaryKeyType + Send + Sync>
null_inner_slot: Option<usize>,
/// Hash seed — must match `create_hashes` so hashes are consistent across calls.
random_state: RandomState,
- /// Reusable scratch buffer mapping `val_idx → inner_slot` across batches.
+ /// Maps `val_idx → inner_slot` for the values array in `cached_values`.
+ ///
+ /// Valid only while `cached_values` is `Some` and matches the incoming
+ /// values array; `usize::MAX` means "not yet resolved". Entries are only
+ /// ever filled in, never invalidated, because `inner` slots are stable
+ /// under append. `take_n` remaps `inner` and therefore drops the cache.
val_to_inner: Vec<usize>,
- /// Reusable hash buffer for the dictionary values array.
+ /// Hashes of the values array in `cached_values`, one per `val_idx`.
val_hashes: Vec<u64>,
- /// The last `dict.values()` Arc hashed in `append_val`. When the incoming
- /// values array is `ptr_eq` to this, `val_hashes` can be reused directly.
+ /// The values array that `val_hashes` and `val_to_inner` were built for.
+ /// When an incoming `dict.values()` is `Arc::ptr_eq` to this, both caches
+ /// are reused instead of being rebuilt. See [`Self::sync_value_cache`].
cached_values: Option<ArrayRef>,
_phantom: PhantomData<K>,
}
@@ -175,6 +181,28 @@ impl<K: ArrowDictionaryKeyType + Send + Sync> DictionaryGroupValuesColumn<K> {
.unwrap();
}
+ /// Makes `val_hashes` and `val_to_inner` correspond to `dict_values`,
+ /// reusing them when they were already built for this same values array.
+ ///
+ /// `take`, `filter` and repartitioning clone the values `Arc` and rewrite
+ /// only the keys, so batches in a partition usually share one values
+ /// array; without this the per-batch cost is O(dictionary cardinality)
+ /// rather than O(rows). `take_n` remaps `inner` and clears
+ /// `cached_values`, forcing a rebuild.
+ fn sync_value_cache(&mut self, dict_values: &ArrayRef) {
+ if self
+ .cached_values
+ .as_ref()
+ .is_some_and(|c| Arc::ptr_eq(c, dict_values))
+ {
+ return;
+ }
+ self.hash_values(dict_values);
+ self.val_to_inner.clear();
+ self.val_to_inner.resize(dict_values.len(), usize::MAX);
+ self.cached_values = Some(Arc::clone(dict_values));
+ }
+
fn find_or_insert_value(
&mut self,
dict_values: &ArrayRef,
@@ -302,24 +330,7 @@ impl<K: ArrowDictionaryKeyType + Send + Sync> GroupColumn
}
Some(val_idx) => {
let dict_values = dict.values();
- // check if the dictionary values array we are hashing was already seen.
- // if its arc was already stored we dont need to rehash the entire array again
- // if its new hash the entire array and store an arc ptr for future use
- let cache_hit = self
- .cached_values
- .as_ref()
- .is_some_and(|c| Arc::ptr_eq(c, dict_values));
- if !cache_hit {
- self.val_hashes.clear();
- self.val_hashes.resize(dict_values.len(), 0);
- create_hashes(
- std::slice::from_ref(dict_values),
- &self.random_state,
- &mut self.val_hashes,
- )
- .unwrap();
- self.cached_values = Some(Arc::clone(dict_values));
- }
+ self.sync_value_cache(dict_values);
self.find_or_insert_value(dict_values, val_idx, self.val_hashes[val_idx])?
}
};
@@ -401,11 +412,12 @@ impl<K: ArrowDictionaryKeyType + Send + Sync> GroupColumn
let dict = array.as_dictionary::<K>();
let dict_keys = dict.keys();
let dict_values = dict.values();
- let num_distinct = dict_values.len();
- self.hash_values(dict_values);
- self.val_to_inner.clear();
- self.val_to_inner.resize(num_distinct, usize::MAX);
+ // Reuses `val_hashes`/`val_to_inner` when this batch carries the same
+ // values array as the last one, which is the common case downstream of
+ // a repartition or filter. Previously this unconditionally re-hashed
+ // and re-initialised both buffers, costing O(D) per batch.
+ self.sync_value_cache(dict_values);
self.group_to_inner.try_reserve(rows.len()).map_err(|e| {
DataFusionError::ArrowError(
@@ -918,4 +930,134 @@ mod tests {
assert_eq!(str_values(&out)[127], Some("v127".into()));
assert_eq!(out.as_dictionary::<Int8Type>().values().len(), 128);
}
+
+ fn dict_with_values(keys: &[Option<i32>], values: &ArrayRef) -> ArrayRef {
+ Arc::new(DictionaryArray::<Int32Type>::new(
+ Int32Array::from(keys.to_vec()),
+ Arc::clone(values),
+ ))
+ }
+
+ /// A cache rebuild resets every `val_to_inner` entry to `usize::MAX`, so
+ /// an entry resolved by an earlier batch surviving a later one proves the
+ /// dictionary was not re-hashed in between.
+ #[test]
+ fn cache_survives_batches_sharing_a_values_array() {
+ let values: ArrayRef =
+ Arc::new(StringArray::from(vec![Some("a"), Some("b"), Some("c")]));
+ let mut col = utf8_col();
+
+ col.vectorized_append(&dict_with_values(&[Some(0)], &values), &[0])
+ .unwrap();
+ let slot_a = col.val_to_inner[0];
+ assert_ne!(slot_a, usize::MAX);
+
+ col.vectorized_append(&dict_with_values(&[Some(1)], &values), &[0])
+ .unwrap();
+
+ assert_eq!(col.val_to_inner[0], slot_a, "cache was rebuilt");
+ assert_ne!(col.val_to_inner[1], usize::MAX);
+ assert_eq!(
+ col.val_to_inner[2],
+ usize::MAX,
+ "unreferenced value resolved"
+ );
+ assert!(
+ col.cached_values
+ .as_ref()
+ .is_some_and(|c| Arc::ptr_eq(c, &values))
+ );
+ assert_eq!(col.inner.len(), 2);
+ }
+
+ /// `take` clones the values `Arc` and rewrites only the keys, which is the
+ /// property the `ptr_eq` check relies on. Pin that arrow behaviour.
+ #[test]
+ fn take_preserves_values_arc_so_cache_hits() {
+ let values: ArrayRef = Arc::new(StringArray::from(vec![Some("a"), Some("b")]));
+ let batch = dict_with_values(&[Some(0), Some(1), Some(0)], &values);
+ let taken = take(&*batch, &Int32Array::from(vec![2, 0]), None).unwrap();
+ assert!(
+ Arc::ptr_eq(taken.as_dictionary::<Int32Type>().values(), &values),
+ "take must not copy the values array"
+ );
+
+ let mut col = utf8_col();
+ col.vectorized_append(&batch, &[0, 1, 2]).unwrap();
+ let slot_b = col.val_to_inner[1];
+ col.vectorized_append(&taken, &[0, 1]).unwrap();
+
+ assert_eq!(col.val_to_inner[1], slot_b, "cache was rebuilt");
+ assert_eq!(col.inner.len(), 2);
+ }
+
+ /// A genuinely different values array must rebuild the cache, and equal
+ /// values across the two arrays must still dedup to one inner slot.
+ #[test]
+ fn distinct_values_arcs_rebuild_cache_but_still_dedup() {
+ let mut col = utf8_col();
+ let first = i32_dict(&[Some(0)], &[Some("a")]);
+ let second = i32_dict(&[Some(0)], &[Some("a")]);
+ col.vectorized_append(&first, &[0]).unwrap();
+ col.vectorized_append(&second, &[0]).unwrap();
+
+ assert!(
+ col.cached_values.as_ref().is_some_and(|c| Arc::ptr_eq(
+ c,
+ second.as_dictionary::<Int32Type>().values()
+ )),
+ "cache should track the most recent values array"
+ );
+ assert_eq!(col.inner.len(), 1, "'a' must dedup across both arrays");
+ assert_eq!(col.len(), 2);
+ }
+
+ /// `take_n` remaps inner slots, so the cached `val_idx → inner_slot` map
+ /// must not survive it.
+ #[test]
+ fn take_n_invalidates_value_cache() {
+ let values: ArrayRef = Arc::new(StringArray::from(vec![Some("a"), Some("b")]));
+ let mut col = utf8_col();
+ col.vectorized_append(&dict_with_values(&[Some(0), Some(1)], &values), &[0, 1])
+ .unwrap();
+
+ let _emitted = col.take_n(1);
+ assert!(
+ col.cached_values.is_none(),
+ "take_n remaps inner slots and must drop the cache"
+ );
+
+ col.vectorized_append(&dict_with_values(&[Some(0), Some(1)], &values), &[0, 1])
+ .unwrap();
+
+ let out = Box::new(col).build();
+ let d = out.as_dictionary::<Int32Type>();
+ let vals = d.values().as_string::<i32>();
+ let got: Vec<&str> = d
+ .keys()
+ .iter()
+ .map(|k| vals.value(k.unwrap() as usize))
+ .collect();
+ assert_eq!(got, vec!["b", "a", "b"]);
+ }
+
+ /// `append_val` fills `val_hashes` but not `val_to_inner`; mixing it with
+ /// `vectorized_append` must not duplicate a value or reuse a stale slot.
+ #[test]
+ fn scalar_and_vectorized_append_share_cache_consistently() {
+ let values: ArrayRef = Arc::new(StringArray::from(vec![Some("a"), Some("b")]));
+ let batch = dict_with_values(&[Some(0), Some(1)], &values);
+ let mut col = utf8_col();
+
+ col.append_val(&batch, 0).unwrap();
+ assert!(
+ col.cached_values
+ .as_ref()
+ .is_some_and(|c| Arc::ptr_eq(c, &values))
+ );
+
+ col.vectorized_append(&batch, &[0, 1]).unwrap();
+ assert_eq!(col.inner.len(), 2, "'a' must not be duplicated");
+ assert_eq!(col.len(), 3);
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org