feat: support listagg / string_agg aggregate (Spark 4.0+)

docs/source/contributor-guide/expression-audits/agg_funcs.md

@@ -72,6 +72,14 @@
 - Performance (tuned 2026-09-09, PR [#5803](https://github.com/apache/datafusion-comet/pull/5803)): as for [collect_list](#collect-list), grouped `collect_set` no longer goes through `GroupsAccumulatorAdapter`. `CollectSetGroupsAccumulator` encodes each batch once for all of its groups and deduplicates against an open-addressed index over a shared arena, replacing one hash table plus one owned `Row` per distinct value per group. 60-93% faster ([#5797](https://github.com/apache/datafusion-comet/issues/5797)). Benchmark: `benches/collect.rs`.
 - `CometCollectSet` reports `Incompatible` for float and double input when `spark.comet.exec.strictFloatingPoint=true`, because the native distinct comparison treats `NaN == NaN` and collapses repeated `NaN`s into a single element while Spark keeps each one. The native path for floating-point input is then opt-in via `spark.comet.expression.CollectSet.allowIncompatible=true`. All other input types are `Compatible`.
 
+## listagg
+
+- Spark 3.4.3 (audited 2026-07-03): does not exist. `ListAgg` was added in Spark 4.0.
+- Spark 3.5.8 (audited 2026-07-03): does not exist.
+- Spark 4.0.1 (audited 2026-07-03): `ListAgg(child, delimiter, orderExpressions)` in `aggregate/collect.scala`. Accepts `StringType` or `BinaryType` inputs; result type matches child. Skips nulls; empty or all-null groups return `NULL`. A `NULL` delimiter is treated as an empty string. `CometListAgg` maps only the simple form: `StringType` child with a literal `StringType`/`NullType` delimiter and no `WITHIN GROUP`. `BinaryType` inputs, `WITHIN GROUP (ORDER BY ...)`, non-literal delimiters, and non-default collations fall back to Spark. `DISTINCT` falls back because Comet rejects multi-column distinct aggregates (`ListAgg` has two children).
+- Spark 4.1.1 (audited 2026-07-03): byte-identical to 4.0.1.
+- Native accumulator (`SparkListAgg`) returns `Utf8` and carries its intermediate state as `Binary`. Spark's `ListAgg` is a `TypedImperativeAggregate` with a single `BinaryType` buffer attribute, and the shuffle Exchange between the partial and final aggregate uses that schema, so the native state must be `Binary` even though partial and final always run in Comet (`supportsNativePartialToSparkFinal` and `supportsSparkPartialToNativeFinal` are false). A `GroupsAccumulator` fast path is provided for grouped aggregation.
+
 ## max_by
 
 - Spark 3.4.3 (2026-07-03): `MaxBy` is a 2-argument `DeclarativeAggregate` registered as `expression[MaxBy]("max_by")`. Buffer is `(valueWithExtremumOrdering, extremumOrdering)`; null orderings are ignored, the value paired with the maximum ordering is returned (and may itself be null), and an all-null-ordering group yields null. Comet implements a native `max_by` aggregate. Only fixed-length value and ordering types are accelerated: a variable-length or nested type (string, binary, struct) falls back to Spark. On its own such a type never reaches Comet, because Spark plans it as `SortAggregate`, which Comet does not convert. The serde check still matters when a `TypedImperativeAggregate` in the same aggregate switches Spark to `ObjectHashAggregate`, since Arrow's row format would compare a string ordering as raw UTF-8 bytes where Spark compares collation sort keys. `max_by` is non-deterministic when several rows tie on the maximum ordering, matching Spark's documented behavior.

docs/source/user-guide/latest/expressions.md

@@ -107,7 +107,7 @@ The tables below list every Spark built-in expression with its current status.
 | `kurtosis` | 🔜 | — | Not yet implemented natively |
 | `last` | ✅ | Native |  |
 | `last_value` | ✅ | Native |  |
-| `listagg` | 🔜 | — | String aggregation |
+| `listagg` | ✅ | Native | Spark 4.0+. `StringType` input with a literal delimiter; `WITHIN GROUP (ORDER BY ...)` and `BinaryType` inputs fall back to Spark. Without `WITHIN GROUP`, concatenation order is the (non-deterministic) group arrival order, matching Spark. |
 | `max` | ✅ | Native |  |
 | `max_by` | ✅ | Native | Value and ordering must be fixed-length types |
 | `mean` | ✅ | Native |  |
@@ -133,7 +133,7 @@ The tables below list every Spark built-in expression with its current status.
 | `stddev` | ✅ | Native |  |
 | `stddev_pop` | ✅ | Native |  |
 | `stddev_samp` | ✅ | Native |  |
-| `string_agg` | 🔜 | — | String aggregation (alias of `listagg`) |
+| `string_agg` | ✅ | Native | Alias of `listagg`; same restrictions apply. |
 | `sum` | ✅ | Native |  |
 | `try_avg` | ✅ | — | Interval types fall back |
 | `try_sum` | ✅ | — |  |

native/core/src/execution/planner.rs

@@ -93,7 +93,8 @@ use datafusion::{
 use datafusion_comet_spark_expr::{
     create_comet_physical_fun, create_comet_physical_fun_with_eval_mode, BinaryOutputStyle,
     BloomFilterAgg, BloomFilterMightContain, CometCollectList, CometCollectSet, CsvWriteOptions,
-    EvalMode, SparkArraysZipFunc, SparkBloomFilterVersion, SparkPercentile, SumInteger, ToCsv,
+    EvalMode, SparkArraysZipFunc, SparkBloomFilterVersion, SparkListAgg, SparkPercentile,
+    SumInteger, ToCsv,
 };
 use iceberg::expr::Bind;
 
@@ -3181,6 +3182,13 @@ impl PhysicalPlanner {
                 let func = AggregateUDF::new_from_impl(HllPlusPlus::new(expr.precision));
                 Self::create_aggr_func_expr("approx_count_distinct", schema, vec![child], func)
             }
+            AggExprStruct::ListAgg(expr) => {
+                let child = self.create_expr(expr.child.as_ref().unwrap(), Arc::clone(&schema))?;
+                let delimiter =
+                    self.create_expr(expr.delimiter.as_ref().unwrap(), Arc::clone(&schema))?;
+                let func = AggregateUDF::new_from_impl(SparkListAgg::new());
+                Self::create_aggr_func_expr("listagg", schema, vec![child, delimiter], func)
+            }
             AggExprStruct::MaxBy(expr) => {
                 let value = self.create_expr(expr.value.as_ref().unwrap(), Arc::clone(&schema))?;
                 let ordering =

native/proto/src/proto/expr.proto

@@ -166,6 +166,7 @@ message AggExpr {
     Mode mode = 23;
     MaxBy maxBy = 24;
     MinBy minBy = 25;
+    ListAgg listAgg = 26;
   }
 
   // Optional filter expression for SQL FILTER (WHERE ...) clause.
@@ -355,6 +356,20 @@ message HllPlusPlus {
   int32 precision = 2;
 }
 
+// Spark 4.0+ LISTAGG / STRING_AGG aggregate.
+//
+// Comet only serializes the simple form: a StringType child with a literal
+// (or NULL) delimiter and no WITHIN GROUP ORDER BY. DISTINCT falls back to
+// Spark because Comet rejects multi-column distinct aggregates in
+// aggExprToProto, so the native side never sees it.
+message ListAgg {
+  Expr child = 1;
+  // Literal delimiter expression. NULL delimiter is normalized to empty string
+  // by Spark's semantics. The result type is always Utf8 (Spark StringType), so
+  // no datatype field is needed.
+  Expr delimiter = 2;
+}
+
 message MaxBy {
   // The value returned by the aggregate (associated with the maximum ordering).
   Expr value = 1;