feat: add native distinct-combined collect_list aggregate support

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..9e1e883d450 100644
+index 1cc09c3d7fc..e25dc3d1993 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,17 @@ index 1cc09c3d7fc..9e1e883d450 100644
          }
 
          // test case for ObjectHashAggregate and SortAggregate
-@@ -750,12 +753,12 @@ class DataFrameAggregateSuite extends QueryTest
+@@ -700,7 +703,8 @@ class DataFrameAggregateSuite extends QueryTest
+         val objHashAggOrSortAggPlan =
+           stripAQEPlan(objHashAggOrSortAggDF.queryExecution.executedPlan)
+         if (useObjectHashAgg) {
+-          assert(objHashAggOrSortAggPlan.isInstanceOf[ObjectHashAggregateExec])
++          assert(objHashAggOrSortAggPlan.isInstanceOf[ObjectHashAggregateExec] ||
++            objHashAggOrSortAggPlan.isInstanceOf[CometHashAggregateExec])
+         } else {
+           assert(objHashAggOrSortAggPlan.isInstanceOf[SortAggregateExec])
+         }
+@@ -750,12 +754,12 @@ class DataFrameAggregateSuite extends QueryTest
        assert(sortAggPlans.isEmpty)
 
        val objHashAggPlans = 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..4774aad5019 100644
+index 6f3090d8908..52523429c87 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,17 @@ index 6f3090d8908..4774aad5019 100644
          }
 
          // test case for ObjectHashAggregate and SortAggregate
-@@ -788,12 +791,12 @@ class DataFrameAggregateSuite extends QueryTest
+@@ -738,7 +741,8 @@ class DataFrameAggregateSuite extends QueryTest
+         val objHashAggOrSortAggPlan =
+           stripAQEPlan(objHashAggOrSortAggDF.queryExecution.executedPlan)
+         if (useObjectHashAgg) {
+-          assert(objHashAggOrSortAggPlan.isInstanceOf[ObjectHashAggregateExec])
++          assert(objHashAggOrSortAggPlan.isInstanceOf[ObjectHashAggregateExec] ||
++            objHashAggOrSortAggPlan.isInstanceOf[CometHashAggregateExec])
+         } else {
+           assert(objHashAggOrSortAggPlan.isInstanceOf[SortAggregateExec])
+         }
+@@ -788,12 +792,12 @@ class DataFrameAggregateSuite extends QueryTest
        assert(sortAggPlans.isEmpty)
 
        val objHashAggPlans = 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..b3d55394d25 100644
+index 9db406ff12f..e59a931c06a 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,17 @@ index 9db406ff12f..b3d55394d25 100644
          }
 
          // test case for ObjectHashAggregate and SortAggregate
-@@ -850,12 +853,12 @@ class DataFrameAggregateSuite extends QueryTest
+@@ -800,7 +803,8 @@ class DataFrameAggregateSuite extends QueryTest
+         val objHashAggOrSortAggPlan =
+           stripAQEPlan(objHashAggOrSortAggDF.queryExecution.executedPlan)
+         if (useObjectHashAgg) {
+-          assert(objHashAggOrSortAggPlan.isInstanceOf[ObjectHashAggregateExec])
++          assert(objHashAggOrSortAggPlan.isInstanceOf[ObjectHashAggregateExec] ||
++            objHashAggOrSortAggPlan.isInstanceOf[CometHashAggregateExec])
+         } else {
+           assert(objHashAggOrSortAggPlan.isInstanceOf[SortAggregateExec])
+         }
+@@ -850,12 +854,12 @@ class DataFrameAggregateSuite extends QueryTest
        assert(sortAggPlans.isEmpty)
 
        val objHashAggPlans = 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..13aeb3f6610 100644
+index bfe15b33768..df8fb4c0a35 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,17 @@ index bfe15b33768..13aeb3f6610 100644
          }
 
          // test case for ObjectHashAggregate and SortAggregate
-@@ -851,12 +854,12 @@ class DataFrameAggregateSuite extends QueryTest
+@@ -801,7 +804,8 @@ class DataFrameAggregateSuite extends QueryTest
+         val objHashAggOrSortAggPlan =
+           stripAQEPlan(objHashAggOrSortAggDF.queryExecution.executedPlan)
+         if (useObjectHashAgg) {
+-          assert(objHashAggOrSortAggPlan.isInstanceOf[ObjectHashAggregateExec])
++          assert(objHashAggOrSortAggPlan.isInstanceOf[ObjectHashAggregateExec] ||
++            objHashAggOrSortAggPlan.isInstanceOf[CometHashAggregateExec])
+         } else {
+           assert(objHashAggOrSortAggPlan.isInstanceOf[SortAggregateExec])
+         }
+@@ -851,12 +855,12 @@ class DataFrameAggregateSuite extends QueryTest
        assert(sortAggPlans.isEmpty)
 
        val objHashAggPlans = collect(aggPlan) {

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

@@ -57,7 +57,7 @@
 - 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`).
+- 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. Comet rewrites intermediate `Partial`/`PartialMerge` object-aggregate buffer attributes from Spark's serialized `BinaryType` to that native list state.
 - 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

native/core/src/execution/merge_as_partial.rs

@@ -30,7 +30,7 @@ use std::fmt::Debug;
 use std::hash::{Hash, Hasher};
 
 use arrow::array::{ArrayRef, BooleanArray};
-use arrow::datatypes::{DataType, FieldRef};
+use arrow::datatypes::{DataType, Field, FieldRef};
 use datafusion::common::Result;
 use datafusion::logical_expr::function::AccumulatorArgs;
 use datafusion::logical_expr::function::StateFieldsArgs;
@@ -41,6 +41,8 @@ use datafusion::logical_expr::{
 use datafusion::physical_expr::aggregate::AggregateFunctionExpr;
 use datafusion::scalar::ScalarValue;
 
+use crate::execution::spark_aggregate_state::PartialMergeStateDecoder;
+
 /// An AggregateUDF wrapper that gives merge semantics in Partial mode.
 ///
 /// When DataFusion runs an AggregateExec in Partial mode, it calls `update_batch`
@@ -60,6 +62,10 @@ pub struct MergeAsPartialUDF {
     cached_state_fields: Vec<FieldRef>,
     /// Cached signature that accepts state field types.
     signature: Signature,
+    /// Override argument fields used only when constructing the inner accumulator.
+    accumulator_expr_fields: Option<Vec<FieldRef>>,
+    /// Decoder for Spark JVM state, or pass-through for native-compatible state.
+    state_decoder: PartialMergeStateDecoder,
     /// Name for this wrapper.
     name: String,
 }
@@ -83,6 +89,9 @@ impl MergeAsPartialUDF {
         let name = format!("merge_as_partial_{}", inner_expr.name());
         let return_type = inner_expr.field().data_type().clone();
         let cached_state_fields = inner_expr.state_fields()?;
+        let accumulator_expr_fields =
+            Self::accumulator_expr_fields(inner_expr, &cached_state_fields);
+        let state_decoder = PartialMergeStateDecoder::new(inner_expr, &cached_state_fields);
 
         // Use a permissive signature since we accept state field types which
         // vary per aggregate function.
@@ -93,9 +102,44 @@ impl MergeAsPartialUDF {
             return_type,
             cached_state_fields,
             signature,
+            accumulator_expr_fields,
+            state_decoder,
             name,
         })
     }
+
+    fn accumulator_expr_fields(
+        inner_expr: &AggregateFunctionExpr,
+        state_fields: &[FieldRef],
+    ) -> Option<Vec<FieldRef>> {
+        // Spark collect_list/collect_set use a single list state field but their accumulator
+        // constructors expect the original scalar input type. MergeAsPartial receives state-typed
+        // input columns, so passing those fields through would make the collect accumulator think
+        // it is aggregating arrays and create nested list state (List(List(T))).
+        match (inner_expr.fun().name(), state_fields) {
+            ("collect_list" | "collect_set", [state_field]) => match state_field.data_type() {
+                DataType::List(item_field) => Some(vec![Field::new(
+                    item_field.name(),
+                    item_field.data_type().clone(),
+                    item_field.is_nullable(),
+                )
+                .into()]),
+                _ => None,
+            },
+            _ => None,
+        }
+    }
+
+    fn inner_accumulator_args<'a>(&'a self, args: AccumulatorArgs<'a>) -> AccumulatorArgs<'a> {
+        let expr_fields = self
+            .accumulator_expr_fields
+            .as_deref()
+            .unwrap_or(args.expr_fields);
+        AccumulatorArgs {
+            expr_fields,
+            ..args
+        }
+    }
 }
 
 impl AggregateUDFImpl for MergeAsPartialUDF {
@@ -120,27 +164,30 @@ impl AggregateUDFImpl for MergeAsPartialUDF {
     }
 
     fn accumulator(&self, args: AccumulatorArgs) -> Result<Box<dyn Accumulator>> {
-        // args.exprs are state-typed (match this wrapper's signature), not the
-        // inner aggregate's original inputs. Safe for built-ins (SUM/COUNT/
-        // MIN/MAX/AVG) which build accumulators from return_type; aggregates
-        // that inspect args.exprs types would need reconsideration.
-        let inner_acc = self.inner_udf.accumulator(args)?;
-        Ok(Box::new(MergeAsPartialAccumulator { inner: inner_acc }))
+        let inner_acc = self
+            .inner_udf
+            .accumulator(self.inner_accumulator_args(args))?;
+        Ok(Box::new(MergeAsPartialAccumulator {
+            inner: inner_acc,
+            state_decoder: self.state_decoder.clone(),
+        }))
     }
 
     fn groups_accumulator_supported(&self, args: AccumulatorArgs) -> bool {
-        // See `accumulator`: args.exprs are state-typed.
-        self.inner_udf.groups_accumulator_supported(args)
+        self.inner_udf
+            .groups_accumulator_supported(self.inner_accumulator_args(args))
     }
 
     fn create_groups_accumulator(
         &self,
         args: AccumulatorArgs,
     ) -> Result<Box<dyn GroupsAccumulator>> {
-        // See `accumulator`: args.exprs are state-typed.
-        let inner_acc = self.inner_udf.create_groups_accumulator(args)?;
+        let inner_acc = self
+            .inner_udf
+            .create_groups_accumulator(self.inner_accumulator_args(args))?;
         Ok(Box::new(MergeAsPartialGroupsAccumulator {
             inner: inner_acc,
+            state_decoder: self.state_decoder.clone(),
         }))
     }
 
@@ -160,6 +207,7 @@ impl AggregateUDFImpl for MergeAsPartialUDF {
 /// Accumulator wrapper that redirects update_batch to merge_batch.
 struct MergeAsPartialAccumulator {
     inner: Box<dyn Accumulator>,
+    state_decoder: PartialMergeStateDecoder,
 }
 
 impl Debug for MergeAsPartialAccumulator {
@@ -171,11 +219,13 @@ impl Debug for MergeAsPartialAccumulator {
 impl Accumulator for MergeAsPartialAccumulator {
     fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
         // Redirect update to merge — this is the key trick.
-        self.inner.merge_batch(values)
+        let decoded = self.state_decoder.decode(values)?;
+        self.inner.merge_batch(decoded.as_ref())
     }
 
     fn merge_batch(&mut self, states: &[ArrayRef]) -> Result<()> {
-        self.inner.merge_batch(states)
+        let decoded = self.state_decoder.decode(states)?;
+        self.inner.merge_batch(decoded.as_ref())
     }
 
     fn evaluate(&mut self) -> Result<ScalarValue> {
@@ -194,6 +244,7 @@ impl Accumulator for MergeAsPartialAccumulator {
 /// GroupsAccumulator wrapper that redirects update_batch to merge_batch.
 struct MergeAsPartialGroupsAccumulator {
     inner: Box<dyn GroupsAccumulator>,
+    state_decoder: PartialMergeStateDecoder,
 }
 
 impl Debug for MergeAsPartialGroupsAccumulator {
@@ -211,8 +262,13 @@ impl GroupsAccumulator for MergeAsPartialGroupsAccumulator {
         total_num_groups: usize,
     ) -> Result<()> {
         // Redirect update to merge — this is the key trick.
-        self.inner
-            .merge_batch(values, group_indices, opt_filter, total_num_groups)
+        let decoded = self.state_decoder.decode(values)?;
+        self.inner.merge_batch(
+            decoded.as_ref(),
+            group_indices,
+            opt_filter,
+            total_num_groups,
+        )
     }
 
     fn merge_batch(
@@ -222,8 +278,13 @@ impl GroupsAccumulator for MergeAsPartialGroupsAccumulator {
         opt_filter: Option<&BooleanArray>,
         total_num_groups: usize,
     ) -> Result<()> {
-        self.inner
-            .merge_batch(values, group_indices, opt_filter, total_num_groups)
+        let decoded = self.state_decoder.decode(values)?;
+        self.inner.merge_batch(
+            decoded.as_ref(),
+            group_indices,
+            opt_filter,
+            total_num_groups,
+        )
     }
 
     fn evaluate(&mut self, emit_to: EmitTo) -> Result<ArrayRef> {

native/core/src/execution/mod.rs

@@ -26,6 +26,7 @@ pub(crate) mod planner;
 pub mod serde;
 pub use datafusion_comet_shuffle as shuffle;
 pub(crate) mod sort;
+pub(crate) mod spark_aggregate_state;
 pub(crate) mod spark_plan;
 pub use datafusion_comet_spark_expr::timezone;
 mod memory_pools;