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:06:30 UTC
(datafusion) branch gh-readonly-queue/main/pr-24075-3ef9a8c2216bf0761580946748c241bc2dc9b816 created (now 26fbf4d015)
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-24075-3ef9a8c2216bf0761580946748c241bc2dc9b816
in repository https://gitbox.apache.org/repos/asf/datafusion.git
at 26fbf4d015 bench: add nested-type (List/Struct/Map) cases to first_value/last_value benchmark (#24075)
This branch includes the following new commits:
new 26fbf4d015 bench: add nested-type (List/Struct/Map) cases to first_value/last_value benchmark (#24075)
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: bench: add nested-type (List/Struct/Map) cases to first_value/last_value benchmark (#24075)
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-24075-3ef9a8c2216bf0761580946748c241bc2dc9b816
in repository https://gitbox.apache.org/repos/asf/datafusion.git
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