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/08/04 07:21:15 UTC
(datafusion) branch main updated: bench: add nested-type (List/Struct/Map) cases to first_value/last_value benchmark (#24075)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion.git
The following commit(s) were added to refs/heads/main by this push:
new 26fbf4d015 bench: add nested-type (List/Struct/Map) cases to first_value/last_value benchmark (#24075)
26fbf4d015 is described below
commit 26fbf4d015d7071115c5fdadf1156291d8a2afb1
Author: Qi Zhu <zh...@gmail.com>
AuthorDate: Tue Aug 4 15:06:07 2026 +0800
bench: add nested-type (List/Struct/Map) cases to first_value/last_value benchmark (#24075)
## Which issue does this PR close?
Groundwork for benchmarking #23628 (native `GroupsAccumulator` for
nested value types in `first_value` / `last_value`).
## Rationale for this change
The `first_last` benchmark only covers primitive value types today.
#23628 adds a native `GroupsAccumulator` for **nested** value types
(`List`, `Struct`, `Map`), which previously fell back to one per-group
`Accumulator` via `GroupsAccumulatorAdapter`. To measure that work we
need nested-type cases in the benchmark.
Landing this first (with the fallback path on current `main`) means that
once #23628 is in flight, a `run benchmark first_last` diff shows the
fallback → native speedup per type automatically.
## What changes are included in this PR?
- Adds `List<Int64>`, `Struct<i64,utf8,f64>`, `Map<utf8,i64>` and a
composite `List<Struct<i64,utf8>>` value column, each with the same
coverage as the primitive cases: `first_value`/`last_value` update +
merge, plus `first_value` evaluate, at 0% and 90% nulls.
- `prepare_typed_groups_accumulator` now mirrors the planner: it uses
the native `GroupsAccumulator` when the value type is supported and
otherwise falls back to a `GroupsAccumulatorAdapter` around one
per-group `Accumulator`. The same benchmark case therefore runs the
fallback on a build without native nested support and the native path on
one that has it.
## Are these changes tested?
Benchmark-only. Runs locally on `main` (all nested cases exercise the
fallback path). As a preview of the intended signal, `first_value struct
update` goes ~112ms (fallback) → ~24ms (native, with #23628) at 1024
groups.
## Are there any user-facing changes?
No — benchmark only.
---
.../functions-aggregate/benches/first_last.rs | 272 +++++++++++++++++++--
1 file changed, 257 insertions(+), 15 deletions(-)
diff --git a/datafusion/functions-aggregate/benches/first_last.rs b/datafusion/functions-aggregate/benches/first_last.rs
index 8f28e126a4..235f11ff30 100644
--- a/datafusion/functions-aggregate/benches/first_last.rs
+++ b/datafusion/functions-aggregate/benches/first_last.rs
@@ -15,10 +15,16 @@
// specific language governing permissions and limitations
// under the License.
-use arrow::array::{ArrayRef, BooleanArray, Int64Array};
+use arrow::array::{
+ Array, ArrayRef, BooleanArray, Int64Array, ListArray, MapArray, StringArray,
+ StructArray,
+};
+use arrow::buffer::{NullBuffer, OffsetBuffer};
use arrow::compute::SortOptions;
-use arrow::datatypes::{DataType, Field, Int64Type, Schema};
-use arrow::util::bench_util::{create_boolean_array, create_primitive_array};
+use arrow::datatypes::{DataType, Field, Fields, Float64Type, Int64Type, Schema};
+use arrow::util::bench_util::{
+ create_boolean_array, create_primitive_array, create_string_array_with_len,
+};
use datafusion_common::instant::Instant;
use std::hint::black_box;
use std::sync::Arc;
@@ -29,14 +35,21 @@ use datafusion_expr::{
use datafusion_functions_aggregate::first_last::{
FirstValue, LastValue, TrivialFirstValueAccumulator, TrivialLastValueAccumulator,
};
+use datafusion_functions_aggregate_common::aggregate::groups_accumulator::GroupsAccumulatorAdapter;
use datafusion_physical_expr::PhysicalSortExpr;
use datafusion_physical_expr::expressions::col;
use criterion::{BatchSize, Criterion, criterion_group, criterion_main};
-fn prepare_groups_accumulator(is_first: bool) -> Box<dyn GroupsAccumulator> {
+/// Build a `GroupsAccumulator` for an arbitrary value type, so the nested-type
+/// (`Struct` / `List`) fast paths added for `first_value` / `last_value` can be
+/// exercised with the same harness as the primitive ones.
+fn prepare_typed_groups_accumulator(
+ is_first: bool,
+ value_type: DataType,
+) -> Box<dyn GroupsAccumulator> {
let schema = Arc::new(Schema::new(vec![
- Field::new("value", DataType::Int64, true),
+ Field::new("value", value_type.clone(), true),
Field::new("ord", DataType::Int64, true),
]));
@@ -46,11 +59,12 @@ fn prepare_groups_accumulator(is_first: bool) -> Box<dyn GroupsAccumulator> {
options: SortOptions::default(),
};
- let value_field: Arc<Field> = Field::new("value", DataType::Int64, true).into();
- let accumulator_args = AccumulatorArgs {
+ let value_field: Arc<Field> = Field::new("value", value_type.clone(), true).into();
+ let value_expr = col("value", &schema).unwrap();
+ let make_args = || AccumulatorArgs {
return_field: Arc::clone(&value_field),
schema: &schema,
- expr_fields: &[value_field],
+ expr_fields: std::slice::from_ref(&value_field),
ignore_nulls: false,
order_bys: std::slice::from_ref(&sort_expr),
is_reversed: false,
@@ -60,20 +74,81 @@ fn prepare_groups_accumulator(is_first: bool) -> Box<dyn GroupsAccumulator> {
"LAST_VALUE(value ORDER BY ord)"
},
is_distinct: false,
- exprs: &[col("value", &schema).unwrap()],
+ exprs: std::slice::from_ref(&value_expr),
};
+ // Mirror the planner: use the native GroupsAccumulator when this value type
+ // is supported and otherwise fall back to a GroupsAccumulatorAdapter around
+ // one per-group Accumulator. Deciding with `groups_accumulator_supported`
+ // (rather than catching `create_groups_accumulator` errors) keeps genuine
+ // construction failures loud. The same case then runs the fallback on a
+ // build without native nested support and the native path on one with it,
+ // so a before/after benchmark run surfaces the win directly.
+ let supported = if is_first {
+ FirstValue::new().groups_accumulator_supported(make_args())
+ } else {
+ LastValue::new().groups_accumulator_supported(make_args())
+ };
+ if !supported {
+ return build_fallback_adapter(is_first, value_type);
+ }
if is_first {
FirstValue::new()
- .create_groups_accumulator(accumulator_args)
+ .create_groups_accumulator(make_args())
.unwrap()
} else {
LastValue::new()
- .create_groups_accumulator(accumulator_args)
+ .create_groups_accumulator(make_args())
.unwrap()
}
}
+/// Build the *fallback* grouped accumulator for a value type: a
+/// `GroupsAccumulatorAdapter` wrapping one per-group `Accumulator`. This is
+/// exactly what nested value types (`List` / `Struct` / `Map`) used before
+/// they gained a native `GroupsAccumulator`, and it is what the planner still
+/// selects when `groups_accumulator_supported` returns `false`. Benching this
+/// side by side with `prepare_typed_groups_accumulator` (the native path)
+/// shows the win from the native `GroupsAccumulator`.
+fn build_fallback_adapter(
+ is_first: bool,
+ value_type: DataType,
+) -> Box<dyn GroupsAccumulator> {
+ Box::new(GroupsAccumulatorAdapter::new(move || {
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("value", value_type.clone(), true),
+ Field::new("ord", DataType::Int64, true),
+ ]));
+ let sort_expr = PhysicalSortExpr {
+ expr: col("ord", &schema)?,
+ options: SortOptions::default(),
+ };
+ let value_field: Arc<Field> =
+ Field::new("value", value_type.clone(), true).into();
+ let value_expr = col("value", &schema)?;
+ let accumulator_args = AccumulatorArgs {
+ return_field: Arc::clone(&value_field),
+ schema: &schema,
+ expr_fields: std::slice::from_ref(&value_field),
+ ignore_nulls: false,
+ order_bys: std::slice::from_ref(&sort_expr),
+ is_reversed: false,
+ name: if is_first {
+ "FIRST_VALUE(value ORDER BY ord)"
+ } else {
+ "LAST_VALUE(value ORDER BY ord)"
+ },
+ is_distinct: false,
+ exprs: std::slice::from_ref(&value_expr),
+ };
+ if is_first {
+ FirstValue::new().accumulator(accumulator_args)
+ } else {
+ LastValue::new().accumulator(accumulator_args)
+ }
+ }))
+}
+
fn create_trivial_accumulator(
is_first: bool,
ignore_nulls: bool,
@@ -104,11 +179,13 @@ fn evaluate_bench(
) {
let n = values.len();
let group_indices: Vec<usize> = (0..n).map(|i| i % num_groups).collect();
+ let value_type = values.data_type().clone();
c.bench_function(name, |b| {
b.iter_batched(
|| {
- let mut accumulator = prepare_groups_accumulator(is_first);
+ let mut accumulator =
+ prepare_typed_groups_accumulator(is_first, value_type.clone());
accumulator
.update_batch(
&[Arc::clone(&values), Arc::clone(&ord)],
@@ -139,6 +216,7 @@ fn update_bench(
) {
let n = values.len();
let group_indices: Vec<usize> = (0..n).map(|i| i % num_groups).collect();
+ let value_type = values.data_type().clone();
// Initialize with worst-case ordering so update_batch forces rows comparison for all groups.
let worst_ord: ArrayRef = Arc::new(Int64Array::from(vec![
@@ -153,7 +231,8 @@ fn update_bench(
c.bench_function(name, |b| {
b.iter_batched(
|| {
- let mut accumulator = prepare_groups_accumulator(is_first);
+ let mut accumulator =
+ prepare_typed_groups_accumulator(is_first, value_type.clone());
accumulator
.update_batch(
&[Arc::clone(&values), Arc::clone(&worst_ord)],
@@ -197,6 +276,7 @@ fn merge_bench(
let n = values.len();
let group_indices: Vec<usize> = (0..n).map(|i| i % num_groups).collect();
let is_set: ArrayRef = Arc::new(BooleanArray::from(vec![true; n]));
+ let value_type = values.data_type().clone();
// Initialize with worst-case ordering so update_batch forces rows comparison for all groups.
let worst_ord: ArrayRef = Arc::new(Int64Array::from(vec![
@@ -212,7 +292,8 @@ fn merge_bench(
b.iter_batched(
|| {
// Prebuild accumulator
- let mut accumulator = prepare_groups_accumulator(is_first);
+ let mut accumulator =
+ prepare_typed_groups_accumulator(is_first, value_type.clone());
accumulator
.update_batch(
&[Arc::clone(&values), Arc::clone(&worst_ord)],
@@ -270,6 +351,167 @@ fn trivial_update_bench(
});
}
+/// A top-level validity buffer with roughly `null_density` nulls, so the
+/// generated nested arrays have null *values* (not just null inner
+/// fields/elements) — matching the `nulls={pct}%` semantics of the primitive
+/// benchmarks, where the value itself is null. Returns `None` at 0% so the
+/// arrays stay fully valid. Derived from arrow's own null generator for a
+/// deterministic, density-accurate pattern.
+fn top_level_nulls(n: usize, null_density: f32) -> Option<NullBuffer> {
+ create_primitive_array::<Int64Type>(n, null_density)
+ .nulls()
+ .cloned()
+}
+
+/// A 3-field struct value column `Struct<Int64, Utf8, Float64>`. `null_density`
+/// controls both the struct-level null values and the inner field nulls.
+fn create_struct_array(n: usize, null_density: f32) -> ArrayRef {
+ let a = Arc::new(create_primitive_array::<Int64Type>(n, null_density)) as ArrayRef;
+ let b =
+ Arc::new(create_string_array_with_len::<i32>(n, null_density, 16)) as ArrayRef;
+ let d = Arc::new(create_primitive_array::<Float64Type>(n, null_density)) as ArrayRef;
+ let fields = Fields::from(vec![
+ Field::new("c0", DataType::Int64, true),
+ Field::new("c1", DataType::Utf8, true),
+ Field::new("c2", DataType::Float64, true),
+ ]);
+ Arc::new(StructArray::new(
+ fields,
+ vec![a, b, d],
+ top_level_nulls(n, null_density),
+ ))
+}
+
+/// A `List<Int64>` value column with fixed-size lists of `list_len` elements.
+fn create_list_array(n: usize, list_len: usize, null_density: f32) -> ArrayRef {
+ let child = Arc::new(create_primitive_array::<Int64Type>(
+ n * list_len,
+ null_density,
+ )) as ArrayRef;
+ let offsets = OffsetBuffer::from_lengths(std::iter::repeat_n(list_len, n));
+ let field = Arc::new(Field::new_list_field(DataType::Int64, true));
+ Arc::new(ListArray::new(
+ field,
+ offsets,
+ child,
+ top_level_nulls(n, null_density),
+ ))
+}
+
+/// A `Map<Utf8, Int64>` value column with `entries_per_row` entries per row.
+/// Values carry `null_density` nulls (keys are never null), matching the null
+/// treatment of the struct / list generators.
+fn create_map_array(n: usize, entries_per_row: usize, null_density: f32) -> ArrayRef {
+ let total = n * entries_per_row;
+ let values =
+ Arc::new(create_primitive_array::<Int64Type>(total, null_density)) as ArrayRef;
+ let keys = Arc::new(StringArray::from_iter_values(
+ (0..total).map(|idx| format!("k{}", idx % entries_per_row)),
+ )) as ArrayRef;
+ let entry_fields = Fields::from(vec![
+ Field::new("keys", DataType::Utf8, false),
+ Field::new("values", DataType::Int64, true),
+ ]);
+ let entries = StructArray::new(entry_fields.clone(), vec![keys, values], None);
+ let offsets = OffsetBuffer::from_lengths(std::iter::repeat_n(entries_per_row, n));
+ let map_field =
+ Arc::new(Field::new("entries", DataType::Struct(entry_fields), false));
+ Arc::new(MapArray::new(
+ map_field,
+ offsets,
+ entries,
+ top_level_nulls(n, null_density),
+ false,
+ ))
+}
+
+/// A composite `List<Struct<a: Int64, b: Utf8>>` column — a list whose
+/// elements are structs (the "array of records" shape). Exercises the
+/// nested-within-nested case, which the generic value-state path must also
+/// handle.
+fn create_list_of_struct_array(n: usize, list_len: usize, null_density: f32) -> ArrayRef {
+ let total = n * list_len;
+ let a =
+ Arc::new(create_primitive_array::<Int64Type>(total, null_density)) as ArrayRef;
+ let b =
+ Arc::new(create_string_array_with_len::<i32>(total, null_density, 8)) as ArrayRef;
+ let struct_fields = Fields::from(vec![
+ Field::new("a", DataType::Int64, true),
+ Field::new("b", DataType::Utf8, true),
+ ]);
+ let child =
+ Arc::new(StructArray::new(struct_fields.clone(), vec![a, b], None)) as ArrayRef;
+ let offsets = OffsetBuffer::from_lengths(std::iter::repeat_n(list_len, n));
+ let list_field =
+ Arc::new(Field::new_list_field(DataType::Struct(struct_fields), true));
+ Arc::new(ListArray::new(
+ list_field,
+ offsets,
+ child,
+ top_level_nulls(n, null_density),
+ ))
+}
+
+fn first_last_nested_benchmark(c: &mut Criterion) {
+ const N: usize = 65536;
+ const NUM_GROUPS: usize = 1024;
+
+ let ord = Arc::new(create_primitive_array::<Int64Type>(N, 0.0)) as ArrayRef;
+
+ for pct in [0, 90] {
+ let null_density = (pct as f32) / 100.0;
+
+ // One column per nested value type. Each type gets the same treatment
+ // as the primitive first_value / last_value benchmarks: update and
+ // merge (both first and last) plus evaluate, at 0% and 90% nulls. On a
+ // build without native nested support these run the fallback adapter;
+ // with this PR they run the native GroupsAccumulator, so the benchmark
+ // bot's before/after diff shows the win per type.
+ let columns: [(&str, ArrayRef); 4] = [
+ ("struct(i64,utf8,f64)", create_struct_array(N, null_density)),
+ ("list<i64>[4]", create_list_array(N, 4, null_density)),
+ ("map<utf8,i64>", create_map_array(N, 4, null_density)),
+ (
+ "list<struct(i64,utf8)>[4]",
+ create_list_of_struct_array(N, 4, null_density),
+ ),
+ ];
+
+ for (type_label, values) in columns {
+ for (fn_label, is_first) in [("first_value", true), ("last_value", false)] {
+ update_bench(
+ c,
+ is_first,
+ &format!("{fn_label} update_bench {type_label} nulls={pct}%"),
+ values.clone(),
+ ord.clone(),
+ None,
+ NUM_GROUPS,
+ );
+ merge_bench(
+ c,
+ is_first,
+ &format!("{fn_label} merge_bench {type_label} nulls={pct}%"),
+ values.clone(),
+ ord.clone(),
+ None,
+ NUM_GROUPS,
+ );
+ }
+ evaluate_bench(
+ c,
+ true,
+ EmitTo::All,
+ &format!("first_value evaluate_bench {type_label} nulls={pct}%, all"),
+ values.clone(),
+ ord.clone(),
+ None,
+ NUM_GROUPS,
+ );
+ }
+ }
+}
+
fn first_last_benchmark(c: &mut Criterion) {
const N: usize = 65536;
const NUM_GROUPS: usize = 1024;
@@ -354,5 +596,5 @@ fn first_last_benchmark(c: &mut Criterion) {
}
}
-criterion_group!(benches, first_last_benchmark);
+criterion_group!(benches, first_last_benchmark, first_last_nested_benchmark);
criterion_main!(benches);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org