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

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

@@ -39,6 +39,14 @@
 - Spark 3.5.8 (2026-05-26)
 - Spark 4.0.1 (2026-05-26)
 
+## 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 `Utf8`. Partial and final always run in the same engine (`supportsMixedPartialFinal` is false), so no cross-engine buffer-schema matching is required. A `GroupsAccumulator` fast path is provided for grouped aggregation.
+
 ## median
 
 - Spark 3.4.3 (audited 2026-06-24): `Median(child)` is a `RuntimeReplaceableAggregate` with `replacement = Percentile(child, Literal(0.5))`. Catalyst rewrites `median(x)` to `percentile(x, 0.5)` before Comet sees the plan, so it is served by `CometPercentile`.

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

@@ -93,7 +93,7 @@ The tables below list every Spark built-in expression with its current status.
 | `kurtosis` | 🔜 | tracking [#4098](https://github.com/apache/datafusion-comet/issues/4098) |
 | `last` | ✅ |  |
 | `last_value` | ✅ |  |
-| `listagg` | 🔜 | String aggregation |
+| `listagg` | ✅ | 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` | ✅ |  |
 | `max_by` | 🔜 | [#3841](https://github.com/apache/datafusion-comet/issues/3841) |
 | `mean` | ✅ |  |
@@ -119,7 +119,7 @@ The tables below list every Spark built-in expression with its current status.
 | `stddev` | ✅ |  |
 | `stddev_pop` | ✅ |  |
 | `stddev_samp` | ✅ |  |
-| `string_agg` | 🔜 | String aggregation (alias of `listagg`) |
+| `string_agg` | ✅ | Alias of `listagg`; same restrictions apply. |
 | `sum` | ✅ |  |
 | `try_avg` | ✅ | Interval types fall back |
 | `try_sum` | ✅ |  |

native/core/src/execution/planner.rs

@@ -74,7 +74,7 @@ use datafusion::{
 use datafusion_comet_spark_expr::{
     create_comet_physical_fun, create_comet_physical_fun_with_eval_mode, BinaryOutputStyle,
     BloomFilterAgg, BloomFilterMightContain, CsvWriteOptions, EvalMode, SparkArraysZipFunc,
-    SparkBloomFilterVersion, SumInteger, ToCsv,
+    SparkBloomFilterVersion, SparkListAgg, SumInteger, ToCsv,
 };
 use datafusion_spark::function::aggregate::collect::SparkCollectSet;
 use iceberg::expr::Bind;
@@ -2653,6 +2653,13 @@ impl PhysicalPlanner {
                 let func = AggregateUDF::new_from_impl(SparkCollectSet::new());
                 Self::create_aggr_func_expr("collect_set", 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)
+            }
         }
     }
 

native/proto/src/proto/expr.proto

@@ -146,6 +146,7 @@ message AggExpr {
     BloomFilterAgg bloomFilterAgg = 16;
     CollectSet collectSet = 17;
     Percentile percentile = 18;
+    ListAgg listAgg = 19;
   }
 
   // Optional filter expression for SQL FILTER (WHERE ...) clause.
@@ -277,6 +278,19 @@ message CollectSet {
   DataType datatype = 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 is handled by
+// Spark's multi-stage plan rewrite before the aggregate reaches Comet.
+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;
+}
+
 enum EvalMode {
   LEGACY = 0;
   TRY = 1;