perf: Optimize array_agg() using GroupsAccumulator (#20504)

## Which issue does this PR close?

- Closes #20465.
- Closes #17446.

## Rationale for this change

This PR optimizes the performance of `array_agg()` by adding support for
the `GroupsAccumulator` API.

The design tries to minimize the amount of per-batch work done in
`update_batch()`: we store a reference to the batch, and a `(group_idx,
row_idx)` pair for each row. In `evaluate()`, we assemble all the
requested output with a single `interleave` call.

This turns out to be significantly faster, because we copy much less
data and assembling the results can be vectorized more effectively. For
example, on a benchmark with 5000 groups and 5000 int64 values per
group, this approach is roughly 190x faster than the previous approach.

Releasing memory after a partial emit is a little more involved than the
previous approach, but with some determination it is still possible.

## What changes are included in this PR?

* Implement the `GroupsAccumulator` API for `array_agg()`
* Add benchmark for `array_agg` of a named struct over a dict, following
the workload in #17446
* Add unit tests
* Improve SLT test coverage
* Remove a redundant SLT test

## Are these changes tested?

Yes, and benchmarked.

## Are there any user-facing changes?

No.

## AI usage

Iterated with the help of multiple AI tools; I've reviewed and
understand the resulting code.

Author: neilconway

datafusion/core/benches/aggregate_query_sql.rs

@@ -284,6 +284,17 @@ fn criterion_benchmark(c: &mut Criterion) {
             )
         })
     });
+
+    c.bench_function("array_agg_struct_query_group_by_mid_groups", |b| {
+        b.iter(|| {
+            query(
+                ctx.clone(),
+                &rt,
+                "SELECT u64_mid, array_agg(named_struct('market', dict10, 'price', f64)) \
+                 FROM t GROUP BY u64_mid",
+            )
+        })
+    });
 }
 
 criterion_group!(benches, criterion_benchmark);

datafusion/core/benches/data_utils/mod.rs

@@ -20,8 +20,9 @@
 use arrow::array::{
     ArrayRef, Float32Array, Float64Array, RecordBatch, StringArray, StringViewBuilder,
     UInt64Array,
-    builder::{Int64Builder, StringBuilder},
+    builder::{Int64Builder, StringBuilder, StringDictionaryBuilder},
 };
+use arrow::datatypes::Int32Type;
 use arrow::datatypes::{DataType, Field, Schema, SchemaRef};
 use datafusion::datasource::MemTable;
 use datafusion::error::Result;
@@ -65,6 +66,11 @@ pub fn create_schema() -> Schema {
         // Integers randomly selected from a narrow range of values such that
         // there are a few distinct values, but they are repeated often.
         Field::new("u64_narrow", DataType::UInt64, false),
+        Field::new(
+            "dict10",
+            DataType::Dictionary(Box::new(DataType::Int32), Box::new(DataType::Utf8)),
+            true,
+        ),
     ])
 }
 
@@ -109,6 +115,15 @@ fn create_record_batch(
         .map(|_| rng.random_range(0..10))
         .collect::<Vec<_>>();
 
+    let mut dict_builder = StringDictionaryBuilder::<Int32Type>::new();
+    for _ in 0..batch_size {
+        if rng.random::<f64>() > 0.9 {
+            dict_builder.append_null();
+        } else {
+            dict_builder.append_value(format!("market_{}", rng.random_range(0..10)));
+        }
+    }
+
     RecordBatch::try_new(
         schema,
         vec![
@@ -118,6 +133,7 @@ fn create_record_batch(
             Arc::new(UInt64Array::from(integer_values_wide)),
             Arc::new(UInt64Array::from(integer_values_mid)),
             Arc::new(UInt64Array::from(integer_values_narrow)),
+            Arc::new(dict_builder.finish()),
         ],
     )
     .unwrap()

datafusion/functions-aggregate-common/src/aggregate/groups_accumulator/nulls.rs

@@ -44,7 +44,7 @@ pub fn set_nulls<T: ArrowNumericType + Send>(
 /// The `NullBuffer` is
 /// * `true` (representing valid) for values that were `true` in filter
 /// * `false` (representing null) for values that were `false` or `null` in filter
-fn filter_to_nulls(filter: &BooleanArray) -> Option<NullBuffer> {
+pub fn filter_to_nulls(filter: &BooleanArray) -> Option<NullBuffer> {
     let (filter_bools, filter_nulls) = filter.clone().into_parts();
     let filter_bools = NullBuffer::from(filter_bools);
     NullBuffer::union(Some(&filter_bools), filter_nulls.as_ref())