feat: native collect_list / array_agg aggregate

dev/diffs/3.4.3.diff

@@ -260,7 +260,7 @@ index cf40e944c09..bdd5be4f462 100644
 
    test("A cached table preserves the partitioning and ordering of its cached SparkPlan") {
 diff --git a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala
-index 1cc09c3d7fc..a84939c045b 100644
+index 1cc09c3d7fc..9e1e883d450 100644
 --- a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala
 +++ b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala
 @@ -26,8 +26,9 @@ import org.scalatest.matchers.must.Matchers.the
@@ -285,7 +285,13 @@ index 1cc09c3d7fc..a84939c045b 100644
          }
 
          // test case for ObjectHashAggregate and SortAggregate
-@@ -755,7 +758,7 @@ class DataFrameAggregateSuite extends QueryTest
+@@ -750,12 +753,12 @@ class DataFrameAggregateSuite extends QueryTest
+       assert(sortAggPlans.isEmpty)
+
+       val objHashAggPlans = collect(aggPlan) {
+-        case objHashAgg: ObjectHashAggregateExec => objHashAgg
++        case objHashAgg @ (_: ObjectHashAggregateExec | _: CometHashAggregateExec) => objHashAgg
+       }
        assert(objHashAggPlans.nonEmpty)
 
        val exchangePlans = collect(aggPlan) {

dev/diffs/3.5.8.diff

@@ -241,7 +241,7 @@ index e5494726695..00937f025c2 100644
 
    test("A cached table preserves the partitioning and ordering of its cached SparkPlan") {
 diff --git a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala
-index 6f3090d8908..a0e9309888d 100644
+index 6f3090d8908..4774aad5019 100644
 --- a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala
 +++ b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala
 @@ -27,8 +27,9 @@ import org.apache.spark.{SparkException, SparkThrowable}
@@ -266,7 +266,13 @@ index 6f3090d8908..a0e9309888d 100644
          }
 
          // test case for ObjectHashAggregate and SortAggregate
-@@ -793,7 +796,7 @@ class DataFrameAggregateSuite extends QueryTest
+@@ -788,12 +791,12 @@ class DataFrameAggregateSuite extends QueryTest
+       assert(sortAggPlans.isEmpty)
+
+       val objHashAggPlans = collect(aggPlan) {
+-        case objHashAgg: ObjectHashAggregateExec => objHashAgg
++        case objHashAgg @ (_: ObjectHashAggregateExec | _: CometHashAggregateExec) => objHashAgg
+       }
        assert(objHashAggPlans.nonEmpty)
 
        val exchangePlans = collect(aggPlan) {

dev/diffs/4.0.2.diff

@@ -378,7 +378,7 @@ index 0f42502f1d9..e9ff802141f 100644
 
        withTempView("t0", "t1", "t2") {
 diff --git a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala
-index 9db406ff12f..19b4bb8e39d 100644
+index 9db406ff12f..b3d55394d25 100644
 --- a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala
 +++ b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala
 @@ -29,8 +29,9 @@ import org.apache.spark.sql.catalyst.util.AUTO_GENERATED_ALIAS
@@ -403,7 +403,13 @@ index 9db406ff12f..19b4bb8e39d 100644
          }
 
          // test case for ObjectHashAggregate and SortAggregate
-@@ -855,7 +858,7 @@ class DataFrameAggregateSuite extends QueryTest
+@@ -850,12 +853,12 @@ class DataFrameAggregateSuite extends QueryTest
+       assert(sortAggPlans.isEmpty)
+
+       val objHashAggPlans = collect(aggPlan) {
+-        case objHashAgg: ObjectHashAggregateExec => objHashAgg
++        case objHashAgg @ (_: ObjectHashAggregateExec | _: CometHashAggregateExec) => objHashAgg
+       }
        assert(objHashAggPlans.nonEmpty)
 
        val exchangePlans = collect(aggPlan) {

dev/diffs/4.1.2.diff

@@ -392,7 +392,7 @@ index 0d807aeae4d..6d7744e771b 100644
 
        withTempView("t0", "t1", "t2") {
 diff --git a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala
-index bfe15b33768..31f8ba4b876 100644
+index bfe15b33768..13aeb3f6610 100644
 --- a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala
 +++ b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala
 @@ -30,8 +30,9 @@ import org.apache.spark.sql.catalyst.util.AUTO_GENERATED_ALIAS
@@ -417,7 +417,13 @@ index bfe15b33768..31f8ba4b876 100644
          }
 
          // test case for ObjectHashAggregate and SortAggregate
-@@ -856,7 +859,7 @@ class DataFrameAggregateSuite extends QueryTest
+@@ -851,12 +854,12 @@ class DataFrameAggregateSuite extends QueryTest
+       assert(sortAggPlans.isEmpty)
+
+       val objHashAggPlans = collect(aggPlan) {
+-        case objHashAgg: ObjectHashAggregateExec => objHashAgg
++        case objHashAgg @ (_: ObjectHashAggregateExec | _: CometHashAggregateExec) => objHashAgg
+       }
        assert(objHashAggPlans.nonEmpty)
 
        val exchangePlans = collect(aggPlan) {

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

@@ -44,6 +44,15 @@
 - Spark 3.5.8 (2026-05-26)
 - Spark 4.0.1 (2026-05-26)
 
+## collect_list
+
+- Spark 3.4.3 (audited 2026-06-24): `CollectList` extends `Collect[ArrayBuffer[Any]]`, returns `ArrayType(child.dataType, containsNull = false)`, ignores NULL inputs in `update()` (Hive-compatible semantics), and yields an empty array as `defaultResult`. `nullable = false`. No `checkInputDataTypes` override, so any input type is accepted (including STRUCT, ARRAY, MAP). Registered as both `collect_list` and `array_agg` aliases in `FunctionRegistry`.
+- Spark 3.5.8 (audited 2026-06-24): identical to 3.4.3.
+- Spark 4.0.1 (audited 2026-06-24): only structural change is adding `with UnaryLike[Expression]` to the case class (no behavior change).
+- Spark 4.1.1 (audited 2026-06-24): identical to 4.0.1.
+- Comet implementation: native side delegates to `datafusion_spark::function::aggregate::collect::SparkCollectList`, which wraps `ArrayAggAccumulator` with `ignore_nulls = true` and converts a final NULL accumulator state to an empty array (matching Spark's `defaultResult`). The native return type is `List(Field, containsNull = true)`, while Spark uses `containsNull = false`. Because nulls are filtered before insertion, no nulls actually appear in the array, so this is a schema-shape difference only and tests using `checkSparkAnswerAndOperator` accept it (same pattern already in use for `collect_set`).
+- Spark 4.2 (preview): `CollectList` and `CollectSet` gain an `ignoreNulls` field (default `true`); `RESPECT NULLS` sets it to `false` and keeps null elements. The native path always drops nulls, so `CometCollectShim` reads the field per Spark version (always `true` on 3.4-4.1) and `CometCollectList` / `CometCollectSet` report `Unsupported` when it is `false`, falling back to Spark.
+
 ## 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

@@ -72,14 +72,14 @@ The tables below list every Spark built-in expression with its current status.
 | `any_value` | ✅ |  |
 | `approx_count_distinct` | 🔜 | tracking [#4098](https://github.com/apache/datafusion-comet/issues/4098) |
 | `approx_percentile` | ✅ | Byte, short, int, long, float, and double input; other input types fall back to Spark |
-| `array_agg` | 🔜 | Array aggregate (related to `collect_list`, [#2524](https://github.com/apache/datafusion-comet/issues/2524)) |
+| `array_agg` | ✅ | Alias for `collect_list` |
 | `avg` | ✅ | Interval types fall back |
 | `bit_and` | ✅ |  |
 | `bit_or` | ✅ |  |
 | `bit_xor` | ✅ |  |
 | `bool_and` | ✅ |  |
 | `bool_or` | ✅ |  |
-| `collect_list` | 🔜 | [#2524](https://github.com/apache/datafusion-comet/issues/2524) |
+| `collect_list` | ✅ |  |
 | `collect_set` | ✅ |  |
 | `corr` | ✅ |  |
 | `count` | ✅ |  |

native/core/src/execution/planner.rs

@@ -78,7 +78,7 @@ use datafusion_comet_spark_expr::{
     BloomFilterAgg, BloomFilterMightContain, CsvWriteOptions, EvalMode, SparkArraysZipFunc,
     SparkBloomFilterVersion, SparkPercentile, SumInteger, ToCsv,
 };
-use datafusion_spark::function::aggregate::collect::SparkCollectSet;
+use datafusion_spark::function::aggregate::collect::{SparkCollectList, SparkCollectSet};
 use iceberg::expr::Bind;
 
 use crate::execution::operators::ExecutionError::GeneralError;
@@ -158,6 +158,53 @@ struct JoinParameters {
     pub join_type: DFJoinType,
 }
 
+/// Return a copy of `data_type` with every nested field marked nullable. Map key fields are left
+/// non-nullable to preserve Arrow's map invariant. Primitive types are returned unchanged.
+fn make_all_fields_nullable(data_type: &DataType) -> DataType {
+    fn nullable_field(field: &Field, nullable: bool) -> FieldRef {
+        Arc::new(
+            Field::new(
+                field.name(),
+                make_all_fields_nullable(field.data_type()),
+                nullable,
+            )
+            .with_metadata(field.metadata().clone()),
+        )
+    }
+    match data_type {
+        DataType::Struct(fields) => {
+            DataType::Struct(fields.iter().map(|f| nullable_field(f, true)).collect())
+        }
+        DataType::List(field) => DataType::List(nullable_field(field, true)),
+        DataType::LargeList(field) => DataType::LargeList(nullable_field(field, true)),
+        DataType::FixedSizeList(field, len) => {
+            DataType::FixedSizeList(nullable_field(field, true), *len)
+        }
+        DataType::Map(entries, sorted) => {
+            // Map entries are a non-nullable struct of {key (non-null), value}. Recurse into the
+            // key/value types but keep the key field non-nullable per Arrow's map invariant.
+            match entries.data_type() {
+                DataType::Struct(kv) => {
+                    let new_kv = kv
+                        .iter()
+                        .enumerate()
+                        .map(|(i, f)| nullable_field(f, i != 0))
+                        .collect();
+                    let new_entries = Field::new(
+                        entries.name(),
+                        DataType::Struct(new_kv),
+                        entries.is_nullable(),
+                    )
+                    .with_metadata(entries.metadata().clone());
+                    DataType::Map(Arc::new(new_entries), *sorted)
+                }
+                _ => data_type.clone(),
+            }
+        }
+        other => other.clone(),
+    }
+}
+
 /// If `expr` evaluates to `Timestamp(_, Some(_))` against `schema`, wrap it in a
 /// metadata-only cast to `Timestamp(_, None)`. This is required because
 /// DataFusion's `SortMergeJoinExec` comparator only supports timezone-less
@@ -2666,9 +2713,16 @@ impl PhysicalPlanner {
             }
             AggExprStruct::CollectSet(expr) => {
                 let child = self.create_expr(expr.child.as_ref().unwrap(), Arc::clone(&schema))?;
+                let child = Self::coerce_collect_child_nullability(child, &schema)?;
                 let func = AggregateUDF::new_from_impl(SparkCollectSet::new());
                 Self::create_aggr_func_expr("collect_set", schema, vec![child], func)
             }
+            AggExprStruct::CollectList(expr) => {
+                let child = self.create_expr(expr.child.as_ref().unwrap(), Arc::clone(&schema))?;
+                let child = Self::coerce_collect_child_nullability(child, &schema)?;
+                let func = AggregateUDF::new_from_impl(SparkCollectList::new());
+                Self::create_aggr_func_expr("collect_list", schema, vec![child], func)
+            }
         }
     }
 
@@ -3273,6 +3327,27 @@ impl PhysicalPlanner {
         Ok(scalar_expr)
     }
 
+    /// `collect_list` / `collect_set` build their result list with all element fields marked
+    /// nullable, regardless of the input's nullability. However `SparkCollectList::return_type`
+    /// (and `SparkCollectSet`) derive the element type directly from the child, preserving any
+    /// non-nullable nested field. When the child is a nested type with a non-nullable inner field
+    /// (e.g. a struct field), the declared aggregate output disagrees with the array the
+    /// accumulator actually produces, and DataFusion's grouped `AggregateExec` fails validating
+    /// its output batch ("column types must match schema types"). Cast the child to the
+    /// all-nullable variant of its type so the declared and produced types stay consistent.
+    fn coerce_collect_child_nullability(
+        child: Arc<dyn PhysicalExpr>,
+        schema: &SchemaRef,
+    ) -> Result<Arc<dyn PhysicalExpr>, ExecutionError> {
+        let child_type = child.data_type(schema.as_ref())?;
+        let nullable_type = make_all_fields_nullable(&child_type);
+        if child_type.equals_datatype(&nullable_type) {
+            Ok(child)
+        } else {
+            Ok(Arc::new(CastExpr::new(child, nullable_type, None)))
+        }
+    }
+
     fn create_aggr_func_expr(
         name: &str,
         schema: SchemaRef,

native/proto/src/proto/expr.proto

@@ -149,6 +149,7 @@ message AggExpr {
     CollectSet collectSet = 17;
     Percentile percentile = 18;
     ApproxPercentile approxPercentile = 19;
+    CollectList collectList = 20;
   }
 
   // Optional filter expression for SQL FILTER (WHERE ...) clause.
@@ -296,6 +297,11 @@ message CollectSet {
   DataType datatype = 2;
 }
 
+message CollectList {
+  Expr child = 1;
+  DataType datatype = 2;
+}
+
 enum EvalMode {
   LEGACY = 0;
   TRY = 1;

spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala

@@ -876,6 +876,29 @@ case class CometExecRule(session: SparkSession)
             }
           }
         }
+
+        // CollectList/CollectSet round-trip an ArrayType buffer that Spark declares as BinaryType.
+        // In a multi-stage aggregate with a PartialMerge stage (e.g. Spark's distinct-aggregate
+        // rewrite), Comet cannot represent that buffer consistently across the intermediate stages
+        // (issue #4724), so a fully-native pipeline crashes. Force the whole chain to fall back to
+        // Spark by tagging the feeding pure-Partial; the PartialMerge/Final stages then fall back
+        // via the buffer-source check in doConvert.
+        //
+        // This block is intentionally separate from the tagging block just above: that one only
+        // fires when the Final itself cannot be converted, but `canAggregateBeConverted` skips
+        // the child-native check, so an all-native distinct `collect_list` chain converts its
+        // Final and slips past the earlier tagging pass. This block catches that case.
+        if (agg.aggregateExpressions.exists(_.mode == PartialMerge) &&
+          QueryPlanSerde.hasNativeArrayBufferAgg(agg.aggregateExpressions)) {
+          findPartialAggInPlan(agg.child).foreach { partial =>
+            if (canAggregateBeConverted(partial, Partial)) {
+              partial.setTagValue(
+                CometExecRule.COMET_UNSAFE_PARTIAL,
+                "Partial aggregate disabled: part of a multi-stage CollectList/CollectSet " +
+                  "aggregate whose intermediate buffer cannot round-trip in Comet (issue #4724)")
+            }
+          }
+        }
       case _ =>
     }
   }

spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala

@@ -393,6 +393,7 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim {
     classOf[BitOrAgg] -> CometBitOrAgg,
     classOf[BitXorAgg] -> CometBitXOrAgg,
     classOf[BloomFilterAggregate] -> CometBloomFilterAggregate,
+    classOf[CollectList] -> CometCollectList,
     classOf[CollectSet] -> CometCollectSet,
     classOf[Corr] -> CometCorr,
     classOf[Count] -> CometCount,
@@ -439,6 +440,25 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim {
     }
   }
 
+  /**
+   * Returns true if any aggregate is CollectList/CollectSet. These produce a native ArrayType
+   * intermediate buffer while Spark declares BinaryType for its serialized
+   * TypedImperativeAggregate buffer, so Comet cannot interpret Spark's Binary buffer, and Comet
+   * cannot yet represent this buffer consistently across the intermediate PartialMerge stages of
+   * a multi-stage aggregate (issue #4724). These aggregates are therefore only safe to run
+   * natively when every stage runs in Comet and there are at most two stages (Partial + Final).
+   *
+   * Percentile has a similar Array-shaped intermediate buffer (see `adjustOutputForNativeState`)
+   * but is not matched here: it already passes through the general mixed-execution guard, so this
+   * check is scoped narrowly to the collect functions.
+   */
+  def hasNativeArrayBufferAgg(aggExprs: Seq[AggregateExpression]): Boolean = {
+    aggExprs.exists(_.aggregateFunction match {
+      case _: CollectList | _: CollectSet => true
+      case _ => false
+    })
+  }
+
   //  A unique id for each expression. ~used to look up QueryContext during error creation.
   private val exprIdCounter = new AtomicLong(0)