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.

Author: zhuqi-lucas

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);