fix: fall back to Spark for mode on FLOAT and DOUBLE before Spark 3.5.2

docs/source/contributor-guide/floating_point.md

@@ -57,7 +57,8 @@ it runs against:
   `OpenHashMap`. `array_distinct`, `array_union`, `array_intersect` and `array_except` look up the
   elements of a flat array in an `OpenHashSet`, but treat all NaNs as one value.
 - `mode` follows `Double.equals` from Spark 3.5.2 and 4.0.0 until Spark 4.2, which folds `-0.0`
-  into `0.0` first (SPARK-57329).
+  into `0.0` first (SPARK-57329). Before 3.5.2 every NaN is a value of its own, so Comet does not
+  run `mode` on `FLOAT` or `DOUBLE` input there (#6738).
 - `array_distinct` and `array_union` keep signed zeros apart in a flat array before Spark 4.2.0
   (before 3.5.2, unless probing merges them). Spark 4.2.0 normalizes their arguments in the plan
   (SPARK-54918). From 4.0.5, 4.1.4 and 4.2.1 they normalize while they evaluate instead

docs/source/user-guide/latest/expressions.md

@@ -118,7 +118,7 @@ The tables below list every Spark built-in expression with its current status.
 | `median` | ✅ | — | Rewrites to `percentile(col, 0.5)` and runs natively for supported percentile inputs |
 | `min` | ✅ | Native |  |
 | `min_by` | ✅ | Native | Value and ordering must be fixed-length types |
-| `mode` | ✅ | Native | `mode(col)` only; Spark breaks ties non-deterministically, so Comet returns the smallest tied value and falls back by default, opt-in via allowIncompatible |
+| `mode` | ✅ | Native | `mode(col)` only; Spark breaks ties non-deterministically, so Comet returns the smallest tied value and falls back by default, opt-in via allowIncompatible; `FLOAT` and `DOUBLE` input always falls back before Spark 3.5.2, where each NaN is a value of its own |
 | `percentile` | ✅ | Native | Single literal percentage on numeric input runs natively; array of percentages and a frequency argument fall back to Spark |
 | `percentile_approx` | ✅ | Native | Alias of `approx_percentile`; same restrictions apply |
 | `percentile_cont` | ✅ | — | Spark 4.0+ `WITHIN GROUP (ORDER BY ...)`; ascending only runs natively, `DESC` falls back to Spark |

native/spark-expr/src/agg_funcs/mode.rs

@@ -40,19 +40,22 @@ use std::sync::Arc;
 ///
 /// # Float keys
 ///
-/// Spark keys the frequency map on the boxed input value and compares keys with
-/// `OpenHashSet`'s `_data(pos) equals k` (`core/.../util/collection/OpenHashSet.scala:122`), i.e.
-/// `java.lang.Double.equals`, which is defined via `doubleToLongBits`. That collapses every `NaN`
-/// bit pattern to one key but keeps `-0.0` and `0.0` apart. Note that
+/// From Spark 3.5.2 and 4.0.0 (SPARK-45599), Spark keys the frequency map on the boxed input value
+/// and compares keys with `OpenHashSet`'s `_data(pos) equals k`
+/// (`core/.../util/collection/OpenHashSet.scala:122`), i.e. `java.lang.Double.equals`, which is
+/// defined via `doubleToLongBits`. That collapses every `NaN` bit pattern to one key but keeps
+/// `-0.0` and `0.0` apart. Earlier releases match keys with `==`, so no two NaNs match and each
+/// NaN counts as a key of its own, which this function does not reproduce: the Scala side does not
+/// run `mode` on `FLOAT` or `DOUBLE` input before 3.5.2. Note that
 /// `NormalizeFloatingNumbers` does *not* apply here: its `apply` only rewrites `WINDOW` and
 /// `JOIN` patterns, so an aggregate's argument reaches `Mode` un-normalized.
 ///
 /// Spark 4.2.0 changed this. SPARK-57329 ("mode() returns incorrect result when input contains
 /// both -0.0 and 0.0") treats the split `-0.0`/`0.0` counts as a bug and normalizes the key at
 /// update time, so from 4.2.0 on the two fold into a single key. `normalize_neg_zero` therefore
-/// tracks the Spark version Comet is running against: it is `false` for Spark 3.4 through 4.1 and
-/// `true` for 4.2.0+. `NaN` canonicalization is unconditional because every supported version
-/// collapses `NaN` via `doubleToLongBits`.
+/// tracks the Spark version Comet is running against: it is `false` for Spark 3.5.2 through 4.1 and
+/// `true` for 4.2.0+. `NaN` canonicalization is unconditional because every version that reaches
+/// this function collapses `NaN` via `doubleToLongBits`.
 ///
 /// Do not "simplify" this to always normalize: `max_by`/`min_by` need the opposite treatment,
 /// because they compare the ordering column with `SQLOrderingUtil.compareDoubles`, which ties

spark/src/main/scala/org/apache/comet/serde/aggregates.scala

@@ -21,10 +21,12 @@ package org.apache.comet.serde
 
 import scala.jdk.CollectionConverters._
 
+import org.apache.spark.SPARK_VERSION
 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.util.ArrayData
 import org.apache.spark.sql.comet.CometExecUtils
+import org.apache.spark.sql.comet.util.Utils
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types.{ArrayType, BinaryType, BooleanType, ByteType, DataType, DateType, DayTimeIntervalType, DecimalType, DoubleType, FloatType, IntegerType, LongType, MapType, NullType, NumericType, ShortType, StringType, StructType, TimestampNTZType, TimestampType, YearMonthIntervalType}
 
@@ -1418,8 +1420,28 @@ object CometMode extends CometAggregateExpressionSerde[Mode] with CometTypeShim
       " iteration order); Comet returns the smallest of the tied values instead" +
       " (https://github.com/apache/datafusion-comet/issues/3970)"
 
+  private val nanKeysReason =
+    "Before Spark 3.5.2 (SPARK-45599), mode counts each NaN as a key of its own, because its" +
+      " frequency map matches keys with == but hashes their bits; native mode counts all NaNs" +
+      " as one value, so FLOAT and DOUBLE input falls back to Spark"
+
   override def getIncompatibleReasons(): Seq[String] = Seq(tieBreakReason)
 
+  override def getUnsupportedReasons(): Seq[String] = Seq(nanKeysReason)
+
+  /**
+   * Whether Spark's frequency map matches float keys with `Double.equals`, so that all NaNs are
+   * one key. That holds from 3.5.2 and 4.0.0 (SPARK-45599). Every 3.4 release, 3.5.0 and 3.5.1
+   * match with `==` but hash the bits, so two NaNs never match.
+   */
+  private[comet] def keysNaNsTogether(version: String): Boolean =
+    Utils.majorMinorPatchVersion(version).exists {
+      case (3, 5, patch) => patch >= 2
+      case (major, minor, _) => major > 3 || (major == 3 && minor > 5)
+    }
+
+  private def isFloat(dt: DataType): Boolean = dt == FloatType || dt == DoubleType
+
   private def isSupportedType(dt: DataType): Boolean = dt match {
     case BooleanType => true
     case ByteType | ShortType | IntegerType | LongType => true
@@ -1448,6 +1470,9 @@ object CometMode extends CometAggregateExpressionSerde[Mode] with CometTypeShim
             "(https://github.com/apache/datafusion-comet/issues/2190)"))
     } else if (!isSupportedType(expr.child.dataType)) {
       Unsupported(Some(s"mode does not support input type ${expr.child.dataType}"))
+    } else if (isFloat(expr.child.dataType) && !keysNaNsTogether(SPARK_VERSION)) {
+      // The answer differs from Spark even without a tie, so allowIncompatible does not apply.
+      Unsupported(Some(nanKeysReason))
     } else {
       Incompatible(Some(tieBreakReason))
     }
@@ -1469,7 +1494,9 @@ object CometMode extends CometAggregateExpressionSerde[Mode] with CometTypeShim
       builder.setDatatype(dataType.get)
       // Spark 4.2.0 (SPARK-57329) normalizes `-0.0` to `0.0` before keying the frequency map;
       // earlier versions key the raw boxed value, where `java.lang.Double.equals` keeps `-0.0`
-      // and `0.0` apart. The native side has to match whichever Spark we are running against.
+      // and `0.0` apart. Float input only reaches here from Spark 3.5.2 and 4.0.0 on, where the
+      // map matches keys with `equals` (see `keysNaNsTogether`). The native side has to match
+      // whichever Spark we are running against.
       builder.setNormalizeNegZero(isSpark42Plus)
       Some(
         ExprOuterClass.AggExpr

spark/src/test/resources/sql-tests/expressions/aggregate/mode.sql

@@ -112,67 +112,8 @@ INSERT INTO mode_nums VALUES
 query
 SELECT grp, mode(b), mode(s), mode(l) FROM mode_nums GROUP BY grp ORDER BY grp
 
--- ============================================================
--- Float / Double
---
--- Whether `-0.0` and `0.0` share a frequency-map key is version-dependent: Spark 3.4-4.1 key on
--- `java.lang.Double.equals` and keep them apart, while Spark 4.2.0+ folds `-0.0` into `0.0`
--- (SPARK-57329). Each group below still has a unique winner under both behaviours.
---
--- Negative zero must be written as `-0.0D`, not `CAST(-0.0 AS DOUBLE)`: an unsuffixed `-0.0` is a
--- DecimalType literal, and Decimal has no signed zero, so the cast yields `+0.0` and the column
--- would silently contain no negative zeros at all.
--- ============================================================
-
-statement
-CREATE TABLE mode_double(v double, grp string) USING parquet
-
-statement
-INSERT INTO mode_double VALUES
-  (1.5, 'a'), (1.5, 'a'), (2.5, 'a'), (NULL, 'a'),
-  (0.0D, 'b'), (-0.0D, 'b'), (-0.0D, 'b'), (7.0, 'b')
-
-query
-SELECT grp, mode(v) FROM mode_double GROUP BY grp ORDER BY grp
-
--- ============================================================
--- Signed zeros: which value wins depends on whether -0.0 and 0.0 share a key
---
--- The counts are -0.0:2, 0.0:2, 5.0:3 (the SPARK-57329 reproducer). Keeping the zeros apart makes
--- 5.0 the unique winner; folding them together makes the zero win with 4. The two candidate
--- answers differ in magnitude, so this catches the divergence even where a `-0.0` vs `0.0`
--- difference would not be visible. Neither behaviour produces a tie, so there is no dependence on
--- Spark's non-deterministic tie-break.
--- ============================================================
-
-statement
-CREATE TABLE mode_signed_zero(v double, grp string) USING parquet
-
-statement
-INSERT INTO mode_signed_zero VALUES
-  (-0.0D, 'a'), (-0.0D, 'a'),
-  (0.0D, 'a'), (0.0D, 'a'),
-  (5.0D, 'a'), (5.0D, 'a'), (5.0D, 'a')
-
-query
-SELECT grp, mode(v) FROM mode_signed_zero GROUP BY grp ORDER BY grp
-
--- ============================================================
--- NaN collapses to a single key on every supported version
---
--- `doubleToLongBits` maps every NaN bit pattern to one value, so the two NaNs outvote the single
--- 1.0 and the mode is NaN.
--- ============================================================
-
-statement
-CREATE TABLE mode_nan(v double, grp string) USING parquet
-
-statement
-INSERT INTO mode_nan VALUES
-  (CAST('NaN' AS DOUBLE), 'a'), (CAST('NaN' AS DOUBLE), 'a'), (1.0D, 'a')
-
-query
-SELECT grp, mode(v) FROM mode_nan GROUP BY grp ORDER BY grp
+-- Float / Double input is in mode_floating_point.sql, which needs Spark 3.5.2 or later. Before
+-- that, mode counts each NaN as a key of its own and falls back (mode_floating_point_spark_3_4.sql).
 
 -- ============================================================
 -- Decimal

spark/src/test/resources/sql-tests/expressions/aggregate/mode_floating_point.sql

@@ -0,0 +1,88 @@
+-- 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.
+
+-- Comet's `mode` is opt-in via allowIncompatible because Spark breaks ties non-deterministically.
+-- Every compared query below has a single value with the strictly-highest frequency per group.
+--
+-- Native mode on FLOAT and DOUBLE needs Spark 3.5.2 or later (SPARK-45599), where the frequency map
+-- matches keys with `Double.equals`. MinSparkVersion only reads the minor version, so a run on
+-- 3.5.0 or 3.5.1 falls back and fails the native check. Earlier releases are in
+-- mode_floating_point_spark_3_4.sql.
+-- MinSparkVersion: 3.5
+-- Config: spark.comet.expression.Mode.allowIncompatible=true
+
+-- ============================================================
+-- Float / Double
+--
+-- Whether `-0.0` and `0.0` share a frequency-map key is version-dependent: Spark 3.5.2-4.1 key on
+-- `java.lang.Double.equals` and keep them apart, while Spark 4.2.0+ folds `-0.0` into `0.0`
+-- (SPARK-57329). Each group below still has a unique winner under both behaviours.
+--
+-- Negative zero must be written as `-0.0D`, not `CAST(-0.0 AS DOUBLE)`: an unsuffixed `-0.0` is a
+-- DecimalType literal, and Decimal has no signed zero, so the cast yields `+0.0` and the column
+-- would silently contain no negative zeros at all.
+-- ============================================================
+
+statement
+CREATE TABLE mode_double(v double, grp string) USING parquet
+
+statement
+INSERT INTO mode_double VALUES
+  (1.5, 'a'), (1.5, 'a'), (2.5, 'a'), (NULL, 'a'),
+  (0.0D, 'b'), (-0.0D, 'b'), (-0.0D, 'b'), (7.0, 'b')
+
+query
+SELECT grp, mode(v) FROM mode_double GROUP BY grp ORDER BY grp
+
+-- ============================================================
+-- Signed zeros: which value wins depends on whether -0.0 and 0.0 share a key
+--
+-- The counts are -0.0:2, 0.0:2, 5.0:3 (the SPARK-57329 reproducer). Keeping the zeros apart makes
+-- 5.0 the unique winner; folding them together makes the zero win with 4. The two candidate
+-- answers differ in magnitude, so this catches the divergence even where a `-0.0` vs `0.0`
+-- difference would not be visible. Neither behaviour produces a tie, so there is no dependence on
+-- Spark's non-deterministic tie-break.
+-- ============================================================
+
+statement
+CREATE TABLE mode_signed_zero(v double, grp string) USING parquet
+
+statement
+INSERT INTO mode_signed_zero VALUES
+  (-0.0D, 'a'), (-0.0D, 'a'),
+  (0.0D, 'a'), (0.0D, 'a'),
+  (5.0D, 'a'), (5.0D, 'a'), (5.0D, 'a')
+
+query
+SELECT grp, mode(v) FROM mode_signed_zero GROUP BY grp ORDER BY grp
+
+-- ============================================================
+-- NaN collapses to a single key on every version this file runs on
+--
+-- `doubleToLongBits` maps every NaN bit pattern to one value, so the two NaNs outvote the single
+-- 1.0 and the mode is NaN.
+-- ============================================================
+
+statement
+CREATE TABLE mode_nan(v double, grp string) USING parquet
+
+statement
+INSERT INTO mode_nan VALUES
+  (CAST('NaN' AS DOUBLE), 'a'), (CAST('NaN' AS DOUBLE), 'a'), (1.0D, 'a')
+
+query
+SELECT grp, mode(v) FROM mode_nan GROUP BY grp ORDER BY grp

spark/src/test/resources/sql-tests/expressions/aggregate/mode_floating_point_spark_3_4.sql

@@ -0,0 +1,46 @@
+-- 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.
+
+-- Before Spark 3.5.2 (SPARK-45599), mode keeps its counts in an OpenHashMap that matches keys with
+-- `==` but hashes their bits, so no two NaNs match and each NaN counts as a key of its own. Native
+-- mode counts all NaNs as one value, so FLOAT and DOUBLE input falls back even with
+-- allowIncompatible set (#6738). The native check on other types is in mode.sql.
+-- MaxSparkVersion: 3.4
+-- Config: spark.comet.expression.Mode.allowIncompatible=true
+
+statement
+CREATE TABLE mode_nan_keys(d double, f float, grp string) USING parquet
+
+-- Three NaNs and two 1.0s. Spark 3.4 counts the NaNs separately, so the mode is 1.0. Counting
+-- them together would make it NaN.
+statement
+INSERT INTO mode_nan_keys VALUES
+  (CAST('NaN' AS DOUBLE), CAST('NaN' AS FLOAT), 'a'),
+  (CAST('NaN' AS DOUBLE), CAST('NaN' AS FLOAT), 'a'),
+  (CAST('NaN' AS DOUBLE), CAST('NaN' AS FLOAT), 'a'),
+  (1.0D, CAST(1.0 AS FLOAT), 'a'),
+  (1.0D, CAST(1.0 AS FLOAT), 'a')
+
+query expect_fallback(counts each NaN as a key of its own)
+SELECT grp, mode(d) FROM mode_nan_keys GROUP BY grp ORDER BY grp
+
+query expect_fallback(counts each NaN as a key of its own)
+SELECT grp, mode(f) FROM mode_nan_keys GROUP BY grp ORDER BY grp
+
+-- Without NaN there is nothing to diverge, but the type is still not supported natively.
+query expect_fallback(counts each NaN as a key of its own)
+SELECT mode(d) FROM mode_nan_keys WHERE d = 1.0D