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/05/23 09:34:28 UTC
(datafusion) 01/01: Benchmark multi-column GROUP BY performance (#22322)
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-22322-fa8903af95b6f3ba9b07e578f1dc8d51fdbfa334
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 47046581d0f0e66c3351400b0fd344d93e6165c2
Author: Nathan <56...@users.noreply.github.com>
AuthorDate: Sat May 23 05:34:06 2026 -0400
Benchmark multi-column GROUP BY performance (#22322)
## Which issue does this PR close?
- Measuring: https://github.com/apache/datafusion/issues/17850
## Rationale for this change
1. Measure multi-column GROUP BY performance with a fair
apples-to-apples comparison
2. The default vectorized per-column approach (`GroupValuesColumn`)
underperforms the row-based approach (`GroupValuesRows`) when distinct
group count is small relative
to input rows. The benchmark confirms this: row-based is 16-19% faster
below ~200K groups, while vectorized wins by 15-33% above ~500K groups.
## What changes are included in this PR?
Adds a benchmark in `datafusion/physical-plan/benches/multi_group_by.rs`
that directly calls `GroupValues::intern()` with identical Int32 data
for both implementations —
no SQL/planning/IO overhead, same schema, same hashing.
Makes `mod row` public so the benchmark can instantiate
`GroupValuesRows` directly.
Test cases:
- **Issue #17850 reproduction** (3 cols, 64 groups, 1M-50M rows) —
confirms row-based wins ~16-19%
- **Low cardinality sweep** (8-4096 groups, 3-4 cols) — row-based wins
15-38%
- **Batch size sensitivity** (1K-32K) — minimal effect on ratio
- **Column scaling with low groups** (2-10 cols) — row-based advantage
grows with columns
- **High cardinality scaling** (1M groups, 2-10 cols) — vectorized wins
22-43%
- **Group count sweep** (16 to 1M groups, 4 cols) — crossover at
~200K-500K groups
## Are these changes tested?
- `cargo fmt --all`
- `cargo clippy -p datafusion-physical-plan --bench multi_group_by -- -D
warnings`
- `cargo bench -p datafusion-physical-plan --bench multi_group_by`
## Are there any user-facing changes?
No. This adds a benchmark only.
---------
Co-authored-by: Nathan Bezualem <nb...@amazon.com>
Co-authored-by: Claude Opus 4.6 (1M context) <no...@anthropic.com>
---
datafusion/physical-plan/Cargo.toml | 4 +
datafusion/physical-plan/benches/multi_group_by.rs | 356 +++++++++++++++++++++
.../src/aggregates/group_values/mod.rs | 4 +-
.../aggregates/group_values/multi_group_by/mod.rs | 2 +-
4 files changed, 363 insertions(+), 3 deletions(-)
diff --git a/datafusion/physical-plan/Cargo.toml b/datafusion/physical-plan/Cargo.toml
index 465fc86cfb..5a05173eb3 100644
--- a/datafusion/physical-plan/Cargo.toml
+++ b/datafusion/physical-plan/Cargo.toml
@@ -121,3 +121,7 @@ required-features = ["test_utils"]
[[bench]]
harness = false
name = "dictionary_group_values"
+
+[[bench]]
+harness = false
+name = "multi_group_by"
diff --git a/datafusion/physical-plan/benches/multi_group_by.rs b/datafusion/physical-plan/benches/multi_group_by.rs
new file mode 100644
index 0000000000..92d0448775
--- /dev/null
+++ b/datafusion/physical-plan/benches/multi_group_by.rs
@@ -0,0 +1,356 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Benchmarks for multi-column GROUP BY performance comparing vectorized
+//! (`GroupValuesColumn`) vs row-based (`GroupValuesRows`) implementations.
+//!
+//! Motivated by <https://github.com/apache/datafusion/issues/17850> which
+//! showed vectorized can regress for low-cardinality, high-row-count scenarios.
+//!
+//! Uses the direct `GroupValues::intern()` API with identical Int32 data for
+//! both implementations — a fair apples-to-apples comparison with the same
+//! hashing and data layout.
+
+use arrow::array::{ArrayRef, Int32Array};
+use arrow::datatypes::{DataType, Field, Schema, SchemaRef};
+use criterion::{BenchmarkId, Criterion, criterion_group, criterion_main};
+use datafusion_physical_plan::aggregates::group_values::GroupValues;
+use datafusion_physical_plan::aggregates::group_values::GroupValuesRows;
+use datafusion_physical_plan::aggregates::group_values::multi_group_by::GroupValuesColumn;
+use std::hint::black_box;
+use std::sync::Arc;
+
+const DEFAULT_BATCH_SIZE: usize = 8192;
+
+fn make_schema(num_cols: usize) -> SchemaRef {
+ let fields: Vec<Field> = (0..num_cols)
+ .map(|i| Field::new(format!("col_{i}"), DataType::Int32, false))
+ .collect();
+ Arc::new(Schema::new(fields))
+}
+
+fn generate_batches(
+ num_cols: usize,
+ num_distinct_groups: usize,
+ num_rows: usize,
+ batch_size: usize,
+) -> Vec<Vec<ArrayRef>> {
+ let per_col_card = (num_distinct_groups as f64)
+ .powf(1.0 / num_cols as f64)
+ .ceil() as usize;
+
+ let num_full_batches = num_rows / batch_size;
+ let remainder = num_rows % batch_size;
+ let num_batches = num_full_batches + if remainder > 0 { 1 } else { 0 };
+
+ (0..num_batches)
+ .map(|batch_idx| {
+ let batch_start = batch_idx * batch_size;
+ let current_batch_size = if batch_idx == num_batches - 1 && remainder > 0 {
+ remainder
+ } else {
+ batch_size
+ };
+ (0..num_cols)
+ .map(|col_idx| {
+ let values: Vec<i32> = (0..current_batch_size)
+ .map(|row| {
+ let global_row = batch_start + row;
+ let group_id = global_row % num_distinct_groups;
+ let divisor = per_col_card.pow(col_idx as u32);
+ ((group_id / divisor) % per_col_card) as i32
+ })
+ .collect();
+ Arc::new(Int32Array::from(values)) as ArrayRef
+ })
+ .collect()
+ })
+ .collect()
+}
+
+fn create_group_values(schema: &SchemaRef, vectorized: bool) -> Box<dyn GroupValues> {
+ if vectorized {
+ Box::new(GroupValuesColumn::<false>::try_new(Arc::clone(schema)).unwrap())
+ } else {
+ Box::new(GroupValuesRows::try_new(Arc::clone(schema)).unwrap())
+ }
+}
+
+fn bench_intern(
+ gv: &mut Box<dyn GroupValues>,
+ batches: &[Vec<ArrayRef>],
+ groups: &mut Vec<usize>,
+) {
+ for batch in batches {
+ groups.clear();
+ gv.intern(batch, groups).unwrap();
+ }
+ black_box(&*groups);
+}
+
+/// Experiment 1: Issue #17850 regression scenario.
+/// 3 columns, 64 groups (4^3), scaling row count.
+fn bench_issue_17850_regression(c: &mut Criterion) {
+ let mut group = c.benchmark_group("issue_17850_regression");
+ group.sample_size(10);
+
+ let num_cols = 3;
+ let num_groups = 64;
+ let schema = make_schema(num_cols);
+
+ for num_rows in [1_000_000, 5_000_000, 10_000_000, 20_000_000, 50_000_000] {
+ let batches =
+ generate_batches(num_cols, num_groups, num_rows, DEFAULT_BATCH_SIZE);
+
+ for vectorized in [true, false] {
+ let label = if vectorized {
+ "vectorized"
+ } else {
+ "row_based"
+ };
+ group.bench_with_input(
+ BenchmarkId::new(label, format!("{num_rows}_rows")),
+ &batches,
+ |b, batches| {
+ b.iter_batched_ref(
+ || {
+ (
+ create_group_values(&schema, vectorized),
+ Vec::<usize>::with_capacity(DEFAULT_BATCH_SIZE),
+ )
+ },
+ |(gv, groups)| bench_intern(gv, batches, groups),
+ criterion::BatchSize::LargeInput,
+ );
+ },
+ );
+ }
+ }
+ group.finish();
+}
+
+/// Experiment 2: Low cardinality sweep.
+fn bench_low_cardinality(c: &mut Criterion) {
+ let mut group = c.benchmark_group("low_cardinality");
+ group.sample_size(15);
+
+ for (num_cols, per_col_card) in
+ [(3usize, 2usize), (3, 4), (3, 8), (4, 2), (4, 4), (4, 8)]
+ {
+ let num_groups = per_col_card.pow(num_cols as u32);
+ let schema = make_schema(num_cols);
+ let batches =
+ generate_batches(num_cols, num_groups, 1_000_000, DEFAULT_BATCH_SIZE);
+
+ for vectorized in [true, false] {
+ let label = if vectorized {
+ "vectorized"
+ } else {
+ "row_based"
+ };
+ group.bench_with_input(
+ BenchmarkId::new(
+ label,
+ format!("cols_{num_cols}_card_{per_col_card}_grp_{num_groups}"),
+ ),
+ &batches,
+ |b, batches| {
+ b.iter_batched_ref(
+ || {
+ (
+ create_group_values(&schema, vectorized),
+ Vec::<usize>::with_capacity(DEFAULT_BATCH_SIZE),
+ )
+ },
+ |(gv, groups)| bench_intern(gv, batches, groups),
+ criterion::BatchSize::LargeInput,
+ );
+ },
+ );
+ }
+ }
+ group.finish();
+}
+
+/// Experiment 3: Batch size sensitivity.
+fn bench_batch_size_sensitivity(c: &mut Criterion) {
+ let mut group = c.benchmark_group("batch_size_sensitivity");
+ group.sample_size(10);
+
+ let num_cols = 3;
+ let num_groups = 64;
+ let schema = make_schema(num_cols);
+
+ for batch_size in [1024, 4096, 8192, 16384, 32768] {
+ let batches = generate_batches(num_cols, num_groups, 1_000_000, batch_size);
+
+ for vectorized in [true, false] {
+ let label = if vectorized {
+ "vectorized"
+ } else {
+ "row_based"
+ };
+ group.bench_with_input(
+ BenchmarkId::new(label, format!("batch_{batch_size}")),
+ &batches,
+ |b, batches| {
+ b.iter_batched_ref(
+ || {
+ (
+ create_group_values(&schema, vectorized),
+ Vec::<usize>::with_capacity(batch_size),
+ )
+ },
+ |(gv, groups)| bench_intern(gv, batches, groups),
+ criterion::BatchSize::LargeInput,
+ );
+ },
+ );
+ }
+ }
+ group.finish();
+}
+
+/// Experiment 4: Column count scaling with low groups.
+fn bench_column_scaling(c: &mut Criterion) {
+ let mut group = c.benchmark_group("column_scaling");
+ group.sample_size(15);
+
+ let cases: &[(usize, usize)] =
+ &[(2, 100), (3, 125), (4, 81), (6, 729), (8, 256), (10, 1024)];
+
+ for &(num_cols, num_groups) in cases {
+ let schema = make_schema(num_cols);
+ let batches =
+ generate_batches(num_cols, num_groups, 1_000_000, DEFAULT_BATCH_SIZE);
+
+ for vectorized in [true, false] {
+ let label = if vectorized {
+ "vectorized"
+ } else {
+ "row_based"
+ };
+ group.bench_with_input(
+ BenchmarkId::new(label, format!("cols_{num_cols}_grp_{num_groups}")),
+ &batches,
+ |b, batches| {
+ b.iter_batched_ref(
+ || {
+ (
+ create_group_values(&schema, vectorized),
+ Vec::<usize>::with_capacity(DEFAULT_BATCH_SIZE),
+ )
+ },
+ |(gv, groups)| bench_intern(gv, batches, groups),
+ criterion::BatchSize::LargeInput,
+ );
+ },
+ );
+ }
+ }
+ group.finish();
+}
+
+/// Experiment 5: High cardinality column scaling (~1M groups).
+fn bench_high_cardinality_scaling(c: &mut Criterion) {
+ let mut group = c.benchmark_group("high_cardinality_scaling");
+ group.sample_size(10);
+
+ for num_cols in [2, 3, 4, 6, 8, 10] {
+ let num_groups = 1_000_000;
+ let schema = make_schema(num_cols);
+ let batches =
+ generate_batches(num_cols, num_groups, 1_000_000, DEFAULT_BATCH_SIZE);
+
+ for vectorized in [true, false] {
+ let label = if vectorized {
+ "vectorized"
+ } else {
+ "row_based"
+ };
+ group.bench_with_input(
+ BenchmarkId::new(label, format!("cols_{num_cols}_grp_1M")),
+ &batches,
+ |b, batches| {
+ b.iter_batched_ref(
+ || {
+ (
+ create_group_values(&schema, vectorized),
+ Vec::<usize>::with_capacity(DEFAULT_BATCH_SIZE),
+ )
+ },
+ |(gv, groups)| bench_intern(gv, batches, groups),
+ criterion::BatchSize::LargeInput,
+ );
+ },
+ );
+ }
+ }
+ group.finish();
+}
+
+/// Experiment 6: Group count sweep with fixed 4 columns.
+fn bench_group_count_sweep(c: &mut Criterion) {
+ let mut group = c.benchmark_group("group_count_sweep");
+ group.sample_size(15);
+
+ let num_cols = 4;
+ let schema = make_schema(num_cols);
+
+ for num_groups in [
+ 16, 64, 256, 1000, 5000, 10_000, 50_000, 100_000, 500_000, 1_000_000,
+ ] {
+ let batches =
+ generate_batches(num_cols, num_groups, 1_000_000, DEFAULT_BATCH_SIZE);
+
+ for vectorized in [true, false] {
+ let label = if vectorized {
+ "vectorized"
+ } else {
+ "row_based"
+ };
+ group.bench_with_input(
+ BenchmarkId::new(label, format!("grp_{num_groups}")),
+ &batches,
+ |b, batches| {
+ b.iter_batched_ref(
+ || {
+ (
+ create_group_values(&schema, vectorized),
+ Vec::<usize>::with_capacity(DEFAULT_BATCH_SIZE),
+ )
+ },
+ |(gv, groups)| bench_intern(gv, batches, groups),
+ criterion::BatchSize::LargeInput,
+ );
+ },
+ );
+ }
+ }
+ group.finish();
+}
+
+criterion_group!(
+ benches,
+ bench_issue_17850_regression,
+ bench_low_cardinality,
+ bench_batch_size_sensitivity,
+ bench_column_scaling,
+ bench_high_cardinality_scaling,
+ bench_group_count_sweep,
+);
+criterion_main!(benches);
diff --git a/datafusion/physical-plan/src/aggregates/group_values/mod.rs b/datafusion/physical-plan/src/aggregates/group_values/mod.rs
index 2f3b1a19e7..ee253e5d7a 100644
--- a/datafusion/physical-plan/src/aggregates/group_values/mod.rs
+++ b/datafusion/physical-plan/src/aggregates/group_values/mod.rs
@@ -31,10 +31,10 @@ use datafusion_expr::EmitTo;
pub mod multi_group_by;
mod row;
+pub use row::GroupValuesRows;
mod single_group_by;
use datafusion_physical_expr::binary_map::OutputType;
use multi_group_by::GroupValuesColumn;
-use row::GroupValuesRows;
pub(crate) use single_group_by::primitive::HashValue;
@@ -130,7 +130,7 @@ pub trait GroupValues: Send {
///
/// `GroupColumn`: crate::aggregates::group_values::multi_group_by::GroupColumn
/// `GroupValuesColumn`: crate::aggregates::group_values::multi_group_by::GroupValuesColumn
-/// `GroupValuesRows`: crate::aggregates::group_values::row::GroupValuesRows
+/// `GroupValuesRows`: crate::aggregates::group_values::GroupValuesRows
pub fn new_group_values(
schema: SchemaRef,
group_ordering: &GroupOrdering,
diff --git a/datafusion/physical-plan/src/aggregates/group_values/multi_group_by/mod.rs b/datafusion/physical-plan/src/aggregates/group_values/multi_group_by/mod.rs
index f603839bee..12d80b1f9b 100644
--- a/datafusion/physical-plan/src/aggregates/group_values/multi_group_by/mod.rs
+++ b/datafusion/physical-plan/src/aggregates/group_values/multi_group_by/mod.rs
@@ -225,7 +225,7 @@ pub struct GroupValuesColumn<const STREAMING: bool> {
/// more general purpose [`GroupValuesRows`]. See the ticket for details:
/// <https://github.com/apache/datafusion/pull/12269>
///
- /// [`GroupValuesRows`]: crate::aggregates::group_values::row::GroupValuesRows
+ /// [`GroupValuesRows`]: crate::aggregates::group_values::GroupValuesRows
group_values: Vec<Box<dyn GroupColumn>>,
/// reused buffer to store hashes
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org