fix: preserve float collect_set semantics in distinct aggregate chains

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

@@ -1214,7 +1214,9 @@ object CometCollectSet extends CometAggregateExpressionSerde[CollectSet] {
         "`collect_set` with `RESPECT NULLS` falls back to Spark, since the native " +
           "implementation always drops null inputs.")
     } else {
-      Nil
+      Seq(
+        "`collect_set` over FLOAT or DOUBLE in a PartialMerge chain (for example, beside " +
+          "`count(DISTINCT ...)`) falls back to Spark to preserve signed-zero and NaN semantics.")
     }
 
   override def getSupportLevel(expr: CollectSet): SupportLevel = {
@@ -1244,6 +1246,16 @@ object CometCollectSet extends CometAggregateExpressionSerde[CollectSet] {
       inputs: Seq[Attribute],
       binding: Boolean,
       conf: SQLConf): Option[ExprOuterClass.AggExpr] = {
+    // Declining the merge also keeps its buffer producers in Spark via the unsafe-partial pass.
+    if (!isSpark42Plus && aggExpr.mode == PartialMerge &&
+      (expr.child.dataType == FloatType || expr.child.dataType == DoubleType)) {
+      withFallbackReason(
+        aggExpr,
+        "collect_set on FLOAT/DOUBLE in PartialMerge requires Spark 4.2 or later " +
+          "to preserve signed-zero and NaN semantics")
+      return None
+    }
+
     val child = aggExpr.mode match {
       // Spark 4.2 (SPARK-57298) normalizes NaN and -0.0 inside CollectSet's buffer conversion
       // (`convertToBufferElement`/`eval` in collect.scala), so the normalization is invisible in

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

@@ -0,0 +1,82 @@
+-- Licensed to the Apache Software Foundation (ASF) under one
+-- or more contributor license agreements.  See the NOTICE file
+-- distributed with this work for additional information
+-- regarding copyright ownership.  The ASF licenses this file
+-- to you under the Apache License, Version 2.0 (the
+-- "License"); you may not use this file except in compliance
+-- with the License.  You may obtain a copy of the License at
+--
+--   http://www.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing,
+-- software distributed under the License is distributed on an
+-- "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+-- KIND, either express or implied.  See the License for the
+-- specific language governing permissions and limitations
+-- under the License.
+
+-- MaxSparkVersion: 4.1
+-- Config: spark.comet.exec.strictFloatingPoint=false
+-- Config: spark.comet.exec.localTableScan.enabled=false
+-- Config: spark.sql.shuffle.partitions=2
+-- ConfigMatrix: spark.sql.execution.useObjectHashAggregateExec=true,false
+-- ConfigMatrix: spark.sql.adaptive.enabled=true,false
+
+statement
+CREATE TABLE cs_distinct_float(grp int, id int, f float, d double) USING parquet
+
+statement
+INSERT INTO cs_distinct_float VALUES
+  (1, 1, CAST('NaN' AS FLOAT), CAST('NaN' AS DOUBLE)),
+  (1, 2, CAST('NaN' AS FLOAT), CAST('NaN' AS DOUBLE)),
+  (1, 3, CAST('-0.0' AS FLOAT), CAST('-0.0' AS DOUBLE)),
+  (1, 4, 0.0, 0.0), (1, 5, 1.0, 1.0), (1, 6, NULL, NULL),
+  (1, 7, CAST('Infinity' AS FLOAT), CAST('Infinity' AS DOUBLE)),
+  (1, 8, CAST('-Infinity' AS FLOAT), CAST('-Infinity' AS DOUBLE)),
+  (2, 1, NULL, NULL),
+  (3, 1, 1.0, 1.0), (3, 2, 1.0, 1.0), (3, 3, 2.0, 2.0)
+
+-- Issue #7203: even with strict floating-point checks disabled, the distinct chain must keep
+-- Spark's signed-zero and repeated-NaN semantics. Check FLOAT and DOUBLE independently.
+query expect_fallback(collect_set on FLOAT/DOUBLE in PartialMerge requires Spark 4.2)
+SELECT grp, count(DISTINCT id), sort_array(collect_set(f))
+FROM cs_distinct_float GROUP BY grp
+
+query expect_fallback(collect_set on FLOAT/DOUBLE in PartialMerge requires Spark 4.2)
+SELECT grp, count(DISTINCT id), sort_array(collect_set(d))
+FROM cs_distinct_float GROUP BY grp
+
+-- VALUES keeps the partial aggregate in Spark and exercises its serialized buffer boundary.
+query expect_fallback(collect_set on FLOAT/DOUBLE in PartialMerge requires Spark 4.2)
+SELECT grp, count(DISTINCT id), sort_array(collect_set(v))
+FROM VALUES
+  (1, 1, CAST('NaN' AS FLOAT)), (1, 2, CAST('NaN' AS FLOAT)),
+  (1, 3, CAST('-0.0' AS FLOAT)), (1, 4, CAST(0.0 AS FLOAT)),
+  (1, 5, CAST(1.0 AS FLOAT)), (1, 6, CAST(NULL AS FLOAT)),
+  (2, 1, CAST(NULL AS FLOAT)) AS t(grp, id, v)
+GROUP BY grp
+
+query expect_fallback(collect_set on FLOAT/DOUBLE in PartialMerge requires Spark 4.2)
+SELECT grp, count(DISTINCT id), sort_array(collect_set(v))
+FROM VALUES
+  (1, 1, CAST('NaN' AS DOUBLE)), (1, 2, CAST('NaN' AS DOUBLE)),
+  (1, 3, CAST('-0.0' AS DOUBLE)), (1, 4, CAST(0.0 AS DOUBLE)),
+  (1, 5, CAST(1.0 AS DOUBLE)), (1, 6, CAST(NULL AS DOUBLE)),
+  (2, 1, CAST(NULL AS DOUBLE)) AS t(grp, id, v)
+GROUP BY grp
+
+-- The new restriction must leave these native paths enabled. Ordinary collect_set on older
+-- Spark still has documented floating-point differences, so use finite nonzero controls.
+query
+SELECT grp, sort_array(collect_set(f)), sort_array(collect_set(d))
+FROM cs_distinct_float WHERE grp = 3 GROUP BY grp
+
+query
+SELECT grp, count(DISTINCT id), sort_array(collect_list(f)), sort_array(collect_list(d))
+FROM cs_distinct_float WHERE grp = 3 GROUP BY grp
+
+query
+SELECT grp, count(DISTINCT id),
+  sort_array(collect_set(named_struct('v', f))),
+  sort_array(collect_set(named_struct('v', d)))
+FROM cs_distinct_float WHERE grp = 3 GROUP BY grp

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

@@ -17,27 +17,31 @@
 
 -- MinSparkVersion: 4.2
 -- Config: spark.comet.exec.strictFloatingPoint=true
+-- Config: spark.sql.shuffle.partitions=2
 -- ConfigMatrix: parquet.enable.dictionary=false,true
+-- ConfigMatrix: spark.sql.execution.useObjectHashAggregateExec=true,false
+-- ConfigMatrix: spark.sql.adaptive.enabled=true,false
 
 statement
-CREATE TABLE cs_norm_scalar(grp string, f float, d double) USING parquet
+CREATE TABLE cs_norm_scalar(grp string, id int, f float, d double) USING parquet
 
 statement
 INSERT INTO cs_norm_scalar VALUES
-  ('ordinary', CAST(1.5 AS FLOAT), CAST(1.1 AS DOUBLE)),
-  ('ordinary', CAST(2.5 AS FLOAT), CAST(2.2 AS DOUBLE)),
-  ('ordinary', CAST(1.5 AS FLOAT), CAST(1.1 AS DOUBLE)),
-  ('ordinary', CAST(NULL AS FLOAT), CAST(NULL AS DOUBLE)),
-  ('nan', CAST('NaN' AS FLOAT), CAST('NaN' AS DOUBLE)),
-  ('nan', CAST('NaN' AS FLOAT), CAST('NaN' AS DOUBLE)),
-  ('nan', CAST(1.0 AS FLOAT), CAST(1.0 AS DOUBLE)),
-  ('infinity', CAST('Infinity' AS FLOAT), CAST('Infinity' AS DOUBLE)),
-  ('infinity', CAST('-Infinity' AS FLOAT), CAST('-Infinity' AS DOUBLE)),
-  ('infinity', CAST('Infinity' AS FLOAT), CAST('Infinity' AS DOUBLE)),
-  ('zero', CAST('-0.0' AS FLOAT), CAST('-0.0' AS DOUBLE)),
-  ('zero', CAST(0.0 AS FLOAT), CAST(0.0 AS DOUBLE)),
-  ('zero', CAST(1.0 AS FLOAT), CAST(1.0 AS DOUBLE)),
-  ('zero', CAST(NULL AS FLOAT), CAST(NULL AS DOUBLE))
+  ('ordinary', 1, CAST(1.5 AS FLOAT), CAST(1.1 AS DOUBLE)),
+  ('ordinary', 2, CAST(2.5 AS FLOAT), CAST(2.2 AS DOUBLE)),
+  ('ordinary', 3, CAST(1.5 AS FLOAT), CAST(1.1 AS DOUBLE)),
+  ('ordinary', 4, CAST(NULL AS FLOAT), CAST(NULL AS DOUBLE)),
+  ('nan', 1, CAST('NaN' AS FLOAT), CAST('NaN' AS DOUBLE)),
+  ('nan', 2, CAST('NaN' AS FLOAT), CAST('NaN' AS DOUBLE)),
+  ('nan', 3, CAST(1.0 AS FLOAT), CAST(1.0 AS DOUBLE)),
+  ('infinity', 1, CAST('Infinity' AS FLOAT), CAST('Infinity' AS DOUBLE)),
+  ('infinity', 2, CAST('-Infinity' AS FLOAT), CAST('-Infinity' AS DOUBLE)),
+  ('infinity', 3, CAST('Infinity' AS FLOAT), CAST('Infinity' AS DOUBLE)),
+  ('zero', 1, CAST('-0.0' AS FLOAT), CAST('-0.0' AS DOUBLE)),
+  ('zero', 2, CAST(0.0 AS FLOAT), CAST(0.0 AS DOUBLE)),
+  ('zero', 3, CAST(1.0 AS FLOAT), CAST(1.0 AS DOUBLE)),
+  ('zero', 4, CAST(NULL AS FLOAT), CAST(NULL AS DOUBLE)),
+  ('null', 1, CAST(NULL AS FLOAT), CAST(NULL AS DOUBLE))
 
 -- Repartition so equal values must also deduplicate across partial buffers.
 query
@@ -46,6 +50,26 @@ FROM (SELECT /*+ REPARTITION(3) */ * FROM cs_norm_scalar)
 GROUP BY grp
 ORDER BY grp
 
+-- Issue #7203: scalar FLOAT/DOUBLE collect_set beside DISTINCT remains native on Spark 4.2.
+query
+SELECT grp, count(DISTINCT id), sort_array(collect_set(f))
+FROM cs_norm_scalar GROUP BY grp
+
+query
+SELECT grp, count(DISTINCT id), sort_array(collect_set(d))
+FROM cs_norm_scalar GROUP BY grp
+
+-- The same aggregate shape also stays native for collect_list and struct-valued collect_set.
+query
+SELECT grp, count(DISTINCT id), sort_array(collect_list(f)), sort_array(collect_list(d))
+FROM cs_norm_scalar WHERE grp = 'ordinary' GROUP BY grp
+
+query
+SELECT grp, count(DISTINCT id),
+  sort_array(collect_set(named_struct('v', f))),
+  sort_array(collect_set(named_struct('v', d)))
+FROM cs_norm_scalar WHERE grp = 'ordinary' GROUP BY grp
+
 -- Exercise the no-GROUP BY aggregate shape while merging partial buffers.
 query
 SELECT sort_array(collect_set(f)), sort_array(collect_set(d))

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

@@ -33,7 +33,7 @@ import org.apache.spark.sql.catalyst.optimizer.EliminateSorts
 import org.apache.spark.sql.catalyst.plans.physical.{HashPartitioning, RangePartitioning}
 import org.apache.spark.sql.comet.{CometBaseAggregateExec, CometFilterExec, CometHashAggregateExec, CometNativeExec, CometProjectExec, CometSortAggregateExec, CometSortExec, CometSortMergeJoinExec}
 import org.apache.spark.sql.comet.execution.shuffle.{CometColumnarShuffle, CometNativeShuffle, CometShuffleExchangeExec}
-import org.apache.spark.sql.execution.{LocalTableScanExec, SparkPlan}
+import org.apache.spark.sql.execution.{LocalTableScanExec, SortExec, SparkPlan}
 import org.apache.spark.sql.execution.SQLExecution
 import org.apache.spark.sql.execution.adaptive.{AdaptiveSparkPlanExec, AdaptiveSparkPlanHelper, ShuffleQueryStageExec}
 import org.apache.spark.sql.execution.aggregate.{BaseAggregateExec, HashAggregateExec, ObjectHashAggregateExec, SortAggregateExec}
@@ -44,7 +44,7 @@ import org.apache.spark.sql.types.{ArrayType, DataTypes, StructField, StructType
 
 import org.apache.comet.CometConf
 import org.apache.comet.CometConf.COMET_EXEC_STRICT_FLOATING_POINT
-import org.apache.comet.CometSparkSessionExtensions.{isSpark35Plus, isSpark40Plus, isSpark41Plus}
+import org.apache.comet.CometSparkSessionExtensions.{isSpark35Plus, isSpark40Plus, isSpark41Plus, isSpark42Plus}
 import org.apache.comet.rules.CometExecRule
 import org.apache.comet.serde.{CometCorr, CometRegrR2, ExprOuterClass, RegrSparkVersions}
 import org.apache.comet.testing.{DataGenOptions, FuzzDataGenerator, ParquetGenerator, SchemaGenOptions}
@@ -601,6 +601,54 @@ class CometAggregateSuite extends CometTestBase with AdaptiveSparkPlanHelper {
     }
   }
 
+  test("float collect_set merges Spark partials natively on Spark 4.2 (issue #7203)") {
+    assume(isSpark42Plus, "Spark 4.2 normalizes floating-point collect_set buffers")
+    // SQL fixtures cover the answers and fallback matrix. This mixed plan needs operator exclusions
+    // and assertions on the consumed plan, which a plain SQL query directive cannot express.
+    for {
+      adaptive <- Seq(false, true)
+      objectHash <- Seq(false, true)
+    } {
+      withSQLConf(
+        COMET_EXEC_STRICT_FLOATING_POINT.key -> "false",
+        SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> adaptive.toString,
+        SQLConf.USE_OBJECT_HASH_AGG.key -> objectHash.toString,
+        SQLConf.SHUFFLE_PARTITIONS.key -> "2",
+        CometConf.COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED.key -> "false") {
+        val data = Seq(
+          (1, 1, Some(Double.NaN)),
+          (1, 2, Some(Double.NaN)),
+          (1, 3, Some(-0.0d)),
+          (1, 4, Some(0.0d)),
+          (1, 5, Some(1.0d)),
+          (1, 6, None),
+          (2, 1, None))
+        val df = data
+          .toDF()
+          .groupBy("_1")
+          .agg(
+            count_distinct(col("_2")),
+            sort_array(collect_set(col("_3").cast("FLOAT"))),
+            sort_array(collect_set(col("_3"))))
+        val (_, plan) = checkSparkAnswerAndOperator(
+          df,
+          Seq(classOf[CometBaseAggregateExec]),
+          classOf[ObjectHashAggregateExec],
+          classOf[SortAggregateExec],
+          classOf[SortExec],
+          classOf[LocalTableScanExec])
+        assert(collect(plan) { case s: LocalTableScanExec => s }.nonEmpty, plan)
+        val nativeAggregates = collect(plan) { case a: CometBaseAggregateExec => a }
+        val sparkAggregates = collect(plan) { case a: BaseAggregateExec => a }
+        assert(nativeAggregates.exists(_.modes.contains(PartialMerge)), plan)
+        assert(nativeAggregates.exists(_.modes.contains(Final)), plan)
+        assert(!nativeAggregates.exists(_.modes == Seq(Partial)), plan)
+        assert(sparkAggregates.nonEmpty, plan)
+        assert(sparkAggregates.forall(_.aggregateExpressions.forall(_.mode == Partial)), plan)
+      }
+    }
+  }
+
   test("collect_set of binary decodes Spark's byte-array buffer elements in PartialMerge") {
     // Spark's CollectSet stores each binary value in its serialized buffer as an UnsafeArrayData
     // of bytes (bufferElementType ArrayType(ByteType)) and unwraps it in eval, while CollectList