From 6747cef65f2658ab5247b65d31d0e675638a92f3 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Mon, 19 Oct 2026 16:29:43 +0100 Subject: [PATCH] fix: fall back to Spark for mode on FLOAT and DOUBLE before Spark 3.5.2 Before SPARK-45599, Spark's OpenHashMap hashes a float by its bits but matches keys with ==, so no two NaNs match and each NaN counts as a key of its own. Native mode counts all NaNs as one value, so with allowIncompatible set it could return NaN where Spark 3.4 returns the value that is actually most frequent. Report FLOAT and DOUBLE input as Unsupported on Spark 3.4, 3.5.0 and 3.5.1, so allowIncompatible does not reach it, and correct the comments that claimed every version collapses NaN. Closes #6738. --- .../contributor-guide/floating_point.md | 3 +- docs/source/user-guide/latest/expressions.md | 2 +- native/spark-expr/src/agg_funcs/mode.rs | 17 ++-- .../org/apache/comet/serde/aggregates.scala | 29 +++++- .../sql-tests/expressions/aggregate/mode.sql | 63 +------------ .../aggregate/mode_floating_point.sql | 88 +++++++++++++++++++ .../mode_floating_point_spark_3_4.sql | 46 ++++++++++ .../comet/exec/CometAggregateSuite.scala | 31 ++++++- 8 files changed, 204 insertions(+), 75 deletions(-) create mode 100644 spark/src/test/resources/sql-tests/expressions/aggregate/mode_floating_point.sql create mode 100644 spark/src/test/resources/sql-tests/expressions/aggregate/mode_floating_point_spark_3_4.sql diff --git a/docs/source/contributor-guide/floating_point.md b/docs/source/contributor-guide/floating_point.md index 758350db8f1..f4a1d12355d 100644 --- a/docs/source/contributor-guide/floating_point.md +++ b/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 diff --git a/docs/source/user-guide/latest/expressions.md b/docs/source/user-guide/latest/expressions.md index ddf3df28923..3d9febe0acd 100644 --- a/docs/source/user-guide/latest/expressions.md +++ b/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 | diff --git a/native/spark-expr/src/agg_funcs/mode.rs b/native/spark-expr/src/agg_funcs/mode.rs index cc0f9529f0d..9937065c653 100644 --- a/native/spark-expr/src/agg_funcs/mode.rs +++ b/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 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 7ea6863bc92..c22bbd8dca8 100644 --- a/spark/src/main/scala/org/apache/comet/serde/aggregates.scala +++ b/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 diff --git a/spark/src/test/resources/sql-tests/expressions/aggregate/mode.sql b/spark/src/test/resources/sql-tests/expressions/aggregate/mode.sql index 6d984e623ee..0cec5d83ab5 100644 --- a/spark/src/test/resources/sql-tests/expressions/aggregate/mode.sql +++ b/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 diff --git a/spark/src/test/resources/sql-tests/expressions/aggregate/mode_floating_point.sql b/spark/src/test/resources/sql-tests/expressions/aggregate/mode_floating_point.sql new file mode 100644 index 00000000000..74109a35658 --- /dev/null +++ b/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 diff --git a/spark/src/test/resources/sql-tests/expressions/aggregate/mode_floating_point_spark_3_4.sql b/spark/src/test/resources/sql-tests/expressions/aggregate/mode_floating_point_spark_3_4.sql new file mode 100644 index 00000000000..3deb68fb9c1 --- /dev/null +++ b/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 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 24e70b23573..362d38b07b0 100644 --- a/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala +++ b/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala @@ -27,8 +27,8 @@ import org.apache.hadoop.fs.Path import org.apache.spark.{CometListenerBusUtils, SparkConf, SparkThrowable} import org.apache.spark.scheduler.{SparkListener, SparkListenerTaskEnd} import org.apache.spark.sql.{Column, CometTestBase, DataFrame, QueryTest, Row} -import org.apache.spark.sql.catalyst.expressions.{Cast, Literal} -import org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateMode, Corr, Final, Partial, PartialMerge, RegrR2} +import org.apache.spark.sql.catalyst.expressions.{AttributeReference, Cast, Literal} +import org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateMode, Corr, Final, Mode, Partial, PartialMerge, RegrR2} 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} @@ -40,13 +40,13 @@ import org.apache.spark.sql.execution.aggregate.{BaseAggregateExec, HashAggregat import org.apache.spark.sql.execution.exchange.{ReusedExchangeExec, ShuffleExchangeExec} import org.apache.spark.sql.functions.{avg, col, collect_list, collect_set, count_distinct, expr, sort_array, sum} import org.apache.spark.sql.internal.SQLConf -import org.apache.spark.sql.types.{ArrayType, DataTypes, StructField, StructType} +import org.apache.spark.sql.types.{ArrayType, DataTypes, DoubleType, FloatType, IntegerType, 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.rules.CometExecRule -import org.apache.comet.serde.{CometCorr, CometRegrR2, ExprOuterClass, RegrSparkVersions} +import org.apache.comet.serde.{CometCorr, CometMode, CometRegrR2, ExprOuterClass, Incompatible, RegrSparkVersions, Unsupported} import org.apache.comet.testing.{DataGenOptions, FuzzDataGenerator, ParquetGenerator, SchemaGenOptions} /** @@ -4480,4 +4480,27 @@ class CometAggregateSuite extends CometTestBase with AdaptiveSparkPlanHelper { } } + // Before SPARK-45599 (fixed in 3.5.2 and 4.0.0), mode's frequency map hashes a float by its bits + // but matches keys with ==, so every NaN is a key of its own (#6738). + test("mode float keys by Spark patch version") { + Seq("3.4.3", "3.5.0", "3.5.1", "3.5.1-SNAPSHOT").foreach { version => + assert(!CometMode.keysNaNsTogether(version), version) + } + Seq("3.5.2", "3.5.9", "3.6.0", "4.0.0", "4.1.3", "4.2.0-SNAPSHOT").foreach { version => + assert(CometMode.keysNaNsTogether(version), version) + } + } + + test("mode support level for float input follows the Spark version") { + val sparkKeysNaNsTogether = CometMode.keysNaNsTogether(org.apache.spark.SPARK_VERSION) + Seq(FloatType, DoubleType, IntegerType).foreach { dataType => + val level = CometMode.getSupportLevel(Mode(AttributeReference("a", dataType)())) + if (dataType != IntegerType && !sparkKeysNaNsTogether) { + assert(level.isInstanceOf[Unsupported], s"$dataType: $level") + } else { + assert(level.isInstanceOf[Incompatible], s"$dataType: $level") + } + } + } + }