This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 69e185aa4d82 [SPARK-57298][SQL] collect_set fails to dedupe
float/double NaN/-0.0 by their semantics
69e185aa4d82 is described below
commit 69e185aa4d82694e5b5694b3e181925f420855cf
Author: Eric Yang <[email protected]>
AuthorDate: Mon Jun 8 17:13:33 2026 -0700
[SPARK-57298][SQL] collect_set fails to dedupe float/double NaN/-0.0 by
their semantics
### What changes were proposed in this pull request?
`CollectSet` now normalizes special floating-point values (NaN and `-0.0`)
before inserting them into its deduplication buffer, so it follows Spark's
float/double equality semantics (all NaNs are equal; `-0.0` equals `0.0`). This
covers both scalar and nested (struct/array) `FLOAT`/`DOUBLE` columns:
- **Top-level `DOUBLE`/`FLOAT`:** the buffer element type becomes
`LONG`/`INT` and values are stored as their normalized bit pattern.
`convertToBufferElement` reuses
`NormalizeFloatingNumbers.DOUBLE_NORMALIZER`/`FLOAT_NORMALIZER` (NaN ->
canonical, `-0.0` -> `0.0`) and then `doubleToLongBits`/`floatToIntBits`;
`eval` converts the bits back. Keying on bits is required because the `HashSet`
buffer compares boxed numbers with primitive equality, where `NaN != NaN`.
Normalizing `-0.0` is [...]
- **Complex types recursively containing `FLOAT`/`DOUBLE` (struct/array):**
values are normalized with `NormalizeFloatingNumbers.normalize` and
materialized as `UnsafeRow`/`UnsafeArrayData`, so the buffer deduplicates on
the canonical binary representation. `MapType` is unaffected (already rejected
by `checkInputDataTypes`).
This reuses the normalization logic Spark already applies to other
hash-based array set operation (`NormalizeFloatingNumbers`, case 5).
### Why are the changes needed?
`collect_set` over `FLOAT`/`DOUBLE` did not follow Spark's floating-point
equality semantics and returned elements that should be considered equal:
```sql
-- Top-level NaN:
SELECT collect_set(v) FROM VALUES (double('NaN')), (double('NaN')) AS t(v);
-- before: [NaN, NaN] after: [NaN]
-- Nested -0.0 / 0.0:
SELECT collect_set(a) FROM VALUES (array(-0.0D)), (array(0.0D)) AS t(a);
-- before: [[-0.0], [0.0]] after: [[0.0]]
SELECT collect_set(named_struct('a', v)) FROM VALUES (-0.0D), (0.0D) AS
t(v);
-- before: [{a:-0.0}, {a:0.0}] after: [{a:0.0}]
```
(Top-level `-0.0`/`0.0` already deduplicate today; this PR preserves that
while fixing the cases above.)
### Does this PR introduce _any_ user-facing change?
Yes. For `collect_set` over `FLOAT`/`DOUBLE` columns, including
struct/array columns that contain them:
- duplicate `NaN` values are no longer returned, and
- `-0.0` and `0.0` are deduplicated even when nested in a struct/array.
In addition, a `-0.0` element is now always returned as `0.0`.
`collect_set` is already documented as non-deterministic, and `-0.0`/`0.0` were
already collapsed at the top level, so this only affects the returned
representation.
### How was this patch tested?
New test cases.
### Was this patch authored or co-authored using generative AI tooling?
Yes. Claude Code
Closes #56360 from jiwen624/collect-set-nan-dedup.
Authored-by: Eric Yang <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
(cherry picked from commit 49908a2cf92870bb15381b2cc80da51b51b491ed)
Signed-off-by: Wenchen Fan <[email protected]>
---
.../catalyst/expressions/aggregate/collect.scala | 32 +++++++++++++++
.../optimizer/NormalizeFloatingNumbers.scala | 2 +-
.../apache/spark/sql/DataFrameAggregateSuite.scala | 45 ++++++++++++++++++++++
3 files changed, 78 insertions(+), 1 deletion(-)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/collect.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/collect.scala
index a03491e349a7..8f15acb536e5 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/collect.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/collect.scala
@@ -26,6 +26,7 @@ import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
import
org.apache.spark.sql.catalyst.analysis.TypeCheckResult.{DataTypeMismatch,
TypeCheckSuccess}
import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.optimizer.NormalizeFloatingNumbers
import org.apache.spark.sql.catalyst.trees.UnaryLike
import org.apache.spark.sql.catalyst.types.PhysicalDataType
import org.apache.spark.sql.catalyst.util.{ArrayData, GenericArrayData,
TypeUtils, UnsafeRowUtils}
@@ -206,6 +207,10 @@ case class CollectSet(
override lazy val bufferElementType = child.dataType match {
case BinaryType => ArrayType(ByteType)
+ // Float/double are keyed by their bit pattern (see
convertToBufferElement), so the
+ // buffer holds the integral bits; eval() converts them back to
float/double.
+ case DoubleType => LongType
+ case FloatType => IntegerType
case other => other
}
@@ -222,6 +227,12 @@ case class CollectSet(
buffer
}
+ @transient private lazy val complexNormalizer: Any => Any = {
+ val ref = BoundReference(0, child.dataType, nullable = true)
+ val proj = UnsafeProjection.create(NormalizeFloatingNumbers.normalize(ref))
+ (value: Any) => InternalRow.copyValue(proj(InternalRow(value)).get(0,
child.dataType))
+ }
+
override def convertToBufferElement(value: Any): Any = child.dataType match {
/*
* collect_set() of BinaryType should not return duplicate elements,
@@ -229,6 +240,17 @@ case class CollectSet(
* so we need to use a different catalyst value for arrays
*/
case BinaryType =>
UnsafeArrayData.fromPrimitiveArray(value.asInstanceOf[Array[Byte]])
+ // mutable.HashSet[Any] compares boxed Double/Float with IEEE equality,
where NaN != NaN,
+ // so normalizing the value alone wouldn't collapse NaNs - keying on
doubleToLongBits/
+ // floatToIntBits does (and the NORMALIZER step keeps -0.0/0.0 deduped).
Complex types
+ // instead dedup on a normalized UnsafeRow's binary form.
+ case DoubleType =>
+ java.lang.Double.doubleToLongBits(
+ NormalizeFloatingNumbers.DOUBLE_NORMALIZER(value).asInstanceOf[Double])
+ case FloatType =>
+ java.lang.Float.floatToIntBits(
+ NormalizeFloatingNumbers.FLOAT_NORMALIZER(value).asInstanceOf[Float])
+ case dt if NormalizeFloatingNumbers.needNormalize(dt) =>
complexNormalizer(value)
case _ => InternalRow.copyValue(value)
}
@@ -239,6 +261,16 @@ case class CollectSet(
case null => null
case v => v.asInstanceOf[ArrayData].toByteArray()
}.toArray[Any]
+ case DoubleType =>
+ buffer.iterator.map {
+ case null => null
+ case v => java.lang.Double.longBitsToDouble(v.asInstanceOf[Long])
+ }.toArray[Any]
+ case FloatType =>
+ buffer.iterator.map {
+ case null => null
+ case v => java.lang.Float.intBitsToFloat(v.asInstanceOf[Int])
+ }.toArray[Any]
case _ => buffer.toArray
}
new GenericArrayData(array)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/NormalizeFloatingNumbers.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/NormalizeFloatingNumbers.scala
index 44add1796169..7d5fd9fe5791 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/NormalizeFloatingNumbers.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/NormalizeFloatingNumbers.scala
@@ -117,7 +117,7 @@ object NormalizeFloatingNumbers extends Rule[LogicalPlan] {
case _ => needNormalize(expr.dataType)
}
- private def needNormalize(dt: DataType): Boolean = dt match {
+ private[sql] def needNormalize(dt: DataType): Boolean = dt match {
case FloatType | DoubleType => true
case StructType(fields) => fields.exists(f => needNormalize(f.dataType))
case ArrayType(et, _) => needNormalize(et)
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala
index 694b087d6c3f..9d78e4c6c1a5 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala
@@ -659,6 +659,51 @@ class DataFrameAggregateSuite extends SharedSparkSession
df.selectExpr("sort_array(collect_set(b) RESPECT NULLS)"),
Seq(Row(Seq(null, 2))))
}
+ test("SPARK-57298: collect_set normalizes NaN and -0.0 for floating-point
types") {
+ checkAnswer(
+ sql("SELECT collect_set(v) FROM VALUES (double('NaN')), (double('NaN'))
AS t(v)"),
+ Row(Seq(Double.NaN)))
+ checkAnswer(
+ sql("SELECT collect_set(v) FROM VALUES (float('NaN')), (float('NaN')) AS
t(v)"),
+ Row(Seq(Float.NaN)))
+
+ checkAnswer(
+ sql("SELECT collect_set(v) FROM VALUES (-0.0D), (0.0D) AS t(v)"),
+ Row(Seq(0.0d)))
+ checkAnswer(
+ sql("SELECT collect_set(v) FROM VALUES (float(-0.0)), (float(0.0)) AS
t(v)"),
+ Row(Seq(0.0f)))
+
+ val df = Seq(Double.NaN, Double.NaN, 0.0d, -0.0d,
1.0d).toDF("v").repartition(3)
+ checkAnswer(df.selectExpr("sort_array(collect_set(v))"), Row(Seq(0.0d,
1.0d, Double.NaN)))
+ }
+
+ test("SPARK-57298: collect_set normalizes NaN and -0.0 nested in complex
types") {
+ checkAnswer(
+ sql("SELECT collect_set(named_struct('a', v)) FROM VALUES (-0.0D),
(0.0D) AS t(v)"),
+ Row(Seq(Row(0.0d))))
+ checkAnswer(
+ sql("SELECT collect_set(a) FROM VALUES (array(-0.0D)), (array(0.0D)) AS
t(a)"),
+ Row(Seq(Seq(0.0d))))
+ checkAnswer(
+ sql("SELECT collect_set(a) FROM VALUES (array(float(-0.0))),
(array(float(0.0))) AS t(a)"),
+ Row(Seq(Seq(0.0f))))
+
+ // Nested NaN already deduplicates today, included as a guardrail against
regressions.
+ checkAnswer(
+ sql("SELECT collect_set(named_struct('a', v)) FROM " +
+ "VALUES (double('NaN')), (double('NaN')) AS t(v)"),
+ Row(Seq(Row(Double.NaN))))
+ checkAnswer(
+ sql("SELECT collect_set(a) FROM " +
+ "VALUES (array(double('NaN'))), (array(double('NaN'))) AS t(a)"),
+ Row(Seq(Seq(Double.NaN))))
+ checkAnswer(
+ sql("SELECT collect_set(a) FROM " +
+ "VALUES (array(float('NaN'))), (array(float('NaN'))) AS t(a)"),
+ Row(Seq(Seq(Float.NaN))))
+ }
+
test("collect functions structs") {
val df = Seq((1, 2, 2), (2, 2, 2), (3, 4, 1))
.toDF("a", "x", "y")
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
