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

[andygrove]

Which issue does this PR close?

Closes #.

Rationale for this change

Spark 4.0 added the LISTAGG / STRING_AGG aggregate to concatenate string values in a group. Comet previously fell back to Spark for the entire aggregate. This PR adds native support for the common shape so groups that use listagg on a StringType column with a literal delimiter can stay in Comet's native execution path.

What changes are included in this PR?

• Adds a Comet-owned SparkListAgg UDAF (native/spark-expr/src/agg_funcs/list_agg.rs) that skips nulls, returns Utf8, and — critically — keeps its intermediate state as Binary to match Spark's TypedImperativeAggregate buffer schema. Emitting Utf8 state would force the Comet shuffle layer to insert a Utf8 → Binary cast that the merge side then cannot read back.
• Adds the CometListAgg serde in spark/src/main/spark-4.x/ and wires it into QueryPlanSerde.aggrSerdeMap through a new sparkVersionSpecificAggregates shim slot (empty for Spark 3.x, provides ListAgg on Spark 4.x). This keeps the base map compiling against Spark 3.x where ListAgg does not exist.
• Adds a new ListAgg protobuf message and wires it through the native planner.
• Restricts native execution to the simple form. WITHIN GROUP (ORDER BY ...), BinaryType inputs, non-literal delimiters, and non-default collations fall back to Spark. DISTINCT falls back naturally because Comet's aggregate dispatcher rejects multi-column distinct aggregates.
• Documents the audit findings in docs/source/contributor-guide/expression-audits/agg_funcs.md and updates the support status in docs/source/user-guide/latest/expressions.md.

Scaffolding produced by the implement-comet-expression skill; the audit-comet-expression skill drove the fallback and coverage-gap tests.

How are these changes tested?

• New Rust unit tests for ListAggAccumulator covering delimiter handling, null-input skipping, empty groups, and multi-partition state merge (native/spark-expr/src/agg_funcs/list_agg.rs).
• New Comet SQL test at spark/src/test/resources/sql-tests/expressions/aggregate/listagg.sql gated to Spark 4.0+ (MinSparkVersion: 4.0), running under ConfigMatrix: parquet.enable.dictionary=false,true to exercise both dictionary-encoded and plain-encoded reads. It covers:
  • Native cases: literal delimiter, string_agg alias, default (empty) delimiter, all-NULL group, empty table, global aggregate, multi-char and empty-string delimiters, empty-string values inside a group, multibyte UTF-8 values (composed accents, CJK, emoji).
  • expect_fallback cases: DISTINCT, WITHIN GROUP (ORDER BY ...) ascending and descending, BinaryType child with default and literal binary delimiter.
• ./mvnw test -Pspark-4.0 -Dsuites="org.apache.comet.CometSqlFileTestSuite listagg" -Dtest=none passes for both dictionary-encoding modes.
• cd native && cargo clippy --all-targets --workspace -- -D warnings passes.

[mbutrovich]

First pass, thanks @andygrove! I verified against .../aggregate/collect.scala and DataFusion's string_agg. No Comet-vs-Spark correctness gap: null-skip, empty-group-returns-NULL, NULL-delimiter-returns-empty-string, empty-string-value handling, and the DISTINCT / WITHIN GROUP / BinaryType / collation fallbacks all match Spark (citations below). The main finding is reuse: this reimplements DataFusion's string_agg and omits its GroupsAccumulator, so grouped queries take the slow path. There is also a dead proto field and a design-rationale comment that rests on a false premise.

[mbutrovich]

ListAggAccumulator is a near-verbatim copy of DataFusion's SimpleStringAggAccumulator (datafusion/functions-aggregate/src/string_agg.rs:451): same delimiter/accumulated/has_value fields and the same update_batch/evaluate/state/size and delimiter-extraction logic (string_agg.rs:122). DataFusion's string_agg already implements exactly this simple form with identical null-skip and NULL-delimiter semantics, and it ships a StringAggGroupsAccumulator fast path (string_agg.rs:321) that this PR does not provide. The cited differences (Utf8 vs LargeUtf8 return, Binary vs LargeUtf8 state) are cosmetic. Preferred fix: wire DataFusion's string_agg with a result cast to Utf8 via scalarFunctionExprToProtoWithReturnType, which deletes this file and gains the grouped fast path. At minimum, add a GroupsAccumulator, since every SQL test here is GROUP BY and currently takes the slow per-group Accumulator path.

[andygrove]

Added a GroupsAccumulator (ListAggGroupsAccumulator) so grouped queries no longer take the per-group Accumulator path.

I kept the Comet-owned UDAF rather than reusing string_agg directly. string_agg returns LargeUtf8 and its SimpleStringAggAccumulator / StringAggGroupsAccumulator are pub(crate), so reuse would mean wiring the whole UDAF plus a LargeUtf8 -> Utf8 cast on the aggregate result. The cited scalarFunctionExprToProtoWithReturnType is a scalar-function path and does not apply to aggregates. Emitting Utf8 at the JVM FFI boundary is the reason the custom UDAF exists.

[mbutrovich]

The module doc and state_fields comment justify the Binary intermediate state by saying a Utf8 state would force a Comet shuffle-side Utf8 -> Binary cast that the merge side cannot read. I could not reconcile that with the rest of the code, so this may be worth revisiting. CometListAgg does not override supportsMixedPartialFinal, so it inherits false (CometAggregateExpressionSerde.scala:93), which as I read it means partial and final always run in the same engine (same as CometCollectSet and CometPercentile), so Spark's Catalyst buffer schema would not be involved. DataFusion's string_agg uses Utf8/LargeUtf8 state and merges without a cast (string_agg.rs:166), and I did not find a Utf8 -> Binary cast in the shuffle layer. If that reading is right, the Binary state adds a per-merge from_utf8 validation and byte round-trip (list_agg.rs:135, :147) for no benefit, and Utf8 state would be simpler. Is there a mixed-execution or shuffle-serialization case I am missing that the Binary state is guarding against? If not, consider Utf8 state, or fold this into the string_agg reuse above.

[andygrove]

The Binary state is actually required. I switched state_fields to Utf8 to try it and it fails at runtime:

could not cast array of type Binary to arrow_array::...::GenericStringType<i32>

CometHashAggregateExec.output for a Partial aggregate is Spark's groupingAttributes ++ aggBufferAttributes, and Spark's ListAgg is a TypedImperativeAggregate whose aggBufferAttributes is a single BinaryType column. The shuffle Exchange between the partial and final aggregate is serialized with that Spark-declared schema, so the native partial state must be Binary.

This holds even though supportsMixedPartialFinal=false: partial and final both run in Comet, but the Exchange schema between them is Spark's plan schema, not DataFusion's. That is why string_agg using Utf8/LargeUtf8 state does not carry over here.

You were right that the old comment was hard to reconcile: the "Utf8 -> Binary cast the merge side cannot read" wording was imprecise. I rewrote the module doc and the state_fields comment to explain the real mechanism.

[mbutrovich]

convert runs only after getSupportLevel returned Compatible, which requires the child to be StringType, and serializeDataType always returns Some for StringType. So the dataType.isEmpty branch is unreachable and the datatype proto field it sets is never read by the planner arm (planner.rs, which hardcodes Utf8). Drop the field and the branch, or replace the branch with an explicit unreachable!-style guard stating the child is guaranteed StringType, so the invariant is self-documenting rather than silent dead code.

[andygrove]

Removed the datatype proto field and the unreachable dataType.isEmpty branch. getSupportLevel guarantees a StringType child, so serializeDataType always returns Some, and the planner hardcodes Utf8.

[mbutrovich]

distinctReason is listed in getUnsupportedReasons() but is never emitted: DISTINCT falls back earlier in aggExprToProto (QueryPlanSerde.scala:640) with the generic multi-column-distinct message, which is what the SQL test at listagg.sql:100 matches. Remove distinctReason or mark it informational, so the user-facing reason list reflects the actual fallback message.

[andygrove]

Removed distinctReason. Added a comment noting DISTINCT falls back earlier in aggExprToProto with the generic multi-column-distinct message, which is what the SQL test matches.

[mbutrovich]

The collation fallback path (CometListAgg.scala:57, :61) is reachable but untested. Add:

statement
CREATE TABLE la_coll(v string COLLATE UTF8_LCASE, grp string) USING parquet

statement
INSERT INTO la_coll VALUES ('a','g1'), ('B','g1')

query expect_fallback(Non-default string collations are not supported)
SELECT grp, listagg(v, ',') FROM la_coll GROUP BY grp ORDER BY grp

Confirm the exact fallback substring against the emitted collationReason. Also add a multi-partition case (repartition before the aggregate) to stress the arrival-order assumption discussed below, since every current case is single-partition.

[andygrove]

Added the multi-partition test (REPARTITION(4) over identical per-group values, so the result stays deterministic while the group is split across partitions and merged in the final aggregate).

The collation fallback turns out not to be testable at the listagg level. Comet's Parquet scan rejects a collated-string schema before the aggregate is considered:

Unsupported schema StructType(StructField(v,StringType(UTF8_LCASE),true),...)

so the whole plan falls back to Spark and CometListAgg's collation guard is never reached. (Also, the delimiter must share the child's collation or Spark's analyzer rejects the call outright with DATATYPE_MISMATCH.) I kept the guard as defense-in-depth (it keeps listagg correct if Comet later gains collated-scan support) and documented this in both the serde and the SQL file rather than adding a test that cannot exercise it.

[mbutrovich]

Without WITHIN GROUP, both Spark (collect.scala, "without order return as is") and this accumulator concatenate in group arrival order, which Spark documents as non-deterministic after shuffle (collect.scala:309). This is the first order-sensitive aggregate Comet accelerates, so there is no precedent. No divergence was found because the SQL tests wrap inputs in (SELECT ... ORDER BY grp, v) and run single-partition, so Spark and Comet observe the same order. That protection is incidental. Add the multi-partition test above and add the non-determinism caveat to the user-guide note, since Spark itself flags it.

[andygrove]

Added the multi-partition merge test and added the non-determinism caveat to the user-guide note ("Without WITHIN GROUP, concatenation order is the (non-deterministic) group arrival order, matching Spark").