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 17ab15d1810..d249a1eb745 100644 --- a/spark/src/main/scala/org/apache/comet/serde/aggregates.scala +++ b/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 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 2b9156346c5..0ff5a21bf94 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, 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 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) } }