From 023d628aec344349b7467127ec7c5687022fe8ac Mon Sep 17 00:00:00 2001 From: LinSimon-901101 Date: Sun, 25 Oct 2026 19:56:37 +0800 Subject: [PATCH 1/5] fix: preserve float collect_set semantics in distinct aggregate chains --- .../org/apache/comet/serde/aggregates.scala | 16 ++- .../comet/exec/CometAggregateSuite.scala | 100 +++++++++++++++++- 2 files changed, 112 insertions(+), 4 deletions(-) diff --git a/spark/src/main/scala/org/apache/comet/serde/aggregates.scala b/spark/src/main/scala/org/apache/comet/serde/aggregates.scala index 70fb6ac0f90..8f243e65b56 100644 --- a/spark/src/main/scala/org/apache/comet/serde/aggregates.scala +++ b/spark/src/main/scala/org/apache/comet/serde/aggregates.scala @@ -22,7 +22,7 @@ package org.apache.comet.serde import scala.jdk.CollectionConverters._ import org.apache.spark.sql.catalyst.expressions.{Attribute, Cast, Divide, Expression, Literal} -import org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateExpression, AggregateFunction, ApproximatePercentile, Average, BitAndAgg, BitOrAgg, BitXorAgg, BloomFilterAggregate, CentralMomentAgg, CollectList, CollectSet, Complete, Corr, Count, Covariance, CovPopulation, CovSample, DeclarativeAggregate, First, HyperLogLogPlusPlus, Last, Max, MaxBy, MaxMinBy, Min, MinBy, Mode, Partial, Percentile, PivotFirst, RegrIntercept, RegrR2, RegrReplacement, RegrSlope, RegrSXY, StddevPop, StddevSamp, Sum, VariancePop, VarianceSamp} +import org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateExpression, AggregateFunction, ApproximatePercentile, Average, BitAndAgg, BitOrAgg, BitXorAgg, BloomFilterAggregate, CentralMomentAgg, CollectList, CollectSet, Complete, Corr, Count, Covariance, CovPopulation, CovSample, DeclarativeAggregate, First, HyperLogLogPlusPlus, Last, Max, MaxBy, MaxMinBy, Min, MinBy, Mode, Partial, PartialMerge, Percentile, PivotFirst, RegrIntercept, RegrR2, RegrReplacement, RegrSlope, RegrSXY, StddevPop, StddevSamp, Sum, VariancePop, VarianceSamp} import org.apache.spark.sql.catalyst.util.ArrayData import org.apache.spark.sql.comet.CometExecUtils import org.apache.spark.sql.internal.SQLConf @@ -1178,7 +1178,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 = { @@ -1208,6 +1210,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 diff --git a/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala b/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala index 47eb0203d9e..2038978f03b 100644 --- a/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala +++ b/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala @@ -31,7 +31,7 @@ import org.apache.spark.sql.catalyst.expressions.{Cast, Literal} import org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateMode, Corr, Final, Partial, PartialMerge, RegrR2} 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} +import org.apache.spark.sql.comet.{CometBaseAggregateExec, CometBatchScanExec, CometFilterExec, CometHashAggregateExec, CometNativeExec, CometNativeScanExec, CometProjectExec, CometScanExec, CometSortAggregateExec, CometSortExec} import org.apache.spark.sql.comet.execution.shuffle.{CometColumnarShuffle, CometNativeShuffle, CometShuffleExchangeExec} import org.apache.spark.sql.execution.LocalTableScanExec import org.apache.spark.sql.execution.SQLExecution @@ -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} @@ -471,6 +471,102 @@ class CometAggregateSuite extends CometTestBase with AdaptiveSparkPlanHelper { } } + for { + dataType <- Seq("FLOAT", "DOUBLE") + nativeInput <- Seq(false, true) + adaptive <- Seq(false, true) + objectHash <- Seq(false, true) + } { + test( + s"collect_set with distinct preserves float semantics (issue #7203): " + + s"$dataType, nativeInput=$nativeInput, adaptive=$adaptive, objectHash=$objectHash") { + 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)) + def check(source: DataFrame): Unit = { + val df = source + .groupBy("_1") + .agg(count_distinct(col("_2")), sort_array(collect_set(col("_3").cast(dataType)))) + val (_, plan) = if (isSpark42Plus) { + checkSparkAnswerAndOperator( + df, + Seq(classOf[CometBaseAggregateExec]), + classOf[ObjectHashAggregateExec], + classOf[SortAggregateExec], + classOf[LocalTableScanExec]) + } else { + checkSparkAnswerAndFallbackReason( + df, + "collect_set on FLOAT/DOUBLE in PartialMerge requires Spark 4.2 or later") + } + if (nativeInput) { + assert( + collect(plan) { + case _: CometScanExec | _: CometBatchScanExec | _: CometNativeScanExec => true + }.nonEmpty, + plan) + } else { + assert(collect(plan) { case s: LocalTableScanExec => s }.nonEmpty, plan) + } + val nativeAggregates = collect(plan) { case a: CometBaseAggregateExec => a } + if (isSpark42Plus) { + assert(nativeAggregates.exists(_.modes.contains(PartialMerge)), plan) + assert(nativeAggregates.exists(_.modes == Seq(Partial)) == nativeInput, plan) + } else { + assert(nativeAggregates.isEmpty, plan) + val sparkAggregates = collect(plan) { case a: BaseAggregateExec => a } + assert(sparkAggregates.exists(_.aggregateExpressions.exists(_.mode == Partial)), plan) + assert( + sparkAggregates.exists(_.aggregateExpressions.exists(_.mode == PartialMerge)), + plan) + assert(sparkAggregates.exists(_.aggregateExpressions.exists(_.mode == Final)), plan) + } + } + if (nativeInput) { + withParquetDataFrame(data, withDictionary = false)(check) + } else { + check(data.toDF()) + } + } + } + } + + test("float collect fallback leaves other native aggregate paths enabled (issue #7203)") { + withSQLConf(COMET_EXEC_STRICT_FLOATING_POINT.key -> "false") { + withParquetTable( + Seq((1, 1, 1.0d), (1, 2, 1.0d), (1, 3, 2.0d)), + "t_collect_float", + withDictionary = false) { + for { + dataType <- Seq("FLOAT", "DOUBLE") + (fn, distinct) <- Seq( + (s"collect_set(CAST(_3 AS $dataType))", false), + (s"collect_list(CAST(_3 AS $dataType))", true), + (s"collect_set(named_struct('v', CAST(_3 AS $dataType)))", true)) + } { + val count = if (distinct) ", count(DISTINCT _2)" else "" + val (_, plan) = checkSparkAnswerAndOperator( + s"SELECT _1, sort_array($fn)$count FROM t_collect_float GROUP BY _1") + val aggregates = collect(plan) { case a: CometBaseAggregateExec => a } + assert(aggregates.exists(_.modes.contains(Partial)), plan) + assert(aggregates.exists(_.modes.contains(Final)), plan) + assert(aggregates.exists(_.modes.contains(PartialMerge)) == distinct, 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 From e9117e2ed389d8ee2b09634b8bf74d03a9f946dd Mon Sep 17 00:00:00 2001 From: LinSimon-901101 Date: Sun, 25 Oct 2026 20:02:00 +0800 Subject: [PATCH 2/5] style: apply scalafix to regression test name --- .../test/scala/org/apache/comet/exec/CometAggregateSuite.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala b/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala index 2038978f03b..b3d5c73aee8 100644 --- a/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala +++ b/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala @@ -478,7 +478,7 @@ class CometAggregateSuite extends CometTestBase with AdaptiveSparkPlanHelper { objectHash <- Seq(false, true) } { test( - s"collect_set with distinct preserves float semantics (issue #7203): " + + "collect_set with distinct preserves float semantics (issue #7203): " + s"$dataType, nativeInput=$nativeInput, adaptive=$adaptive, objectHash=$objectHash") { withSQLConf( COMET_EXEC_STRICT_FLOATING_POINT.key -> "false", From 6029ad293aded67b9233ea75b97f37866bf0d0d2 Mon Sep 17 00:00:00 2001 From: LinSimon-901101 Date: Sun, 25 Oct 2026 20:08:04 +0800 Subject: [PATCH 3/5] test: check native merge above Spark sort partials --- .../scala/org/apache/comet/exec/CometAggregateSuite.scala | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala b/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala index b3d5c73aee8..113170df3e3 100644 --- a/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala +++ b/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, CometBatchScanExec, CometFilterExec, CometHashAggregateExec, CometNativeExec, CometNativeScanExec, CometProjectExec, CometScanExec, CometSortAggregateExec, CometSortExec} import org.apache.spark.sql.comet.execution.shuffle.{CometColumnarShuffle, CometNativeShuffle, CometShuffleExchangeExec} -import org.apache.spark.sql.execution.LocalTableScanExec +import org.apache.spark.sql.execution.{LocalTableScanExec, SortExec} 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} @@ -504,6 +504,7 @@ class CometAggregateSuite extends CometTestBase with AdaptiveSparkPlanHelper { Seq(classOf[CometBaseAggregateExec]), classOf[ObjectHashAggregateExec], classOf[SortAggregateExec], + classOf[SortExec], classOf[LocalTableScanExec]) } else { checkSparkAnswerAndFallbackReason( @@ -520,12 +521,15 @@ class CometAggregateSuite extends CometTestBase with AdaptiveSparkPlanHelper { 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 } if (isSpark42Plus) { assert(nativeAggregates.exists(_.modes.contains(PartialMerge)), plan) + assert(nativeAggregates.exists(_.modes.contains(Final)), plan) assert(nativeAggregates.exists(_.modes == Seq(Partial)) == nativeInput, plan) + assert(sparkAggregates.isEmpty == nativeInput, plan) + assert(sparkAggregates.forall(_.aggregateExpressions.forall(_.mode == Partial)), plan) } else { assert(nativeAggregates.isEmpty, plan) - val sparkAggregates = collect(plan) { case a: BaseAggregateExec => a } assert(sparkAggregates.exists(_.aggregateExpressions.exists(_.mode == Partial)), plan) assert( sparkAggregates.exists(_.aggregateExpressions.exists(_.mode == PartialMerge)), From f1cb09406d43c013f6c647e9680830ab693978cc Mon Sep 17 00:00:00 2001 From: LinSimon-901101 Date: Tue, 27 Oct 2026 01:26:36 +0800 Subject: [PATCH 4/5] test: move float collect_set regressions to SQL fixtures --- ...collect_set_distinct_floating_fallback.sql | 82 +++++++++++++ .../aggregate/collect_set_normalization.sql | 54 ++++++--- .../comet/exec/CometAggregateSuite.scala | 114 +++++------------- 3 files changed, 152 insertions(+), 98 deletions(-) create mode 100644 spark/src/test/resources/sql-tests/expressions/aggregate/collect_set_distinct_floating_fallback.sql diff --git a/spark/src/test/resources/sql-tests/expressions/aggregate/collect_set_distinct_floating_fallback.sql b/spark/src/test/resources/sql-tests/expressions/aggregate/collect_set_distinct_floating_fallback.sql new file mode 100644 index 00000000000..e732afb699b --- /dev/null +++ b/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 diff --git a/spark/src/test/resources/sql-tests/expressions/aggregate/collect_set_normalization.sql b/spark/src/test/resources/sql-tests/expressions/aggregate/collect_set_normalization.sql index 1f0ce012378..80e67d15e32 100644 --- a/spark/src/test/resources/sql-tests/expressions/aggregate/collect_set_normalization.sql +++ b/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)) diff --git a/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala b/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala index 113170df3e3..ef5c06cd1ab 100644 --- a/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala +++ b/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala @@ -31,7 +31,7 @@ import org.apache.spark.sql.catalyst.expressions.{Cast, Literal} import org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateMode, Corr, Final, Partial, PartialMerge, RegrR2} 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, CometBatchScanExec, CometFilterExec, CometHashAggregateExec, CometNativeExec, CometNativeScanExec, CometProjectExec, CometScanExec, CometSortAggregateExec, CometSortExec} +import org.apache.spark.sql.comet.{CometBaseAggregateExec, CometFilterExec, CometHashAggregateExec, CometNativeExec, CometProjectExec, CometSortAggregateExec, CometSortExec} import org.apache.spark.sql.comet.execution.shuffle.{CometColumnarShuffle, CometNativeShuffle, CometShuffleExchangeExec} import org.apache.spark.sql.execution.{LocalTableScanExec, SortExec} import org.apache.spark.sql.execution.SQLExecution @@ -471,15 +471,14 @@ class CometAggregateSuite extends CometTestBase with AdaptiveSparkPlanHelper { } } - for { - dataType <- Seq("FLOAT", "DOUBLE") - nativeInput <- Seq(false, true) - adaptive <- Seq(false, true) - objectHash <- Seq(false, true) - } { - test( - "collect_set with distinct preserves float semantics (issue #7203): " + - s"$dataType, nativeInput=$nativeInput, adaptive=$adaptive, objectHash=$objectHash") { + 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, @@ -494,79 +493,28 @@ class CometAggregateSuite extends CometTestBase with AdaptiveSparkPlanHelper { (1, 5, Some(1.0d)), (1, 6, None), (2, 1, None)) - def check(source: DataFrame): Unit = { - val df = source - .groupBy("_1") - .agg(count_distinct(col("_2")), sort_array(collect_set(col("_3").cast(dataType)))) - val (_, plan) = if (isSpark42Plus) { - checkSparkAnswerAndOperator( - df, - Seq(classOf[CometBaseAggregateExec]), - classOf[ObjectHashAggregateExec], - classOf[SortAggregateExec], - classOf[SortExec], - classOf[LocalTableScanExec]) - } else { - checkSparkAnswerAndFallbackReason( - df, - "collect_set on FLOAT/DOUBLE in PartialMerge requires Spark 4.2 or later") - } - if (nativeInput) { - assert( - collect(plan) { - case _: CometScanExec | _: CometBatchScanExec | _: CometNativeScanExec => true - }.nonEmpty, - plan) - } else { - 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 } - if (isSpark42Plus) { - assert(nativeAggregates.exists(_.modes.contains(PartialMerge)), plan) - assert(nativeAggregates.exists(_.modes.contains(Final)), plan) - assert(nativeAggregates.exists(_.modes == Seq(Partial)) == nativeInput, plan) - assert(sparkAggregates.isEmpty == nativeInput, plan) - assert(sparkAggregates.forall(_.aggregateExpressions.forall(_.mode == Partial)), plan) - } else { - assert(nativeAggregates.isEmpty, plan) - assert(sparkAggregates.exists(_.aggregateExpressions.exists(_.mode == Partial)), plan) - assert( - sparkAggregates.exists(_.aggregateExpressions.exists(_.mode == PartialMerge)), - plan) - assert(sparkAggregates.exists(_.aggregateExpressions.exists(_.mode == Final)), plan) - } - } - if (nativeInput) { - withParquetDataFrame(data, withDictionary = false)(check) - } else { - check(data.toDF()) - } - } - } - } - - test("float collect fallback leaves other native aggregate paths enabled (issue #7203)") { - withSQLConf(COMET_EXEC_STRICT_FLOATING_POINT.key -> "false") { - withParquetTable( - Seq((1, 1, 1.0d), (1, 2, 1.0d), (1, 3, 2.0d)), - "t_collect_float", - withDictionary = false) { - for { - dataType <- Seq("FLOAT", "DOUBLE") - (fn, distinct) <- Seq( - (s"collect_set(CAST(_3 AS $dataType))", false), - (s"collect_list(CAST(_3 AS $dataType))", true), - (s"collect_set(named_struct('v', CAST(_3 AS $dataType)))", true)) - } { - val count = if (distinct) ", count(DISTINCT _2)" else "" - val (_, plan) = checkSparkAnswerAndOperator( - s"SELECT _1, sort_array($fn)$count FROM t_collect_float GROUP BY _1") - val aggregates = collect(plan) { case a: CometBaseAggregateExec => a } - assert(aggregates.exists(_.modes.contains(Partial)), plan) - assert(aggregates.exists(_.modes.contains(Final)), plan) - assert(aggregates.exists(_.modes.contains(PartialMerge)) == distinct, plan) - } + 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) } } } From 252bd8a66ab700cefbb43a8bbe917e3e941b252f Mon Sep 17 00:00:00 2001 From: LinSimon-901101 Date: Wed, 28 Oct 2026 20:19:38 +0800 Subject: [PATCH 5/5] test: compare parallel writer paths without relying on task order --- .../org/apache/comet/parquet/CometParquetWriterSuite.scala | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/spark/src/test/scala/org/apache/comet/parquet/CometParquetWriterSuite.scala b/spark/src/test/scala/org/apache/comet/parquet/CometParquetWriterSuite.scala index e3a22569a24..5f43465f3f2 100644 --- a/spark/src/test/scala/org/apache/comet/parquet/CometParquetWriterSuite.scala +++ b/spark/src/test/scala/org/apache/comet/parquet/CometParquetWriterSuite.scala @@ -2344,7 +2344,9 @@ class CometParquetWriterSuite extends CometParquetWriterTestBase { captureWritePlan(sql("INSERT INTO v2_file_target SELECT * FROM v2_file_source")) assertHasCometNativeWriteExec(plan) assert(RecordingFileSystem.created.size == 2, RecordingFileSystem.created) - assert(RecordingFileSystem.closed.asScala == RecordingFileSystem.created.asScala) + // Parallel tasks can close their files in a different order than they created them. + assert( + RecordingFileSystem.closed.asScala.sorted == RecordingFileSystem.created.asScala.sorted) checkAnswer(spark.table("v2_file_target"), df) } }