You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/09/17 22:51:38 UTC
(datafusion-comet) 01/01: fix: normalize floating-point values in native collect_set (#5166)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-5166-10f0fdd8ca4b86d48237ec54171d0b64d38a8644
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
commit 23a9ff6445dd0b87e99f03aa0a1c98f221551215
Author: Peter Lee <pe...@gmail.com>
AuthorDate: Thu Sep 17 21:56:32 2026 +0000
fix: normalize floating-point values in native collect_set (#5166)
* fix: normalize floating-point values in native collect_set
* fix: complete collect_set normalization handling
* add unsupport reason and explain why adding the spark 4.2 gate
* docs: address review feedback on normalization visibility and Spark-internal risk
- Document the array-input JVM codegen dispatch trade-off in
getCompatibleNotes() so it appears in the generated Compatibility Guide.
- Reference SPARK-57298 in the serde comment and explain why double
normalization cannot occur (normalization lives in CollectSet's buffer
conversion, not the plan; needNormalize short-circuits on
KnownFloatingPointNormalized).
- Expand the CometExecUtils.normalizeFloatingNumbers doc comment with the
compatibility risk of the private[sql] API and the compile-time failure
signal across Spark profiles.
Co-Authored-By: Claude Fable 5 <no...@anthropic.com>
---------
Co-authored-by: Claude Fable 5 <no...@anthropic.com>
---
.../scala/org/apache/comet/serde/aggregates.scala | 60 +++++++++--
.../apache/spark/sql/comet/CometExecUtils.scala | 15 ++-
.../expressions/aggregate/collect_set.sql | 37 -------
.../aggregate/collect_set_floating_fallback.sql | 49 +++++++++
.../aggregate/collect_set_normalization.sql | 112 +++++++++++++++++++++
5 files changed, 227 insertions(+), 46 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 aef649debc..b055669854 100644
--- a/spark/src/main/scala/org/apache/comet/serde/aggregates.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/aggregates.scala
@@ -22,8 +22,9 @@ package org.apache.comet.serde
import scala.jdk.CollectionConverters._
import org.apache.spark.sql.catalyst.expressions.{Attribute, Cast, Expression, Literal}
-import org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateExpression, ApproximatePercentile, Average, BitAndAgg, BitOrAgg, BitXorAgg, BloomFilterAggregate, CentralMomentAgg, CollectList, CollectSet, Corr, Count, Covariance, CovPopulation, CovSample, First, HyperLogLogPlusPlus, Last, Max, MaxBy, MaxMinBy, Min, MinBy, Mode, Percentile, RegrIntercept, RegrR2, RegrReplacement, RegrSlope, RegrSXY, StddevPop, StddevSamp, Sum, VariancePop, VarianceSamp}
+import org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateExpression, ApproximatePercentile, Average, BitAndAgg, BitOrAgg, BitXorAgg, BloomFilterAggregate, CentralMomentAgg, CollectList, CollectSet, Complete, Corr, Count, Covariance, CovPopulation, CovSample, First, HyperLogLogPlusPlus, Last, Max, MaxBy, MaxMinBy, Min, MinBy, Mode, Partial, Percentile, 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
import org.apache.spark.sql.types.{BinaryType, BooleanType, ByteType, DataType, DateType, DecimalType, DoubleType, FloatType, IntegerType, LongType, NumericType, ShortType, StringType, TimestampNTZType, TimestampType}
@@ -1038,11 +1039,41 @@ object CometBloomFilterAggregate extends CometAggregateExpressionSerde[BloomFilt
object CometCollectSet extends CometAggregateExpressionSerde[CollectSet] {
- override def getIncompatibleReasons(): Seq[String] = Seq(
- "Comet deduplicates NaN values (treats `NaN == NaN`) while Spark treats each NaN as a" +
- s" distinct value. When `${COMET_EXEC_STRICT_FLOATING_POINT.key}=true`, `collect_set`" +
- " on floating-point types falls back to Spark unless" +
- " `spark.comet.expression.CollectSet.allowIncompatible=true` is set.")
+ override def getCompatibleNotes(): Seq[String] =
+ if (isSpark42Plus) {
+ Seq(
+ "On Spark 4.2+, `collect_set` inputs are normalized with Spark's recursive" +
+ " floating-point normalizer to match SPARK-57298. For array inputs containing" +
+ " floating-point values the normalizer produces `ArrayTransform`, which Comet" +
+ " executes through the JVM codegen dispatcher instead of fully natively (scalar and" +
+ " struct inputs stay native). When `spark.comet.exec.scalaUDF.codegen.enabled=false`," +
+ " those array cases fall back to Spark.")
+ } else {
+ Nil
+ }
+
+ override def getIncompatibleReasons(): Seq[String] = {
+ if (isSpark42Plus) {
+ Nil
+ } else {
+ Seq(
+ "Before Spark 4.2, Comet deduplicates NaN values (treats `NaN == NaN`) while Spark" +
+ " treats each NaN as a distinct value. Comet treats -0.0 and 0.0 as distinct while" +
+ " Spark treats them as equal." +
+ s" When `${COMET_EXEC_STRICT_FLOATING_POINT.key}=true`, `collect_set` on" +
+ " floating-point types falls back to Spark on those versions unless" +
+ " `spark.comet.expression.CollectSet.allowIncompatible=true` is set.")
+ }
+ }
+
+ override def getUnsupportedReasons(): Seq[String] =
+ if (isSpark42Plus) {
+ Seq(
+ "`collect_set` with `RESPECT NULLS` falls back to Spark, since the native " +
+ "implementation always drops null inputs.")
+ } else {
+ Nil
+ }
override def getSupportLevel(expr: CollectSet): SupportLevel = {
// The native path always drops null inputs. Spark 4.2 added an `ignoreNulls` field to
@@ -1052,12 +1083,14 @@ object CometCollectSet extends CometAggregateExpressionSerde[CollectSet] {
// analysis time, and CometCollectShim.ignoreNulls hardcodes true, making this a no-op.
if (!CometCollectShim.ignoreNulls(expr)) {
Unsupported(Some("collect_set with RESPECT NULLS (ignoreNulls = false) is not supported"))
+ } else if (isSpark42Plus) {
+ Compatible()
} else {
SupportLevel
.strictFloatingPointReason(
expr.children.head.dataType,
"collect_set on floating-point types " +
- "(Comet deduplicates NaN values while Spark treats each NaN as distinct)")
+ "(Comet deduplicates NaN values and distinguishes -0.0 from 0.0, unlike Spark)")
.map(reason => Incompatible(Some(reason)))
.getOrElse(Compatible())
}
@@ -1069,7 +1102,18 @@ object CometCollectSet extends CometAggregateExpressionSerde[CollectSet] {
inputs: Seq[Attribute],
binding: Boolean,
conf: SQLConf): Option[ExprOuterClass.AggExpr] = {
- val child = expr.children.head
+ 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
+ // the plan Comet receives and must be reapplied here on the raw input. `normalize` is
+ // idempotent: it short-circuits on `KnownFloatingPointNormalized`, so an already-normalized
+ // child is never wrapped twice. Keep older versions unchanged to avoid adding JVM codegen
+ // dispatch for nested arrays; their floating-point behavior is documented as incompatible.
+ case Partial | Complete if isSpark42Plus =>
+ CometExecUtils.normalizeFloatingNumbers(expr.children.head)
+ case _ =>
+ expr.children.head
+ }
val childExpr = exprToProto(child, inputs, binding)
val dataType = serializeDataType(expr.dataType)
diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/CometExecUtils.scala b/spark/src/main/scala/org/apache/spark/sql/comet/CometExecUtils.scala
index 8145e563f4..07c44fd195 100644
--- a/spark/src/main/scala/org/apache/spark/sql/comet/CometExecUtils.scala
+++ b/spark/src/main/scala/org/apache/spark/sql/comet/CometExecUtils.scala
@@ -24,7 +24,8 @@ import scala.reflect.ClassTag
import org.apache.spark.{Partition, SparkContext, TaskContext}
import org.apache.spark.rdd.RDD
-import org.apache.spark.sql.catalyst.expressions.{Attribute, NamedExpression, SortOrder}
+import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression, NamedExpression, SortOrder}
+import org.apache.spark.sql.catalyst.optimizer.NormalizeFloatingNumbers
import org.apache.spark.sql.comet.execution.arrow.CometArrowStream
import org.apache.spark.sql.comet.util.Utils
import org.apache.spark.sql.execution.SparkPlan
@@ -36,6 +37,18 @@ import org.apache.comet.serde.QueryPlanSerde.{exprToProto, serializeDataType}
object CometExecUtils {
+ /**
+ * Expose Spark's package-private recursive floating-point normalizer to Comet serde.
+ *
+ * Compatibility note: `NormalizeFloatingNumbers.normalize` is `private[sql]` with no stability
+ * guarantee. It has existed with this signature in every Spark version Comet supports (3.4
+ * through 4.2). Because this is a direct compile-time reference (not reflection), any rename or
+ * signature change in a future Spark version fails the build for that profile loudly; if that
+ * happens, shim this method per Spark version like other `Shim*` classes.
+ */
+ def normalizeFloatingNumbers(expr: Expression): Expression =
+ NormalizeFloatingNumbers.normalize(expr)
+
/**
* Create an empty RDD with the given number of partitions.
*/
diff --git a/spark/src/test/resources/sql-tests/expressions/aggregate/collect_set.sql b/spark/src/test/resources/sql-tests/expressions/aggregate/collect_set.sql
index bf37b8a3b7..1fca2925de 100644
--- a/spark/src/test/resources/sql-tests/expressions/aggregate/collect_set.sql
+++ b/spark/src/test/resources/sql-tests/expressions/aggregate/collect_set.sql
@@ -15,7 +15,6 @@
-- specific language governing permissions and limitations
-- under the License.
--- Config: spark.comet.exec.strictFloatingPoint=true
-- ConfigMatrix: parquet.enable.dictionary=false,true
-- ============================================================
@@ -147,42 +146,6 @@ SELECT grp, sort_array(collect_set(i)) FROM cs_src_intbig GROUP BY grp ORDER BY
query
SELECT grp, sort_array(collect_set(bi)) FROM cs_src_intbig GROUP BY grp ORDER BY grp
--- ============================================================
--- Float (with NULLs, NaN, Inf, -Inf, +0, -0)
--- Comet deduplicates NaN while Spark does not; with
--- strictFloatingPoint=true collect_set falls back to Spark.
--- ============================================================
-
-statement
-CREATE TABLE cs_src_float(v float, grp string) USING parquet
-
-statement
-INSERT INTO cs_src_float VALUES
- (1.5, 'a'), (2.5, 'a'), (1.5, 'a'), (NULL, 'a'),
- (CAST('NaN' AS FLOAT), 'b'), (CAST('NaN' AS FLOAT), 'b'), (1.0, 'b'),
- (CAST('Infinity' AS FLOAT), 'c'), (CAST('-Infinity' AS FLOAT), 'c'), (CAST('Infinity' AS FLOAT), 'c'),
- (CAST(0.0 AS FLOAT), 'd'), (float('-0.0'), 'd'), (1.0, 'd'), (NULL, 'd')
-
-query expect_fallback(not fully compatible with Spark)
-SELECT grp, sort_array(collect_set(v)) FROM cs_src_float GROUP BY grp ORDER BY grp
-
--- ============================================================
--- Double (with NULLs, NaN, Inf, -Inf, +0, -0)
--- ============================================================
-
-statement
-CREATE TABLE cs_src_double(v double, grp string) USING parquet
-
-statement
-INSERT INTO cs_src_double VALUES
- (1.1, 'a'), (2.2, 'a'), (1.1, 'a'), (NULL, 'a'),
- (CAST('NaN' AS DOUBLE), 'b'), (CAST('NaN' AS DOUBLE), 'b'), (1.0, 'b'),
- (CAST('Infinity' AS DOUBLE), 'c'), (CAST('-Infinity' AS DOUBLE), 'c'), (CAST('Infinity' AS DOUBLE), 'c'),
- (0.0, 'd'), (double('-0.0'), 'd'), (1.0, 'd'), (NULL, 'd')
-
-query expect_fallback(not fully compatible with Spark)
-SELECT grp, sort_array(collect_set(v)) FROM cs_src_double GROUP BY grp ORDER BY grp
-
-- ============================================================
-- String (with NULLs and empty string)
-- ============================================================
diff --git a/spark/src/test/resources/sql-tests/expressions/aggregate/collect_set_floating_fallback.sql b/spark/src/test/resources/sql-tests/expressions/aggregate/collect_set_floating_fallback.sql
new file mode 100644
index 0000000000..55bdf13c27
--- /dev/null
+++ b/spark/src/test/resources/sql-tests/expressions/aggregate/collect_set_floating_fallback.sql
@@ -0,0 +1,49 @@
+-- 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=true
+
+statement
+CREATE TABLE cs_fallback_float(v float, grp string) USING parquet
+
+statement
+INSERT INTO cs_fallback_float VALUES
+ (1.5, 'a'), (2.5, 'a'), (1.5, 'a'), (NULL, 'a'),
+ (CAST('NaN' AS FLOAT), 'b'), (CAST('NaN' AS FLOAT), 'b'), (1.0, 'b'),
+ (CAST('Infinity' AS FLOAT), 'c'), (CAST('-Infinity' AS FLOAT), 'c'),
+ (CAST('Infinity' AS FLOAT), 'c'),
+ (CAST(0.0 AS FLOAT), 'd'), (CAST('-0.0' AS FLOAT), 'd'), (1.0, 'd'), (NULL, 'd')
+
+query expect_fallback(not fully compatible with Spark)
+SELECT grp, sort_array(collect_set(v))
+FROM cs_fallback_float GROUP BY grp ORDER BY grp
+
+statement
+CREATE TABLE cs_fallback_double(v double, grp string) USING parquet
+
+statement
+INSERT INTO cs_fallback_double VALUES
+ (1.1, 'a'), (2.2, 'a'), (1.1, 'a'), (NULL, 'a'),
+ (CAST('NaN' AS DOUBLE), 'b'), (CAST('NaN' AS DOUBLE), 'b'), (1.0, 'b'),
+ (CAST('Infinity' AS DOUBLE), 'c'), (CAST('-Infinity' AS DOUBLE), 'c'),
+ (CAST('Infinity' AS DOUBLE), 'c'),
+ (0.0, 'd'), (CAST('-0.0' AS DOUBLE), 'd'), (1.0, 'd'), (NULL, 'd')
+
+query expect_fallback(not fully compatible with Spark)
+SELECT grp, sort_array(collect_set(v))
+FROM cs_fallback_double GROUP BY grp ORDER 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
new file mode 100644
index 0000000000..1f0ce01237
--- /dev/null
+++ b/spark/src/test/resources/sql-tests/expressions/aggregate/collect_set_normalization.sql
@@ -0,0 +1,112 @@
+-- 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.
+
+-- MinSparkVersion: 4.2
+-- Config: spark.comet.exec.strictFloatingPoint=true
+-- ConfigMatrix: parquet.enable.dictionary=false,true
+
+statement
+CREATE TABLE cs_norm_scalar(grp string, 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))
+
+-- Repartition so equal values must also deduplicate across partial buffers.
+query
+SELECT grp, sort_array(collect_set(f)), sort_array(collect_set(d))
+FROM (SELECT /*+ REPARTITION(3) */ * FROM cs_norm_scalar)
+GROUP BY grp
+ORDER 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))
+FROM (
+ SELECT /*+ REPARTITION(3) */ f, d
+ FROM cs_norm_scalar
+ WHERE grp IN ('nan', 'zero')
+)
+
+statement
+CREATE TABLE cs_norm_nested(
+ grp string,
+ s struct<v:double>,
+ a array<float>,
+ deep_a array<struct<v:double>>,
+ deep_s struct<a:array<double>>) USING parquet
+
+statement
+INSERT INTO cs_norm_nested VALUES
+ ('nan',
+ named_struct('v', CAST('NaN' AS DOUBLE)),
+ array(CAST('NaN' AS FLOAT)),
+ array(named_struct('v', CAST('NaN' AS DOUBLE))),
+ named_struct('a', array(CAST('NaN' AS DOUBLE)))),
+ ('nan',
+ named_struct('v', CAST('NaN' AS DOUBLE)),
+ array(CAST('NaN' AS FLOAT)),
+ array(named_struct('v', CAST('NaN' AS DOUBLE))),
+ named_struct('a', array(CAST('NaN' AS DOUBLE)))),
+ ('zero',
+ named_struct('v', CAST('-0.0' AS DOUBLE)),
+ array(CAST('-0.0' AS FLOAT)),
+ array(named_struct('v', CAST('-0.0' AS DOUBLE))),
+ named_struct('a', array(CAST('-0.0' AS DOUBLE)))),
+ ('zero',
+ named_struct('v', CAST(0.0 AS DOUBLE)),
+ array(CAST(0.0 AS FLOAT)),
+ array(named_struct('v', CAST(0.0 AS DOUBLE))),
+ named_struct('a', array(CAST(0.0 AS DOUBLE)))),
+ ('null',
+ CAST(NULL AS STRUCT<v:DOUBLE>),
+ CAST(NULL AS ARRAY<FLOAT>),
+ CAST(NULL AS ARRAY<STRUCT<v:DOUBLE>>),
+ CAST(NULL AS STRUCT<a:ARRAY<DOUBLE>>)),
+ ('null',
+ named_struct('v', CAST(NULL AS DOUBLE)),
+ array(CAST(NULL AS FLOAT)),
+ array(CAST(NULL AS STRUCT<v:DOUBLE>)),
+ named_struct('a', array(CAST(NULL AS DOUBLE)))),
+ ('null',
+ named_struct('v', CAST(NULL AS DOUBLE)),
+ array(CAST(NULL AS FLOAT)),
+ array(CAST(NULL AS STRUCT<v:DOUBLE>)),
+ named_struct('a', array(CAST(NULL AS DOUBLE))))
+
+query
+SELECT grp,
+ sort_array(collect_set(s)),
+ sort_array(collect_set(a)),
+ sort_array(collect_set(deep_a)),
+ sort_array(collect_set(deep_s))
+FROM (SELECT /*+ REPARTITION(3) */ * FROM cs_norm_nested)
+GROUP BY grp
+ORDER BY grp
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org