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 7050fef2e71b [SPARK-57329][SQL] mode() returns incorrect result when
input contains both -0.0 and 0.0
7050fef2e71b is described below
commit 7050fef2e71b21c8878e1e8ae615f0e70e6b4f79
Author: Eric Yang <[email protected]>
AuthorDate: Sun Jun 21 14:47:02 2026 -0700
[SPARK-57329][SQL] mode() returns incorrect result when input contains both
-0.0 and 0.0
### What changes were proposed in this pull request?
This PR fixes `mode()` (and `pandas_mode()`) returning an incorrect result
when the input contains both -0.0 and 0.0. Mode/PandasMode accumulate value
frequencies in an `OpenHashMap[AnyRef, Long]` keyed by the raw boxed input
value. For FLOAT/DOUBLE, `java.lang.Double.equals/hashCode` are defined via
`doubleToLongBits`, which distinguishes -0.0 from 0.0. As a result the
frequency of a single SQL value is split across two buffer entries, and eval's
`maxBy(_._2)` can then pick a value w [...]
The fix normalizes the floating-point component of the buffer key at
update time, reusing the existing NormalizeFloatingNumbers helpers:
- top-level DOUBLE/FLOAT → DOUBLE_NORMALIZER / FLOAT_NORMALIZER;
- complex types containing float/double → an UnsafeProjection built from
NormalizeFloatingNumbers.normalize;
- all other types → unchanged (InternalRow.copyValue).
### Why are the changes needed?
It is a correctness bug. Spark treats -0.0 = 0.0 under SQL/GROUP BY
semantics everywhere else (e.g. GROUP BY, join keys, array_distinct,
collect_set), so mode() must do the same. Today it does not:
```SQL
SELECT mode(c) FROM VALUES
(0.0D),(0.0D),(-0.0D),(-0.0D),(9.0D),(9.0D),(9.0D) AS t(c);
-- returns 9.0, but the correct mode is 0.0
```
### Does this PR introduce _any_ user-facing change?
Yes, a bug fix.
### How was this patch tested?
New test case added
### Was this patch authored or co-authored using generative AI tooling?
Yes. Claude Code
Closes #56382 from jiwen624/mode-float-zero-dedup.
Authored-by: Eric Yang <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
(cherry picked from commit e20fbb1ed46296ee4f56328262e5a29bc34222cc)
Signed-off-by: Wenchen Fan <[email protected]>
---
.../sql/catalyst/expressions/aggregate/Mode.scala | 27 ++++++++++--
.../apache/spark/sql/DataFrameAggregateSuite.scala | 50 ++++++++++++++++++++++
2 files changed, 73 insertions(+), 4 deletions(-)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Mode.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Mode.scala
index 7a4f04bf04f7..30e00ac68004 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Mode.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Mode.scala
@@ -20,17 +20,30 @@ package org.apache.spark.sql.catalyst.expressions.aggregate
import org.apache.spark.SparkIllegalArgumentException
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.analysis.{ExpressionBuilder,
UnresolvedWithinGroup}
-import org.apache.spark.sql.catalyst.expressions.{Ascending, Descending,
Expression, ExpressionDescription, ImplicitCastInputTypes, SortOrder}
+import org.apache.spark.sql.catalyst.expressions.{Ascending, BoundReference,
Descending, Expression, ExpressionDescription, ImplicitCastInputTypes,
SortOrder, UnsafeProjection}
import org.apache.spark.sql.catalyst.expressions.Cast.toSQLExpr
+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, CollationFactory,
GenericArrayData, MapData, UnsafeRowUtils}
import org.apache.spark.sql.errors.DataTypeErrors.toSQLType
import org.apache.spark.sql.errors.QueryCompilationErrors
-import org.apache.spark.sql.types.{AbstractDataType, AnyDataType, ArrayType,
BooleanType, DataType, MapType, StringType, StructField, StructType}
+import org.apache.spark.sql.types.{AbstractDataType, AnyDataType, ArrayType,
BooleanType, DataType, DoubleType, FloatType, MapType, StringType, StructField,
StructType}
import org.apache.spark.unsafe.types.UTF8String
import org.apache.spark.util.collection.OpenHashMap
+private[aggregate] object ModeKeyNormalizer {
+ def forType(dataType: DataType): Any => Any = dataType match {
+ case DoubleType => NormalizeFloatingNumbers.DOUBLE_NORMALIZER
+ case FloatType => NormalizeFloatingNumbers.FLOAT_NORMALIZER
+ case dt if NormalizeFloatingNumbers.needNormalize(dt) =>
+ val ref = BoundReference(0, dt, nullable = true)
+ val proj =
UnsafeProjection.create(NormalizeFloatingNumbers.normalize(ref))
+ (value: Any) => InternalRow.copyValue(proj(InternalRow(value)).get(0,
dt))
+ case _ => (value: Any) => InternalRow.copyValue(value)
+ }
+}
+
case class Mode(
child: Expression,
mutableAggBufferOffset: Int = 0,
@@ -54,13 +67,16 @@ case class Mode(
override def prettyName: String = "mode"
+ @transient private lazy val keyNormalizer: Any => Any =
+ ModeKeyNormalizer.forType(child.dataType)
+
override def update(
buffer: OpenHashMap[AnyRef, Long],
input: InternalRow): OpenHashMap[AnyRef, Long] = {
val key = child.eval(input)
if (key != null) {
- buffer.changeValue(InternalRow.copyValue(key).asInstanceOf[AnyRef], 1L,
_ + 1L)
+ buffer.changeValue(keyNormalizer(key).asInstanceOf[AnyRef], 1L, _ + 1L)
}
buffer
}
@@ -299,13 +315,16 @@ case class PandasMode(
override def prettyName: String = "pandas_mode"
+ @transient private lazy val keyNormalizer: Any => Any =
+ ModeKeyNormalizer.forType(child.dataType)
+
override def update(
buffer: OpenHashMap[AnyRef, Long],
input: InternalRow): OpenHashMap[AnyRef, Long] = {
val key = child.eval(input)
if (key != null) {
- buffer.changeValue(InternalRow.copyValue(key).asInstanceOf[AnyRef], 1L,
_ + 1L)
+ buffer.changeValue(keyNormalizer(key).asInstanceOf[AnyRef], 1L, _ + 1L)
} else if (!ignoreNA) {
buffer.changeValue(null, 1L, _ + 1L)
}
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 31428ab7108a..b18734d12e70 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
@@ -1091,6 +1091,56 @@ class DataFrameAggregateSuite extends SharedSparkSession
)
}
+ test("SPARK-57329: mode normalizes -0.0/0.0 in the frequency buffer") {
+ checkAnswer(
+ Seq(0.0d, 0.0d, -0.0d, -0.0d, 9.0d, 9.0d,
9.0d).toDF("d").select(expr("mode(d)")),
+ Row(0.0d))
+ checkAnswer(
+ Seq(0.0f, 0.0f, -0.0f, -0.0f, 9.0f, 9.0f,
9.0f).toDF("f").select(expr("mode(f)")),
+ Row(0.0f))
+
+ checkAnswer(
+ Seq(Array(-0.0d), Array(-0.0d), Array(0.0d), Array(0.0d),
+ Array(9.0d), Array(9.0d),
Array(9.0d)).toDF("a").select(expr("mode(a)")),
+ Row(Seq(0.0d)))
+
+ // pandas_mode shares the same normalization path; cover it explicitly. It
is an
+ // internal expression, so invoke it via Column.internalFn rather than SQL.
+ checkAnswer(
+ Seq(0.0d, 0.0d, -0.0d, -0.0d, 9.0d).toDF("d")
+ .select(Column.internalFn("pandas_mode", col("d"), lit(true))),
+ Row(Seq(0.0d)))
+ checkAnswer(
+ Seq(0.0f, 0.0f, -0.0f, -0.0f, 9.0f).toDF("f")
+ .select(Column.internalFn("pandas_mode", col("f"), lit(true))),
+ Row(Seq(0.0f)))
+ checkAnswer(
+ Seq(Array(-0.0d), Array(-0.0d), Array(0.0d), Array(0.0d),
Array(9.0d)).toDF("a")
+ .select(Column.internalFn("pandas_mode", col("a"), lit(true))),
+ Row(Seq(Seq(0.0d))))
+
+ // Struct complex type: same recursive NormalizeFloatingNumbers path as
the array case
+ // above, but a different shape. -0.0/0.0 collapse to 4 occurrences,
outvoting 9.0's 3.
+ checkAnswer(
+ sql("SELECT mode(named_struct('a', v)) FROM " +
+ "VALUES (-0.0D), (-0.0D), (0.0D), (0.0D), (9.0D), (9.0D), (9.0D) AS
t(v)"),
+ Row(Row(0.0d)))
+
+ // NaN with differing bit patterns nested in a complex type. The
normalization lambda
+ // canonicalizes the NaN bits so the two patterns collapse to 4
occurrences, outvoting
+ // 9.0's 3. Without normalization each pattern forms its own group of 2
and 9.0 (3) wins,
+ // so the expected NaN result genuinely distinguishes fixed from buggy
behavior.
+ val nan1 = java.lang.Double.longBitsToDouble(0x7ff8000000000000L)
+ val nan2 = java.lang.Double.longBitsToDouble(0x7ff8000000000001L)
+ assert(nan1.isNaN && nan2.isNaN &&
+ java.lang.Double.doubleToRawLongBits(nan1) !=
+ java.lang.Double.doubleToRawLongBits(nan2))
+ checkAnswer(
+ Seq(nan1, nan1, nan2, nan2, 9.0d, 9.0d, 9.0d).toDF("v")
+ .select(struct(col("v")).as("s")).select(expr("mode(s)")),
+ Row(Row(Double.NaN)))
+ }
+
test("SPARK-27581: DataFrame count_distinct(\"*\") shouldn't fail with
AnalysisException") {
val df = sql("select id % 100 from range(100000)")
val distinctCount1 = df.select(expr("count(distinct(*))"))
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
