test: docs and test-coverage hardening for native collect_list / collect_set

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

@@ -57,9 +57,19 @@
 - 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, so this is a schema-shape difference only and tests using `checkSparkAnswerAndOperator` accept it (the same pattern applies to [collect_set](#collect-set)).
+- Buffer shape (applies equally to `collect_set`): both are `TypedImperativeAggregate`s, so Spark's `aggBufferAttributes` declares the intermediate buffer as `BinaryType` (serialized state) while the native accumulator's `state_fields` is a `List`. `CometBaseAggregate.adjustOutputForNativeState` rewrites the Comet-side Partial output to the list shape. Neither collector can split a Partial and Final across Comet and Spark, and a multi-stage distinct rewrite (which inserts a `PartialMerge` stage) forces the whole chain back to Spark ([#4724](https://github.com/apache/datafusion-comet/issues/4724)).
 - 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.
 
+## collect_set
+
+- Spark 3.4.3 (audited 2026-07-27): `CollectSet` extends `Collect[mutable.HashSet[Any]]`, returns `ArrayType(child.dataType, containsNull = false)`, ignores NULL inputs in `update()` (the same Hive-compatible semantics as `collect_list`), and yields an empty array as `defaultResult`. `nullable = false`. Unlike `CollectList` it overrides `checkInputDataTypes` and rejects any input whose type recursively contains a `MapType` (`UNSUPPORTED_INPUT_TYPE`). `convertToBufferElement` copies the value with `InternalRow.copyValue`, except for `BinaryType`, which is wrapped in an `UnsafeArrayData` so that byte arrays dedup by content rather than by identity. Deduplication is Scala `mutable.HashSet` equality on the boxed value, which for floating-point types is numeric `==`: repeated `NaN`s are each kept as separate elements, while `0.0` and `-0.0` collapse to one. Registered only as `collect_set` in `FunctionRegistry` (there is no second alias, unlike `collect_list`/`array_agg`).
+- Spark 3.5.8 (audited 2026-07-27): identical to 3.4.3.
+- Spark 4.0.1 (audited 2026-07-27): adds `with UnaryLike[Expression]` to the case class, and `checkInputDataTypes` additionally requires `UnsafeRowUtils.isBinaryStable(child.dataType)`, so non-default-collation strings are rejected along with maps. Deduplication semantics unchanged.
+- Spark 4.1.1 (audited 2026-07-27): identical to 4.0.1.
+- Comet implementation: the native side delegates to `datafusion_spark::function::aggregate::collect::SparkCollectSet`, which wraps `DistinctArrayAggAccumulator` with `ignore_nulls = true` in a `NullToEmptyListAccumulator` so a final NULL accumulator state becomes an empty array. The `containsNull` mismatch against Spark's declared output type, and its rationale, are identical to [collect_list](#collect-list).
+- `CometCollectSet` reports `Incompatible` for float and double input when `spark.comet.exec.strictFloatingPoint=true`, because the native distinct comparison treats `NaN == NaN` and collapses repeated `NaN`s into a single element while Spark keeps each one. The native path for floating-point input is then opt-in via `spark.comet.expression.CollectSet.allowIncompatible=true`. All other input types are `Compatible`.
+
 ## 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`.

spark/src/test/resources/sql-tests/expressions/aggregate/collect_list.sql

@@ -247,21 +247,54 @@ query
 SELECT grp, sort_array(collect_list(v)) FROM cl_src_date GROUP BY grp ORDER BY grp
 
 -- ============================================================
--- Timestamp (with NULLs)
+-- Timestamp / Timestamp NTZ (with NULLs). Both are in
+-- QueryPlanSerde.supportedDataType, so both run through the
+-- native SparkCollectList accumulator.
 -- ============================================================
 
 statement
-CREATE TABLE cl_src_ts(v timestamp, grp string) USING parquet
+CREATE TABLE cl_src_ts(v timestamp, v_ntz timestamp_ntz, grp string) USING parquet
 
 statement
 INSERT INTO cl_src_ts VALUES
-  (TIMESTAMP '2024-01-01 00:00:00', 'a'), (TIMESTAMP '2024-06-15 12:30:00', 'a'),
-  (TIMESTAMP '2024-01-01 00:00:00', 'a'), (NULL, 'a'),
-  (TIMESTAMP '1970-01-01 00:00:00', 'b'), (NULL, 'b')
+  (TIMESTAMP '2024-01-01 00:00:00', TIMESTAMP_NTZ '2024-01-01 00:00:00', 'a'),
+  (TIMESTAMP '2024-06-15 12:30:00', TIMESTAMP_NTZ '2024-06-15 12:30:00', 'a'),
+  (TIMESTAMP '2024-01-01 00:00:00', TIMESTAMP_NTZ '2024-01-01 00:00:00', 'a'),
+  (NULL, NULL, 'a'),
+  (TIMESTAMP '1970-01-01 00:00:00', TIMESTAMP_NTZ '1970-01-01 00:00:00', 'b'),
+  (NULL, NULL, 'b')
 
 query
 SELECT grp, sort_array(collect_list(v)) FROM cl_src_ts GROUP BY grp ORDER BY grp
 
+query
+SELECT grp, sort_array(collect_list(v_ntz)) FROM cl_src_ts GROUP BY grp ORDER BY grp
+
+-- ============================================================
+-- ANSI intervals (with NULLs). Neither YearMonthIntervalType nor
+-- DayTimeIntervalType is in QueryPlanSerde.supportedDataType, so
+-- the scan rejects the column and the query falls back to Spark.
+-- ============================================================
+
+statement
+CREATE TABLE cl_src_interval(ym interval year to month, dt interval day to second, grp string)
+USING parquet
+
+statement
+INSERT INTO cl_src_interval VALUES
+  (INTERVAL '1-2' YEAR TO MONTH, INTERVAL '3 04:05:06' DAY TO SECOND, 'a'),
+  (INTERVAL '0-0' YEAR TO MONTH, INTERVAL '0 00:00:00' DAY TO SECOND, 'a'),
+  (INTERVAL '1-2' YEAR TO MONTH, INTERVAL '3 04:05:06' DAY TO SECOND, 'a'),
+  (NULL, NULL, 'a'),
+  (INTERVAL '-3-4' YEAR TO MONTH, INTERVAL '-5 06:07:08' DAY TO SECOND, 'b'),
+  (NULL, NULL, 'b')
+
+query expect_fallback(Unsupported ym of type YearMonthIntervalType)
+SELECT grp, sort_array(collect_list(ym)) FROM cl_src_interval GROUP BY grp ORDER BY grp
+
+query expect_fallback(Unsupported dt of type DayTimeIntervalType)
+SELECT grp, sort_array(collect_list(dt)) FROM cl_src_interval GROUP BY grp ORDER BY grp
+
 -- ============================================================
 -- Mixed with other aggregates
 -- ============================================================

spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala

@@ -87,22 +87,31 @@ class CometAggregateSuite extends CometTestBase with AdaptiveSparkPlanHelper {
     // dependent PartialMerge/Final stages must also fall back rather than crash. See issue #4724
     // for enabling the fully-native distinct path.
     import org.apache.spark.sql.functions.{collect_list, collect_set, sort_array}
-    // Non-native source (LocalTableScan): the buffer-producing Partial runs in Spark.
+    // Non-native source (LocalTableScan): the buffer-producing Partial runs in Spark, so the
+    // dependent PartialMerge/Final stages fall back via the buffer-source check in doConvert.
+    // tagUnsafePartialAggregates does not fire here because the Partial was never convertible.
+    def bufferSourceFallback(fn: String): String = s"Incompatible aggregate function(s): $fn"
     val df = Seq((1, 3, "a"), (1, 2, "b"), (3, 4, "c"), (3, 4, "c"), (3, 5, "d"))
       .toDF("x", "y", "z")
-    checkSparkAnswer(
-      df.groupBy(col("x")).agg(count_distinct(col("y")), sort_array(collect_list(col("z")))))
-    checkSparkAnswer(
-      df.groupBy(col("x")).agg(count_distinct(col("y")), sort_array(collect_set(col("z")))))
-
-    // Native source (Parquet): the whole multi-stage distinct chain must still fall back to
-    // Spark consistently (issue #4724), rather than running a fully-native pipeline that crashes.
+    checkSparkAnswerAndFallbackReason(
+      df.groupBy(col("x")).agg(count_distinct(col("y")), sort_array(collect_list(col("z")))),
+      bufferSourceFallback("collect_list"))
+    checkSparkAnswerAndFallbackReason(
+      df.groupBy(col("x")).agg(count_distinct(col("y")), sort_array(collect_set(col("z")))),
+      bufferSourceFallback("collect_set"))
+
+    // Native source (Parquet): the Partial would otherwise convert, so tagUnsafePartialAggregates
+    // must disable it and force the whole multi-stage distinct chain back to Spark (issue #4724),
+    // rather than running a fully-native pipeline that crashes.
+    val multiStageFallback = "multi-stage CollectList/CollectSet aggregate whose intermediate " +
+      "buffer cannot round-trip in Comet"
     withParquetTable(
       Seq((1, 3, "a"), (1, 2, "b"), (3, 4, "c"), (3, 4, "c"), (3, 5, "d")),
       "t17616") {
       for (fn <- Seq("collect_list", "collect_set")) {
-        checkSparkAnswer(
-          sql(s"SELECT _1, count(distinct _2), sort_array($fn(_3)) FROM t17616 GROUP BY _1"))
+        checkSparkAnswerAndFallbackReasons(
+          sql(s"SELECT _1, count(distinct _2), sort_array($fn(_3)) FROM t17616 GROUP BY _1"),
+          Set(multiStageFallback, bufferSourceFallback(fn)))
       }
     }
   }
@@ -293,6 +302,33 @@ class CometAggregateSuite extends CometTestBase with AdaptiveSparkPlanHelper {
     }
   }
 
+  // collect_list has no Spark-compatible intermediate buffer: Spark declares BinaryType while the
+  // native accumulator produces a list. Disabling either half of the aggregate must therefore
+  // cascade so that neither half runs in Comet - a Spark final cannot read a Comet-produced list,
+  // and adjustOutputForNativeState would misinterpret Spark's Binary buffer as a list if a Comet
+  // final ran above a Spark partial. count(*) is included so the aggregate also carries a
+  // buffer-compatible function, which is the shape most likely to be split by mistake.
+  Seq(
+    ("Comet partial + Spark final", CometConf.COMET_ENABLE_FINAL_HASH_AGGREGATE),
+    ("Spark partial + Comet final", CometConf.COMET_ENABLE_PARTIAL_HASH_AGGREGATE)).foreach {
+    case (name, disabledHalf) =>
+      test(s"mixed engine collect_list: $name matches Spark") {
+        val data = (0 until 100).map(i => (if (i % 11 == 0) None else Some(i), i % 7))
+        withParquetTable(data, "tbl") {
+          withSQLConf(
+            disabledHalf.key -> "false",
+            CometConf.COMET_EXEC_SHUFFLE_ENABLED.key -> "true",
+            CometConf.COMET_SHUFFLE_MODE.key -> "jvm") {
+            val df = sql("SELECT _2, sort_array(collect_list(_1)), count(*) FROM tbl GROUP BY _2")
+            checkSparkAnswer(df)
+            // Without the cascade the surviving half would still convert, so pinning this at zero
+            // is what keeps the test from passing on a regression.
+            assert(getNumCometHashAggregate(df) == 0)
+          }
+        }
+      }
+  }
+
   test("Aggregation without aggregate expressions should use correct result expressions") {
     withSQLConf(
       CometConf.COMET_ENABLED.key -> "true",