feat: add native distinct-combined collect_list aggregate support

[peterxcli]

Which issue does this PR close?

Closes #4724.
Closes #4766.

Rationale for this change

collect_list / array_agg and collect_set use Spark TypedImperativeAggregate buffers that Spark declares as serialized BinaryType, while Comet’s native implementations keep the real aggregate state as an Arrow/Spark ArrayType.

The existing schema adjustment only handled simple two-stage Partial -> Final collect aggregates. Spark’s distinct-aggregate rewrite can introduce multi-stage plans with PartialMerge stages, for example:

SELECT x, count(DISTINCT y), collect_list(z) FROM t GROUP BY x

Without correcting the intermediate buffer schema for these stages, a fully-native pipeline can fail when native list state is treated as Spark binary state. This change makes the native array state round-trip through Partial, PartialMerge, and mixed {Partial, PartialMerge} stages so collect_list / collect_set can run fully native in distinct-combined aggregate plans.

What changes are included in this PR?

• Adds native collect_list / array_agg aggregate support:

  • Adds CollectList to the aggregate expression proto.
  • Adds JVM serde for Spark CollectList.
  • Registers CollectList -> CometCollectList.
  • Wires native planning to datafusion_spark::function::aggregate::collect::SparkCollectList.

• Extends collect aggregate native-state schema adjustment:

  • Updates CometObjectHashAggregateExec.adjustOutputForNativeState to handle both CollectList and CollectSet.
  • Applies the rewrite to intermediate Partial, PartialMerge, and mixed {Partial, PartialMerge} stages, not only pure Partial.
  • Rewrites Spark’s serialized BinaryType buffer attributes to Comet’s native ArrayType(elementType, containsNull = true) state type.

• Adds regression coverage for fully-native distinct-combined collect aggregates.

• Updates expression support docs and aggregate audit notes for collect_list / array_agg.

How are these changes tested?

Ran:

cargo fmt --manifest-path native/Cargo.toml --all
cargo check --manifest-path native/Cargo.toml -p datafusion-comet
make core
./mvnw test -Dtest=none -Dsuites="org.apache.comet.exec.CometAggregateSuite collect_list/collect_set combined with distinct aggregate runs fully native" -Dscalastyle.skip=true -Pspark-3.5
./mvnw spotless:check -Dscalastyle.skip=true -Pspark-3.5

explain the sql with:

SELECT x, count(DISTINCT y), collect_list(z)
  FROM t
  GROUP BY x

was:

ObjectHashAggregate(keys=[x#16], functions=[collect_list(z#18, 0, 0), count(distinct y#17)])
+- Exchange hashpartitioning(x#16, 2), ENSURE_REQUIREMENTS, [plan_id=84]
   +- ObjectHashAggregate(keys=[x#16], functions=[merge_collect_list(z#18, 0, 0), partial_count(distinct y#17)])
      +- ObjectHashAggregate(keys=[x#16, y#17], functions=[merge_collect_list(z#18, 0, 0)])
         +- Exchange hashpartitioning(x#16, y#17, 2), ENSURE_REQUIREMENTS, [plan_id=81]
            +- ObjectHashAggregate(keys=[x#16, y#17], functions=[partial_collect_list(z#18, 0, 0)])
               +- CometNativeColumnarToRow
                  +- CometNativeScan parquet [x#16,y#17,z#18] Batched: true, DataFilters: [], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/var/folders/6h/_n7pxl3n7knbwg5lwlrkzr840000gn/T/comet_distinct_c..., PartitionFilters: [], PushedFilters: [], ReadSchema: struct<x:int,y:int,z:string>

after:

== Physical Plan ==
CometNativeColumnarToRow
+- CometHashAggregate [x#16, buf#24, count#27L], [Final], [x#16], [collect_list(z#18, 0, 0), count(distinct y#17)]
   +- CometExchange hashpartitioning(x#16, 2), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=69]
      +- CometHashAggregate [x#16, y#17, buf#24], [PartialMerge, Partial], [x#16], [merge_collect_list(z#18, 0, 0), partial_count(distinct y#17)]
         +- CometHashAggregate [x#16, y#17, buf#24], [PartialMerge], [x#16, y#17], [merge_collect_list(z#18, 0, 0)]
            +- CometExchange hashpartitioning(x#16, y#17, 2), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=66]
               +- CometHashAggregate [x#16, y#17, z#18], [Partial], [x#16, y#17], [partial_collect_list(z#18, 0, 0)]
                  +- CometNativeScan parquet [x#16,y#17,z#18] Batched: true, DataFilters: [], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/var/folders/6h/_n7pxl3n7knbwg5lwlrkzr840000gn/T/comet_distinct_c..., PartitionFilters: [], PushedFilters: [], ReadSchema: struct<x:int,y:int,z:string>

[hsiang-c]

👍 Could we add a few SQL file tests: https://datafusion.apache.org/comet/contributor-guide/sql-file-tests.html#sql-file-tests, thank you!

[peterxcli]

Noticed that I didn't solve the root cause, we need a native UnsafeRow / UnsafeArrayData bytes decoder. addressing...

[peterxcli]

@andygrove would you mind taking a look when you have a chance? This is related to #4720. Thanks!