From cdf6aab08552c70ea02d9d76ee5e8ef3565ef9cc Mon Sep 17 00:00:00 2001 From: RIchard Baah Date: Sun, 16 Aug 2026 12:14:09 -0400 Subject: [PATCH 1/4] introduce new dictionary benchmarks --- .../benches/dictionary_group_values.rs | 87 ++++++++++++++++++- 1 file changed, 85 insertions(+), 2 deletions(-) diff --git a/datafusion/physical-plan/benches/dictionary_group_values.rs b/datafusion/physical-plan/benches/dictionary_group_values.rs index ded52aebd1100..b175c4d392bc7 100644 --- a/datafusion/physical-plan/benches/dictionary_group_values.rs +++ b/datafusion/physical-plan/benches/dictionary_group_values.rs @@ -29,7 +29,7 @@ use criterion::{ }; use datafusion_expr::EmitTo; use datafusion_physical_plan::aggregates::group_values::new_group_values; -use datafusion_physical_plan::aggregates::order::GroupOrdering; +use datafusion_physical_plan::aggregates::order::{GroupOrdering, GroupOrderingFull}; use rand::rngs::StdRng; use rand::seq::SliceRandom; use rand::{Rng, SeedableRng}; @@ -172,5 +172,88 @@ fn bench_repeated_intern_emit(c: &mut Criterion) { group.finish(); } -criterion_group!(benches, bench_intern_emit, bench_repeated_intern_emit); +// GroupOrdering::Full -> GroupValuesColumn::: scalar append_val/equal_to path. +fn bench_scalar_append_equal(c: &mut Criterion) { + let mut group = c.benchmark_group("dict_scalar_append_equal"); + let schema = dict_schema(); + let null_density = 0.1; + let size = SIZES[1]; + + let mut cards = CARDS_RELATIVE.to_vec(); + cards.push(size); + for cardinality in cards { + let array = make_dict(size, cardinality, null_density, SEED); + group.throughput(Throughput::Elements(size as u64)); + group.bench_function( + bench_id("scalar_append_equal", size, cardinality, null_density), + |b| { + b.iter_batched_ref( + || { + ( + new_group_values( + schema.clone(), + &GroupOrdering::Full(GroupOrderingFull::new()), + ) + .unwrap(), + Vec::::with_capacity(size), + ) + }, + |(gv, groups)| { + gv.intern(std::slice::from_ref(&array), groups).unwrap(); + black_box(&*groups); + black_box(gv.emit(EmitTo::All).unwrap()); + }, + BatchSize::SmallInput, + ); + }, + ); + } + group.finish(); +} + +// EmitTo::First exercises the take-n path; two interns + partial emit per iteration. +fn bench_take_n(c: &mut Criterion) { + let mut group = c.benchmark_group("dict_take_n"); + let schema = dict_schema(); + let null_density = 0.10; + let size = SIZES[1]; + + let mut cards = CARDS_RELATIVE.to_vec(); + cards.push(size); + for cardinality in cards { + let batch_a = make_dict(size, cardinality, null_density, SEED); + let batch_b = make_dict(size, cardinality, null_density, SEED.wrapping_add(1)); + group.throughput(Throughput::Elements((size * 2 * N_BATCHES) as u64)); + group.bench_function(bench_id("take_n", size, cardinality, null_density), |b| { + b.iter_batched_ref( + || { + ( + new_group_values(schema.clone(), &GroupOrdering::None).unwrap(), + Vec::::with_capacity(size), + ) + }, + |(gv, groups)| { + for _ in 0..N_BATCHES { + gv.intern(std::slice::from_ref(&batch_a), groups).unwrap(); + black_box(&*groups); + gv.intern(std::slice::from_ref(&batch_b), groups).unwrap(); + black_box(&*groups); + black_box(gv.emit(EmitTo::First(1)).unwrap()); + } + black_box(gv.emit(EmitTo::All).unwrap()); + }, + BatchSize::SmallInput, + ); + }); + } + group.finish(); +} + +criterion_group!( + benches, + bench_intern_emit, + bench_repeated_intern_emit, + bench_scalar_append_equal, + bench_take_n +); criterion_main!(benches); From 529295b9e6c3678582a428020bbb728e2853d3aa Mon Sep 17 00:00:00 2001 From: RIchard Baah Date: Sun, 16 Aug 2026 12:20:27 -0400 Subject: [PATCH 2/4] tweak benchmarks --- .../benches/dictionary_group_values.rs | 13 +++++-------- 1 file changed, 5 insertions(+), 8 deletions(-) diff --git a/datafusion/physical-plan/benches/dictionary_group_values.rs b/datafusion/physical-plan/benches/dictionary_group_values.rs index b175c4d392bc7..c0dafccd7316d 100644 --- a/datafusion/physical-plan/benches/dictionary_group_values.rs +++ b/datafusion/physical-plan/benches/dictionary_group_values.rs @@ -221,9 +221,8 @@ fn bench_take_n(c: &mut Criterion) { let mut cards = CARDS_RELATIVE.to_vec(); cards.push(size); for cardinality in cards { - let batch_a = make_dict(size, cardinality, null_density, SEED); - let batch_b = make_dict(size, cardinality, null_density, SEED.wrapping_add(1)); - group.throughput(Throughput::Elements((size * 2 * N_BATCHES) as u64)); + let batch = make_dict(size, cardinality, null_density, SEED); + group.throughput(Throughput::Elements((size * N_BATCHES) as u64)); group.bench_function(bench_id("take_n", size, cardinality, null_density), |b| { b.iter_batched_ref( || { @@ -234,13 +233,11 @@ fn bench_take_n(c: &mut Criterion) { }, |(gv, groups)| { for _ in 0..N_BATCHES { - gv.intern(std::slice::from_ref(&batch_a), groups).unwrap(); + gv.intern(std::slice::from_ref(&batch), groups).unwrap(); black_box(&*groups); - gv.intern(std::slice::from_ref(&batch_b), groups).unwrap(); - black_box(&*groups); - black_box(gv.emit(EmitTo::First(1)).unwrap()); + black_box(gv.emit(EmitTo::First(size / 2)).unwrap()); } - black_box(gv.emit(EmitTo::All).unwrap()); + black_box(gv.emit(EmitTo::First(gv.len())).unwrap()); }, BatchSize::SmallInput, ); From 4a7a6649f983b12444a481dd37b16aa4733418ed Mon Sep 17 00:00:00 2001 From: RIchard Baah Date: Sun, 16 Aug 2026 12:35:45 -0400 Subject: [PATCH 3/4] impl --- .../group_values/multi_group_by/dictionary.rs | 24 ++++++++++++++++--- 1 file changed, 21 insertions(+), 3 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 501b13d0cd183..717aab5d0010d 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 @@ -57,6 +57,9 @@ pub struct DictionaryGroupValuesColumn val_to_inner: Vec, /// Reusable hash buffer for the dictionary values array. val_hashes: Vec, + /// 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. + cached_values: Option, _phantom: PhantomData, } @@ -73,6 +76,7 @@ impl DictionaryGroupValuesColumn { random_state: AGGREGATION_HASH_SEED, val_to_inner: Vec::default(), val_hashes: Vec::default(), + cached_values: None, _phantom: PhantomData, } } @@ -159,6 +163,7 @@ impl DictionaryGroupValuesColumn { } fn hash_values(&mut self, values: &ArrayRef) { + self.cached_values = None; self.val_hashes.clear(); self.val_hashes.resize(values.len(), 0); create_hashes( @@ -296,9 +301,22 @@ impl GroupColumn } Some(val_idx) => { let dict_values = dict.values(); - let single = dict_values.slice(val_idx, 1); - self.hash_values(&single); - self.find_or_insert_value(dict_values, val_idx, self.val_hashes[0])? + 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.find_or_insert_value(dict_values, val_idx, self.val_hashes[val_idx])? } }; self.group_to_inner.push(inner_slot); From fb9f491df85adb9569fd1623f3aeb9bad64b4fcd Mon Sep 17 00:00:00 2001 From: RIchard Baah Date: Tue, 18 Aug 2026 16:31:47 -0400 Subject: [PATCH 4/4] add comment explaining arc cache --- .../benches/dictionary_group_values.rs | 84 +------------------ .../group_values/multi_group_by/dictionary.rs | 3 + 2 files changed, 5 insertions(+), 82 deletions(-) diff --git a/datafusion/physical-plan/benches/dictionary_group_values.rs b/datafusion/physical-plan/benches/dictionary_group_values.rs index c0dafccd7316d..ded52aebd1100 100644 --- a/datafusion/physical-plan/benches/dictionary_group_values.rs +++ b/datafusion/physical-plan/benches/dictionary_group_values.rs @@ -29,7 +29,7 @@ use criterion::{ }; use datafusion_expr::EmitTo; use datafusion_physical_plan::aggregates::group_values::new_group_values; -use datafusion_physical_plan::aggregates::order::{GroupOrdering, GroupOrderingFull}; +use datafusion_physical_plan::aggregates::order::GroupOrdering; use rand::rngs::StdRng; use rand::seq::SliceRandom; use rand::{Rng, SeedableRng}; @@ -172,85 +172,5 @@ fn bench_repeated_intern_emit(c: &mut Criterion) { group.finish(); } -// GroupOrdering::Full -> GroupValuesColumn::: scalar append_val/equal_to path. -fn bench_scalar_append_equal(c: &mut Criterion) { - let mut group = c.benchmark_group("dict_scalar_append_equal"); - let schema = dict_schema(); - let null_density = 0.1; - let size = SIZES[1]; - - let mut cards = CARDS_RELATIVE.to_vec(); - cards.push(size); - for cardinality in cards { - let array = make_dict(size, cardinality, null_density, SEED); - group.throughput(Throughput::Elements(size as u64)); - group.bench_function( - bench_id("scalar_append_equal", size, cardinality, null_density), - |b| { - b.iter_batched_ref( - || { - ( - new_group_values( - schema.clone(), - &GroupOrdering::Full(GroupOrderingFull::new()), - ) - .unwrap(), - Vec::::with_capacity(size), - ) - }, - |(gv, groups)| { - gv.intern(std::slice::from_ref(&array), groups).unwrap(); - black_box(&*groups); - black_box(gv.emit(EmitTo::All).unwrap()); - }, - BatchSize::SmallInput, - ); - }, - ); - } - group.finish(); -} - -// EmitTo::First exercises the take-n path; two interns + partial emit per iteration. -fn bench_take_n(c: &mut Criterion) { - let mut group = c.benchmark_group("dict_take_n"); - let schema = dict_schema(); - let null_density = 0.10; - let size = SIZES[1]; - - let mut cards = CARDS_RELATIVE.to_vec(); - cards.push(size); - for cardinality in cards { - let batch = make_dict(size, cardinality, null_density, SEED); - group.throughput(Throughput::Elements((size * N_BATCHES) as u64)); - group.bench_function(bench_id("take_n", size, cardinality, null_density), |b| { - b.iter_batched_ref( - || { - ( - new_group_values(schema.clone(), &GroupOrdering::None).unwrap(), - Vec::::with_capacity(size), - ) - }, - |(gv, groups)| { - for _ in 0..N_BATCHES { - gv.intern(std::slice::from_ref(&batch), groups).unwrap(); - black_box(&*groups); - black_box(gv.emit(EmitTo::First(size / 2)).unwrap()); - } - black_box(gv.emit(EmitTo::First(gv.len())).unwrap()); - }, - BatchSize::SmallInput, - ); - }); - } - group.finish(); -} - -criterion_group!( - benches, - bench_intern_emit, - bench_repeated_intern_emit, - bench_scalar_append_equal, - bench_take_n -); +criterion_group!(benches, bench_intern_emit, bench_repeated_intern_emit); criterion_main!(benches); 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 717aab5d0010d..d0c167e1d2d59 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 @@ -301,6 +301,9 @@ impl 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()