(datafusion) 01/01: bench: cover dictionary group keys with a shared values array (#25198)

[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-25198-b376290aad68bac68010234cd25fcfcfbc92d6d2
in repository https://gitbox.apache.org/repos/asf/datafusion.git

commit 7917a9a65a0343a6b5536299de5482a899d77c51
Author: Lía Adriana <li...@datadoghq.com>
AuthorDate: Tue Sep 15 12:01:28 2026 +0000

    bench: cover dictionary group keys with a shared values array (#25198)
    
    ## Which issue does this PR close?
    
    [Related to #24822. No issue
    closed.](https://github.com/apache/datafusion/pull/25185)
    
    ## Rationale for this change
    
    The benchmarks in this file build a fresh values array per batch and cap
    cardinality at the batch size. Neither shape matches a real plan: `take`
    and `filter` clone the values `Arc` and rewrite only the keys, so
    batches downstream of a repartition share one values array, and because
    it is never compacted its cardinality can far exceed the rows in a
    batch. Per-batch work that scales with dictionary cardinality is
    invisible under the current parameters.
    
    ## What changes are included in this PR?
    
    Adds `bench_shared_values_arc`: 32 batches of 8192 rows sharing one
    values array, at cardinalities 8192, 100k and 500k.
    
    ## What is the testing strategy for this PR?
    
    Benchmark-only. Ran the bench locally and `cargo clippy --all-targets -D
    warnings`.
    
    ## Are there any user-facing changes?
    
    No
---
 .../benches/dictionary_group_values.rs             | 74 +++++++++++++++++++++-
 1 file changed, 73 insertions(+), 1 deletion(-)

diff --git a/datafusion/physical-plan/benches/dictionary_group_values.rs b/datafusion/physical-plan/benches/dictionary_group_values.rs
index e0cbc3577c..502869787e 100644
--- a/datafusion/physical-plan/benches/dictionary_group_values.rs
+++ b/datafusion/physical-plan/benches/dictionary_group_values.rs
@@ -247,11 +247,83 @@ fn bench_take_n(c: &mut Criterion) {
     group.finish();
 }
 
+/// Batches that share one dictionary values array, as produced downstream of
+/// a repartition or filter: `take`/`filter` clone the values `Arc` and rewrite
+/// only the keys, so the values array is never compacted and its cardinality
+/// reflects the whole upstream stream rather than a single batch.
+///
+/// The other benchmarks here always allocate a fresh values array per batch
+/// and cap cardinality at the batch size, so neither the `Arc` reuse nor the
+/// `cardinality >> rows` regime is covered by them.
+fn bench_shared_values_arc(c: &mut Criterion) {
+    // More batches than `N_BATCHES`: the cost this exercises is paid once per
+    // batch, so a longer run per values array is what a real partition looks
+    // like (thousands of batches sharing one dictionary).
+    const BATCHES: usize = 32;
+
+    let mut group = c.benchmark_group("dict_shared_values_arc");
+    let size = SIZES[0];
+
+    for &cardinality in &[size, 32 * 1024, 64 * 1024, 100_000, 500_000] {
+        // Seeded per cardinality so each configuration is reproducible on its
+        // own and adding one does not change the data used by the others.
+        let mut rng = StdRng::seed_from_u64(SEED);
+        // One values array, shared by every batch.
+        let strings: Vec<String> =
+            (0..cardinality).map(|i| format!("v_{i:08}")).collect();
+        let values: ArrayRef = Arc::new(StringArray::from(
+            strings.iter().map(|s| Some(s.as_str())).collect::<Vec<_>>(),
+        ));
+
+        let batches: Vec<ArrayRef> = (0..BATCHES)
+            .map(|_| {
+                let keys: Vec<i32> = (0..size)
+                    .map(|_| rng.random_range(0..cardinality) as i32)
+                    .collect();
+                Arc::new(DictionaryArray::<Int32Type>::new(
+                    PrimitiveArray::<Int32Type>::from(keys),
+                    Arc::clone(&values),
+                )) as ArrayRef
+            })
+            .collect();
+
+        let schema = dict_schema();
+        group.throughput(Throughput::Elements((size * BATCHES) as u64));
+        group.bench_function(
+            BenchmarkId::new(
+                "shared_values_arc",
+                format!("size_{size}_card_{cardinality}"),
+            ),
+            |b| {
+                b.iter_batched_ref(
+                    || {
+                        (
+                            new_group_values(schema.clone(), &GroupOrdering::None)
+                                .unwrap(),
+                            Vec::<usize>::with_capacity(size),
+                        )
+                    },
+                    |(gv, groups)| {
+                        for arr in &batches {
+                            gv.intern(std::slice::from_ref(arr), groups).unwrap();
+                            black_box(&*groups);
+                        }
+                        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
+    bench_take_n,
+    bench_shared_values_arc
 );
 criterion_main!(benches);


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org