From abd3b3a926384d6d43ba1ad2e45e8316da7673ba Mon Sep 17 00:00:00 2001 From: 0lai0 Date: Wed, 16 Sep 2026 09:59:40 +0300 Subject: [PATCH 1/5] feat: admit scalar floating-point sort keys under strict floating point `CometSortOrder.getSupportLevel` delegated to the recursive `SupportLevel.strictFloatingPointReason`, which rejects a data type that contains a float or double at any nesting level. Under `spark.comet.exec.strictFloatingPoint=true` that rejected scalar FLOAT and DOUBLE sort keys as well as nested ones. Scalar floating-point comparison keys are normalized before native Sort, TopK, Window, WindowGroupLimit, and range partitioning (#5469): NaN payloads are folded together and signed zeros tied, matching Spark's SQLOrderingUtil.compareDoubles/compareFloats. Only the comparison key is normalized, so returned values keep their original NaN representation and zero sign. Those keys are therefore compatible even in strict mode. Match scalar FLOAT/DOUBLE ahead of the shared helper so they report Compatible, and leave every other data type on the existing recursive path. Floats nested in arrays, structs, and maps still sort by Arrow's raw total ordering, under which -0.0 sorts below 0.0 and a sign-bit NaN sorts below -Infinity, so they keep falling back (#5507). The three other callers of `strictFloatingPointReason` are untouched, including `SortArray`, so `sort_array` on a floating-point element type still falls back. The description of `spark.comet.exec.strictFloatingPoint` is updated to say which sorts are still affected. It is scraped into the generated config reference, so it is the only warning many users will read. The subject of the fallback reason is shared with `getIncompatibleReasons()` through one private val, since that text is published per Spark version in the generated compatibility pages and the two must not drift. Tests: the existing scalar test now asserts native execution instead of fallback, and sorts on a unique id last. -0.0 and +0.0 are peers under Spark's comparison, so ordering on the float columns alone leaves ties whose relative order neither engine promises, and Comet and Spark already lay those out differently today with strict mode off. The sibling array and struct tests still assert the strict-mode fallback. A generated matrix covers both types, both directions, both null orderings and compound keys, one test pins that the sort leaves NaN payloads and zero signs untouched, and one covers the TopK path, which reaches the same gate but a different operator. --- .../scala/org/apache/comet/CometConf.scala | 6 +- .../apache/comet/serde/CometSortOrder.scala | 32 +++- .../apache/comet/CometExpressionSuite.scala | 142 +++++++++++++++++- 3 files changed, 163 insertions(+), 17 deletions(-) diff --git a/spark/src/main/scala/org/apache/comet/CometConf.scala b/spark/src/main/scala/org/apache/comet/CometConf.scala index fc858a75a18..e53d351fd98 100644 --- a/spark/src/main/scala/org/apache/comet/CometConf.scala +++ b/spark/src/main/scala/org/apache/comet/CometConf.scala @@ -933,7 +933,11 @@ object CometConf extends ShimCometConf { .category(CATEGORY_EXEC) .doc( "When enabled, fall back to Spark for floating-point operations that may differ from " + - s"Spark, such as when comparing or sorting -0.0 and 0.0. $COMPAT_GUIDE.") + "Spark, such as comparing -0.0 and 0.0, sorting floating-point values nested in " + + "arrays, structs, or maps, or sorting the elements of a floating-point array with " + + "`sort_array`. Scalar `ORDER BY`, window ordering and range partitioning keys are " + + "unaffected, because Comet normalizes those comparison keys to match Spark. " + + s"$COMPAT_GUIDE.") .booleanConf .createWithDefault(false) diff --git a/spark/src/main/scala/org/apache/comet/serde/CometSortOrder.scala b/spark/src/main/scala/org/apache/comet/serde/CometSortOrder.scala index 838a229bb91..f9a31972748 100644 --- a/spark/src/main/scala/org/apache/comet/serde/CometSortOrder.scala +++ b/spark/src/main/scala/org/apache/comet/serde/CometSortOrder.scala @@ -20,22 +20,38 @@ package org.apache.comet.serde import org.apache.spark.sql.catalyst.expressions.{Ascending, Attribute, Descending, NullsFirst, NullsLast, SortOrder} +import org.apache.spark.sql.types.{DoubleType, FloatType} import org.apache.comet.CometConf import org.apache.comet.serde.QueryPlanSerde.exprToProtoInternal object CometSortOrder extends CometExpressionSerde[SortOrder] { + /** + * Subject of both the runtime fallback reason and the generated compatibility docs. Shared so + * the two cannot describe the policy differently. + */ + private val nestedFloatingPointSort = + "Sorting on floating-point values nested in arrays, structs, or maps" + override def getIncompatibleReasons(): Seq[String] = Seq( - "When `" + CometConf.COMET_EXEC_STRICT_FLOATING_POINT.key + "=true`, sorting on" + - " floating-point types is not 100% compatible with Spark") + s"$nestedFloatingPointSort is not 100% compatible with Spark when " + + s"`${CometConf.COMET_EXEC_STRICT_FLOATING_POINT.key}=true`") - override def getSupportLevel(expr: SortOrder): SupportLevel = { - // https://github.com/apache/datafusion-comet/issues/2626 - SupportLevel - .strictFloatingPointReason(expr.child.dataType, "Sorting on floating-point") - .map(reason => Incompatible(Some(reason))) - .getOrElse(Compatible()) + override def getSupportLevel(expr: SortOrder): SupportLevel = expr.child.dataType match { + // Scalar FLOAT/DOUBLE comparison keys are normalized natively (NaN payloads folded together, + // signed zeros tied) for Sort, TopK, Window, WindowGroupLimit, and range partitioning, which + // matches Spark's SQLOrderingUtil. Only the comparison key is normalized; returned values keep + // their original NaN representation and zero sign. So these are compatible even in strict mode. + case _: FloatType | _: DoubleType => Compatible() + // Floating-point values nested in arrays, structs, or maps are still compared with Arrow's raw + // total ordering, under which -0.0 sorts below 0.0 and a sign-bit NaN sorts below -Infinity. + // https://github.com/apache/datafusion-comet/issues/5507 + case dt => + SupportLevel + .strictFloatingPointReason(dt, nestedFloatingPointSort) + .map(reason => Incompatible(Some(reason))) + .getOrElse(Compatible()) } override def convert( diff --git a/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala b/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala index 620e75b43e3..3e43b40028b 100644 --- a/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala @@ -25,8 +25,8 @@ import org.apache.hadoop.fs.Path import org.apache.spark.sql.{Column, CometTestBase, DataFrame, Row} import org.apache.spark.sql.catalyst.expressions.{Alias, Cast, FromUnixTime, InSet, Literal, StructsToJson, TruncDate, TruncTimestamp} import org.apache.spark.sql.catalyst.optimizer.{ConvertToLocalRelation, OptimizeIn, SimplifyExtractValueOps} -import org.apache.spark.sql.comet.CometProjectExec -import org.apache.spark.sql.execution.{ProjectExec, SparkPlan} +import org.apache.spark.sql.comet.{CometProjectExec, CometSortExec, CometTakeOrderedAndProjectExec} +import org.apache.spark.sql.execution.{LocalTableScanExec, ProjectExec, SparkPlan} import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper import org.apache.spark.sql.functions._ import org.apache.spark.sql.internal.SQLConf @@ -60,16 +60,142 @@ class CometExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelper { schema, 1000, DataGenOptions(generateNegativeZero = true)) - df.createOrReplaceTempView("tbl") + + withTempDir { dir => + // `-0.0` and `+0.0` are peers under Spark's comparison, so `ORDER BY c0, c1` alone leaves + // genuine ties whose relative order neither engine promises. Materialize a unique `id` to + // Parquet and sort on it last, making the ordering total so the row-by-row comparison below + // tests the sort keys rather than an unspecified tie order. + val path = new Path(dir.toString, "tbl").toString + df.withColumn("id", monotonically_increasing_id()).write.parquet(path) + spark.read.parquet(path).createOrReplaceTempView("tbl") + + // Scalar floating-point sort keys are normalized natively, so strict mode admits them even + // with allowIncompatible off. Nested floating-point keys are not, and the two tests below + // still assert the strict-mode fallback. + withSQLConf( + CometConf.getExprAllowIncompatConfigKey("SortOrder") -> "false", + CometConf.COMET_EXEC_STRICT_FLOATING_POINT.key -> "true") { + checkSparkAnswerAndOperator( + sql("select * from tbl order by 1, 2, 3"), + Seq(classOf[CometSortExec])) + } + } + } + // https://github.com/apache/datafusion-comet/issues/5506 + // + // Scalar floating-point sort keys are admitted under strict floating point because their + // comparison keys are normalized natively. Signed zeros survive a Parquet round trip, so these + // cases are materialized to Parquet, which also keeps constant folding from deleting the sort. + // NaN payloads do NOT survive that round trip (every payload is canonicalized on write), so + // payload fidelity is covered separately below over a local relation. + // + // Every query ends with a unique `id` so the ordering is total: -0.0 and +0.0 are peers under + // Spark's comparison, and neither engine promises an order within a tie. + private val strictFpSortRows = Seq( + (0, Some(-0.0f), Some(-0.0d), 1), + (1, Some(0.0f), Some(0.0d), 0), + (2, Some(0.0f), Some(0.0d), 1), + (3, Some(-0.0f), Some(-0.0d), 0), + (4, Some(1.0f), Some(1.0d), 0), + (5, Some(-1.0f), Some(-1.0d), 1), + (6, None, None, 0), + (7, None, None, 1), + (8, Some(Float.NaN), Some(Double.NaN), 0), + (9, Some(Float.PositiveInfinity), Some(Double.PositiveInfinity), 1), + (10, Some(Float.NegativeInfinity), Some(Double.NegativeInfinity), 0)) + + for { + col <- Seq("f", "d") + direction <- Seq("ASC", "DESC") + nullOrder <- Seq("NULLS FIRST", "NULLS LAST") + compound <- Seq(false, true) + } { + val label = s"$col $direction $nullOrder" + (if (compound) " with compound key" else "") + test(s"strict floating point: scalar sort on $label") { + withTempDir { dir => + val path = new Path(dir.toString, "strict_fp_sort").toString + strictFpSortRows.toDF("id", "f", "d", "s").write.parquet(path) + spark.read.parquet(path).createOrReplaceTempView("strict_fp_sort") + + val secondary = if (compound) ", s DESC" else "" + val query = + s"SELECT id, f, d, s FROM strict_fp_sort ORDER BY $col $direction $nullOrder$secondary, id" + + withSQLConf( + CometConf.getExprAllowIncompatConfigKey("SortOrder") -> "false", + CometConf.COMET_EXEC_STRICT_FLOATING_POINT.key -> "true") { + checkSparkAnswerAndOperator(sql(query), Seq(classOf[CometSortExec])) + } + } + } + } + + test("strict floating point: scalar TopK sort key") { + // TakeOrderedAndProject reaches the native path through the same CometSortOrder gate as a + // plain sort, but lands on a different operator (Sort with a fetch). Widening admission + // therefore opens this path too, so assert it rather than inferring it. + withTempDir { dir => + val path = new Path(dir.toString, "strict_fp_topk").toString + strictFpSortRows.toDF("id", "f", "d", "s").write.parquet(path) + spark.read.parquet(path).createOrReplaceTempView("strict_fp_topk") + + withSQLConf( + CometConf.getExprAllowIncompatConfigKey("SortOrder") -> "false", + CometConf.COMET_EXEC_STRICT_FLOATING_POINT.key -> "true") { + for (col <- Seq("f", "d")) { + checkSparkAnswerAndOperator( + sql(s"SELECT id, f, d FROM strict_fp_topk ORDER BY $col ASC NULLS LAST, id LIMIT 4"), + Seq(classOf[CometTakeOrderedAndProjectExec])) + } + } + } + } + + test("strict floating point: scalar sort keeps NaN payloads and zero signs unchanged") { + // A local relation preserves the raw bits that a Parquet round trip would canonicalize, so + // this is where "only the comparison key is normalized" can actually be asserted. + val negNan = java.lang.Double.longBitsToDouble(0xfff8000000000002L) + val posNan = java.lang.Double.longBitsToDouble(0x7ff8000000000002L) + val negNanF = java.lang.Float.intBitsToFloat(0xffc00002) + val posNanF = java.lang.Float.intBitsToFloat(0x7fc00002) + val rows = Seq( + (0, negNanF, negNan), + (1, posNanF, posNan), + (2, -0.0f, -0.0d), + (3, 0.0f, 0.0d), + (4, 1.0f, 1.0d)) + val expected = rows.map { case (id, f, d) => + id -> (java.lang.Float.floatToRawIntBits(f), java.lang.Double.doubleToRawLongBits(d)) + }.toMap + + rows.toDF("id", "f", "d").createOrReplaceTempView("strict_fp_bits") withSQLConf( CometConf.getExprAllowIncompatConfigKey("SortOrder") -> "false", CometConf.COMET_EXEC_STRICT_FLOATING_POINT.key -> "true") { - checkSparkAnswerAndFallbackReasons( - "select * from tbl order by 1, 2", - Set( - "unsupported range partitioning sort order", - "Sorting on floating-point is not 100% compatible with Spark")) + for (col <- Seq("f", "d")) { + val query = s"SELECT id, f, d FROM strict_fp_bits ORDER BY $col, id" + // LocalTableScanExec has no native counterpart enabled by default; it is the source of + // the unmodified bits, not part of what this test asserts. + checkSparkAnswerAndOperator( + sql(query), + Seq(classOf[CometSortExec]), + classOf[LocalTableScanExec]) + + val actual = sql(query).collect().toSeq + actual.foreach { row => + val bits = ( + java.lang.Float.floatToRawIntBits(row.getFloat(1)), + java.lang.Double.doubleToRawLongBits(row.getDouble(2))) + assert( + bits == expected(row.getInt(0)), + s"row ${row.getInt(0)} had its floating-point bits rewritten by the sort") + } + // The two NaN payloads are peers, so they must be adjacent and ordered by the tiebreaker. + val nanIds = actual.map(_.getInt(0)).filter(id => id == 0 || id == 1) + assert(nanIds == Seq(0, 1), s"NaN peers were not tied and ordered by id: $nanIds") + } } } From b67513151812f16b1f5034ed1e6c361565ef19ea Mon Sep 17 00:00:00 2001 From: 0lai0 Date: Wed, 16 Sep 2026 10:00:02 +0300 Subject: [PATCH 2/5] feat: admit scalar floating-point range partitioning under strict mode `supportedRangePartitioningDataType` rejected FLOAT and DOUBLE whenever `spark.comet.exec.strictFloatingPoint=true`. This is a second admission gate, independent of `CometSortOrder`, so narrowing only the sort gate left a plan that sorts natively but still shuffles on the JVM: `CometSort` over `CometColumnarExchange` rather than `CometExchange`/`CometNativeShuffle`. The native range partitioner normalizes both sides of its comparison. Incoming keys are the serialized sort-order expressions, which carry the `NormalizeNaNAndZero` wrapper, and the sampled boundary rows are normalized before being row-encoded with the same converter. Scalar floating-point keys therefore partition consistently with Spark's ordering, so the strict-mode rejection buys no correctness. Accept scalar FLOAT and DOUBLE, and drop the now-unreachable strict-mode fallback reason. Nested types are unaffected: they already fall through to the catch-all that rejects them. The two policy tests are merged into one parameterized over strict mode, since the expected outcome no longer depends on it. The merged test also drops `SortOrder.allowIncompatible=true`, which the old strict-mode test needed to reach this gate, so it now proves the whole path works without an escape hatch. --- .../shuffle/CometShuffleExchangeExec.scala | 18 +++----- .../comet/exec/CometNativeShuffleSuite.scala | 44 +++++++------------ 2 files changed, 20 insertions(+), 42 deletions(-) diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometShuffleExchangeExec.scala b/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometShuffleExchangeExec.scala index 1f5d06a53af..157c7a1e12e 100644 --- a/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometShuffleExchangeExec.scala +++ b/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometShuffleExchangeExec.scala @@ -525,8 +525,6 @@ object CometShuffleExchangeExec case SinglePartition => // we already checked that the input types are supported case RangePartitioning(orderings, _) => - val strictFloatingPoint = CometConf.COMET_EXEC_STRICT_FLOATING_POINT.get(conf) - /** * Determine which data types are supported as partition columns in native shuffle. * @@ -537,8 +535,10 @@ object CometShuffleExchangeExec def supportedRangePartitioningDataType(dt: DataType): Boolean = dt match { // Collated strings require collation-aware ordering; Comet only compares raw bytes. case st: StringType if isStringCollationType(st) => false - case _: FloatType | _: DoubleType => - !strictFloatingPoint + // The native range partitioner normalizes its comparison keys and its sampled boundary + // rows the same way the native sort does, so scalar floats match Spark's ordering even + // under spark.comet.exec.strictFloatingPoint=true. + case _: FloatType | _: DoubleType => true case _: BooleanType | _: ByteType | _: ShortType | _: IntegerType | _: LongType | _: StringType | _: BinaryType | _: TimestampType | _: TimestampNTZType | _: DecimalType | _: DateType => @@ -562,15 +562,7 @@ object CometShuffleExchangeExec } for (dt <- orderings.map(_.dataType).distinct) { if (!supportedRangePartitioningDataType(dt)) { - val reason = dt match { - case _: FloatType | _: DoubleType if strictFloatingPoint => - s"Range partitioning on $dt is not 100% compatible with Spark, and Comet is " + - s"running with ${CometConf.COMET_EXEC_STRICT_FLOATING_POINT.key}=true. " + - s"${CometConf.COMPAT_GUIDE}" - case _ => - s"unsupported range partitioning data type for native shuffle: $dt" - } - reasons += reason + reasons += s"unsupported range partitioning data type for native shuffle: $dt" } } case RoundRobinPartitioning(_) => diff --git a/spark/src/test/scala/org/apache/comet/exec/CometNativeShuffleSuite.scala b/spark/src/test/scala/org/apache/comet/exec/CometNativeShuffleSuite.scala index ff6ef3b10f0..42dae0d186a 100644 --- a/spark/src/test/scala/org/apache/comet/exec/CometNativeShuffleSuite.scala +++ b/spark/src/test/scala/org/apache/comet/exec/CometNativeShuffleSuite.scala @@ -1153,38 +1153,24 @@ class CometNativeShuffleSuite extends CometTestBase with AdaptiveSparkPlanHelper (doubleValue, i) } - test("range partitioning on floating-point falls back when strictFloatingPoint=true") { - withSQLConf( - CometConf.COMET_SHUFFLE_NATIVE_RANGE_PARTITIONING_ENABLED.key -> "true", - CometConf.COMET_EXEC_STRICT_FLOATING_POINT.key -> "true", - // Bypass the CometSortOrder-level Incompatible check so that only - // supportedRangePartitioningDataType is exercised as the guard. - CometConf.getExprAllowIncompatConfigKey("SortOrder") -> "true") { - withParquetTable(floatingPointRangePartitionData, "tbl") { - Seq(("FLOAT", "FloatType"), ("DOUBLE", "DoubleType")).foreach { - case (sqlType, sparkType) => + // The native range partitioner normalizes its comparison keys and its sampled boundary rows the + // same way the native sort does, so scalar floating-point keys match Spark's ordering whether or + // not strict floating point is on. Neither gate needs the allowIncompatible escape hatch. + Seq("true", "false").foreach { strict => + test( + s"range partitioning on floating-point uses native shuffle when " + + s"strictFloatingPoint=$strict") { + withSQLConf( + CometConf.COMET_SHUFFLE_NATIVE_RANGE_PARTITIONING_ENABLED.key -> "true", + CometConf.COMET_EXEC_STRICT_FLOATING_POINT.key -> strict, + CometConf.getExprAllowIncompatConfigKey("SortOrder") -> "false") { + withParquetTable(floatingPointRangePartitionData, "tbl") { + Seq("FLOAT", "DOUBLE").foreach { sqlType => val df = sql(s"SELECT CAST(_1 AS $sqlType) AS c, _2 FROM tbl") .repartitionByRange(4, $"c") - checkSparkAnswerAndFallbackReason( - df, - s"Range partitioning on $sparkType is not 100% compatible with Spark") - } - } - } - } - - test( - "range partitioning on floating-point uses native shuffle when strictFloatingPoint=false") { - withSQLConf( - CometConf.COMET_SHUFFLE_NATIVE_RANGE_PARTITIONING_ENABLED.key -> "true", - CometConf.COMET_EXEC_STRICT_FLOATING_POINT.key -> "false") { - withParquetTable(floatingPointRangePartitionData, "tbl") { - Seq("FLOAT", "DOUBLE").foreach { sqlType => - val df = sql(s"SELECT CAST(_1 AS $sqlType) AS c, _2 FROM tbl") - .repartitionByRange(4, $"c") - - checkShuffleAnswer(df, 1) + checkShuffleAnswer(df, 1) + } } } } From 8ba90ebd5648936410732fc15190271e885294f9 Mon Sep 17 00:00:00 2001 From: 0lai0 Date: Wed, 16 Sep 2026 10:00:02 +0300 Subject: [PATCH 3/5] test: cover the other paths the sort-order gate admits `CometSortOrder` is the admission gate for nine serde sites, so narrowing it for scalar floating point opens more than `ORDER BY`. Each of these was verified rather than assumed. Sort-merge join synthesizes sort orders and therefore reaches the same gate, which means strict mode used to make FP-keyed joins fall back as a side effect. Comet does not normalize the join keys itself. It compares `join_on`, and correctness rests on Catalyst's NormalizeFloatingNumbers having already wrapped both sides, so the new test pins that dependency. Its fixtures are local relations rather than Parquet tables because a Parquet round trip canonicalizes every NaN payload, which would collapse the two NaN cases into one. The WindowGroupLimit peer test is parameterized over strict mode, and a second window test is added that is deliberately not gated on Spark 3.5, so window ordering has coverage on every supported version. It orders by a floating point column and sums over the default RANGE frame, which spans the peer group, so -0.0 and +0.0 being peers is visible in the result and the expected values do not depend on tie order. Columnar shuffle performs range partitioning on the JVM with Spark's own RangePartitioner, but still probes whether Comet can serialize the sort order, so strict mode used to abandon an exchange it would have executed identically. The new test pins the floating-point half of that. Consulting a native-serde gate on a path that never goes native is tracked in #5971. --- .../exec/CometColumnarShuffleSuite.scala | 22 +++++++++ .../apache/comet/exec/CometJoinSuite.scala | 44 ++++++++++++++++- .../comet/exec/CometWindowExecSuite.scala | 48 +++++++++++++++++-- 3 files changed, 110 insertions(+), 4 deletions(-) diff --git a/spark/src/test/scala/org/apache/comet/exec/CometColumnarShuffleSuite.scala b/spark/src/test/scala/org/apache/comet/exec/CometColumnarShuffleSuite.scala index bf354356386..a44aac42b3a 100644 --- a/spark/src/test/scala/org/apache/comet/exec/CometColumnarShuffleSuite.scala +++ b/spark/src/test/scala/org/apache/comet/exec/CometColumnarShuffleSuite.scala @@ -792,6 +792,28 @@ abstract class CometColumnarShuffleSuite extends CometTestBase with AdaptiveSpar * Checks that `df` produces the same answer as Spark does, and has the `expectedNum` Comet * exchange operators. */ + test("range partitioning on floating-point uses columnar shuffle under strictFloatingPoint") { + // The columnar path partitions on the JVM with Spark's own RangePartitioner, but it still + // probes whether Comet can serialize the sort order. Before #5506 that probe reported + // Incompatible for scalar float and double under strict mode, so this exchange fell back to + // Spark's shuffle for no compatibility reason. Consulting a native-serde gate on a path that + // never goes native is tracked separately in #5971; this test pins the floating-point half of + // it, which #5506 fixed. + withSQLConf( + CometConf.COMET_EXEC_STRICT_FLOATING_POINT.key -> "true", + CometConf.getExprAllowIncompatConfigKey("SortOrder") -> "false") { + withParquetTable( + (0 until 20).map(i => (if (i % 3 == 0) -0.0d else i.toDouble, i)), + "range_fp_tbl") { + Seq("FLOAT", "DOUBLE").foreach { sqlType => + val df = sql(s"SELECT CAST(_1 AS $sqlType) AS c, _2 FROM range_fp_tbl") + .repartitionByRange(4, col("c")) + checkShuffleAnswer(df, 1) + } + } + } + } + private def checkShuffleAnswer(df: DataFrame, expectedNum: Int): Unit = { checkCometExchange(df, expectedNum, false) checkSparkAnswer(df) diff --git a/spark/src/test/scala/org/apache/comet/exec/CometJoinSuite.scala b/spark/src/test/scala/org/apache/comet/exec/CometJoinSuite.scala index 43f837c9ba1..3ca9d5ac5ae 100644 --- a/spark/src/test/scala/org/apache/comet/exec/CometJoinSuite.scala +++ b/spark/src/test/scala/org/apache/comet/exec/CometJoinSuite.scala @@ -31,7 +31,7 @@ import org.apache.spark.sql.catalyst.analysis.UnresolvedRelation import org.apache.spark.sql.catalyst.expressions.{And, AttributeReference, IsNotNull} import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight} import org.apache.spark.sql.comet.{CometBroadcastExchangeExec, CometBroadcastHashJoinExec, CometBroadcastNestedLoopJoinExec, CometFilterExec, CometHashJoinExec, CometNativeScanExec, CometSortMergeJoinExec, CometUnionExec} -import org.apache.spark.sql.execution.SparkPlan +import org.apache.spark.sql.execution.{LocalTableScanExec, SparkPlan} import org.apache.spark.sql.execution.adaptive.{AdaptiveSparkPlanExec, AQEShuffleReadExec} import org.apache.spark.sql.execution.exchange.ReusedExchangeExec import org.apache.spark.sql.internal.SQLConf @@ -180,6 +180,48 @@ class CometJoinSuite extends CometTestBase { } } + test("SortMergeJoin with floating-point key runs natively under strict floating point") { + // CometSortOrder is the admission gate for the sort orders this join synthesizes, so + // narrowing it for scalar floats (#5506) admits FP-keyed sort-merge joins under strict mode + // too. Comet does not normalize the join keys itself: it compares `join_on`, and correctness + // rests on Catalyst's NormalizeFloatingNumbers having already wrapped both sides. This test + // pins that, so a future change to that rule fails here rather than silently. + // + // The fixtures are local relations rather than Parquet tables because a Parquet round trip + // canonicalizes every NaN payload, which would collapse the two NaN cases into one. + val left = Seq( + (1, -0.0d), + (2, 0.0d), + (3, 1.0d), + (4, java.lang.Double.longBitsToDouble(0x7ff8000000000002L)), + (5, java.lang.Double.longBitsToDouble(0xfff8000000000002L))) + val right = Seq( + (10, 0.0d), + (20, -0.0d), + (30, 1.0d), + (40, java.lang.Double.longBitsToDouble(0xfff8000000000002L)), + (50, Double.NaN)) + + withSQLConf( + CometConf.COMET_EXEC_STRICT_FLOATING_POINT.key -> "true", + CometConf.getExprAllowIncompatConfigKey("SortOrder") -> "false", + SQLConf.ADAPTIVE_AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1", + SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1", + SQLConf.PREFER_SORTMERGEJOIN.key -> "true") { + withTempView("fp_left", "fp_right") { + left.toDF("lid", "k").createOrReplaceTempView("fp_left") + right.toDF("rid", "k").createOrReplaceTempView("fp_right") + + // LocalTableScanExec has no native counterpart enabled by default and is only the source + // of the unmodified key bits here. + checkSparkAnswerAndOperator( + sql("SELECT l.lid, r.rid FROM fp_left l JOIN fp_right r ON l.k = r.k"), + Seq(classOf[CometSortMergeJoinExec]), + classOf[LocalTableScanExec]) + } + } + } + test("SortMergeJoin with TimestampType key supports outer joins") { withSQLConf( SQLConf.SESSION_LOCAL_TIMEZONE.key -> "Asia/Kathmandu", diff --git a/spark/src/test/scala/org/apache/comet/exec/CometWindowExecSuite.scala b/spark/src/test/scala/org/apache/comet/exec/CometWindowExecSuite.scala index 9d917d537e4..6450e7c56c8 100644 --- a/spark/src/test/scala/org/apache/comet/exec/CometWindowExecSuite.scala +++ b/spark/src/test/scala/org/apache/comet/exec/CometWindowExecSuite.scala @@ -117,8 +117,15 @@ class CometWindowExecSuite extends CometTestBase { digits.toString() } - for (orderColumn <- Seq("f", "d")) { - test(s"window group limit: $orderColumn NaN and signed zero peers at the cutoff") { + for { + orderColumn <- Seq("f", "d") + // Scalar floating-point order keys are admitted under strict floating point (#5506), so the + // native plan and the rank distribution must be identical either way. + strictFloatingPoint <- Seq("false", "true") + } { + test( + s"window group limit: $orderColumn NaN and signed zero peers at the cutoff " + + s"(strictFloatingPoint=$strictFloatingPoint)") { assume(isSpark35Plus, "WindowGroupLimit was added in Spark 3.5") val positiveFloatNaN = java.lang.Float.intBitsToFloat(0x7fc00001) @@ -159,7 +166,10 @@ class CometWindowExecSuite extends CometTestBase { withSQLConf( SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false", SQLConf.SHUFFLE_PARTITIONS.key -> "1", - CometConf.COMET_EXEC_STRICT_FLOATING_POINT.key -> "false", + CometConf.COMET_EXEC_STRICT_FLOATING_POINT.key -> strictFloatingPoint, + // Defaults to true under CometTestBase; pin it so strict mode exercises the shipped + // admission policy rather than the test harness's escape hatch. + CometConf.getExprAllowIncompatConfigKey("SortOrder") -> "false", CometConf.COMET_EXEC_WINDOW_GROUP_LIMIT_ENABLED.key -> "true") { withTempView("floating_window_peers") { // LocalTableScan preserves NaN payloads, unlike a Parquet round trip. Check the input @@ -200,6 +210,38 @@ class CometWindowExecSuite extends CometTestBase { } } + for (orderColumn <- Seq("f", "d")) { + test(s"window: scalar floating-point order key under strict floating point ($orderColumn)") { + // Deliberately not gated on Spark 3.5, unlike the WindowGroupLimit tests above, so the + // admission narrowed in #5506 has window coverage on every supported Spark version. + // The default RANGE frame spans all peers, so -0.0 and +0.0 landing in one peer group is + // directly visible in the running sum, and the result does not depend on tie order. + withTempDir { dir => + val path = new Path(dir.toString, "window_strict_fp").toString + Seq( + (1, -0.0f, -0.0d), + (2, 0.0f, 0.0d), + (3, 1.0f, 1.0d), + (4, -1.0f, -1.0d), + (5, Float.NaN, Double.NaN)) + .toDF("id", "f", "d") + .write + .parquet(path) + spark.read.parquet(path).createOrReplaceTempView("window_strict_fp") + + withSQLConf( + CometConf.COMET_EXEC_STRICT_FLOATING_POINT.key -> "true", + CometConf.getExprAllowIncompatConfigKey("SortOrder") -> "false", + CometConf.COMET_EXEC_WINDOW_GROUP_LIMIT_ENABLED.key -> "false") { + checkSparkAnswerAndOperator( + sql(s"""SELECT id, SUM(id) OVER (ORDER BY $orderColumn) AS running + |FROM window_strict_fp ORDER BY id""".stripMargin), + Seq(classOf[CometWindowExec])) + } + } + } + } + for { partitionColumn <- Seq("f", "d") (function, groupLimit) <- Seq( From 9fcfbb636da5335f1c57ef9a3b755dfb9443229c Mon Sep 17 00:00:00 2001 From: 0lai0 Date: Wed, 16 Sep 2026 10:00:02 +0300 Subject: [PATCH 4/5] docs: describe which floating-point sorts strict mode still affects The floating-point compatibility page still carried #5469's note that the strict policy was unchanged for scalar sort keys, pointing at #5506. The tuning page stated flatly that sorting on floating-point data is not compatible with Spark, which was already wrong for scalar keys after #5469. Both now scope the claim to ordering keys. `ORDER BY`, window ordering and range partitioning on scalar FLOAT and DOUBLE stay native under strict mode. Values nested in arrays, structs and maps still fall back, and so does `sort_array` on a floating-point element type, which sorts elements rather than ordering rows and is gated separately. Neither page is generated. `GenerateDocs` emits the per-expression compatibility pages and the config reference, so those pick up the source changes in this branch on their own. --- .../latest/compatibility/floating-point.md | 8 +++++--- docs/source/user-guide/latest/tuning.md | 17 ++++++++++++++--- 2 files changed, 19 insertions(+), 6 deletions(-) diff --git a/docs/source/user-guide/latest/compatibility/floating-point.md b/docs/source/user-guide/latest/compatibility/floating-point.md index b39ef7f645f..6d3a3c112b0 100644 --- a/docs/source/user-guide/latest/compatibility/floating-point.md +++ b/docs/source/user-guide/latest/compatibility/floating-point.md @@ -43,6 +43,8 @@ Native sorting of floating-point values nested in arrays or structs still uses A ordering. Nested keys can therefore produce different ordering or rank results from Spark; see [#5507](https://github.com/apache/datafusion-comet/issues/5507). -The existing `spark.comet.exec.strictFloatingPoint=true` fallback policy is unchanged, including -its conservative fallback for scalar floating-point sort keys. Narrowing that scalar-sort -admission policy is tracked in [#5506](https://github.com/apache/datafusion-comet/issues/5506). +Because those scalar comparison keys match Spark, `spark.comet.exec.strictFloatingPoint=true` no +longer forces a fallback for them: scalar `FLOAT` and `DOUBLE` sort keys, window and rank order +keys, and range partitioning keys all stay native under strict mode. Floating-point values nested +in arrays, structs, or maps still fall back under strict mode, because their ordering is the raw +total ordering described above. diff --git a/docs/source/user-guide/latest/tuning.md b/docs/source/user-guide/latest/tuning.md index 25b805d1ccb..dc004206fa8 100644 --- a/docs/source/user-guide/latest/tuning.md +++ b/docs/source/user-guide/latest/tuning.md @@ -216,9 +216,20 @@ suggested) to overlap I/O across files at the cost of extra memory. ## Optimizing Sorting on Floating-Point Values -Sorting on floating-point data types (or complex types containing floating-point values) is not compatible with -Spark if the data contains both zero and negative zero. This is likely an edge case that is not of concern for many users -and sorting on floating-point data can be enabled by setting `spark.comet.expression.SortOrder.allowIncompatible=true`. +Comet normalizes NaN payloads and signed zeros in scalar `FLOAT` and `DOUBLE` ordering keys, so `ORDER BY`, window +ordering and range partitioning on them match Spark and stay native even with +`spark.comet.exec.strictFloatingPoint=true`. Only the comparison key is normalized; returned values keep their original +NaN representation and zero sign. + +Floating-point values nested in arrays, structs, or maps are compared with Arrow's raw total ordering instead, which can +differ from Spark when the data contains both zero and negative zero, or more than one NaN representation. This is likely +an edge case that is not of concern for many users. Setting `spark.comet.exec.strictFloatingPoint=true` makes those +nested cases fall back to Spark, and they can be forced back onto the native path with +`spark.comet.expression.SortOrder.allowIncompatible=true`. + +`sort_array` is separate. It sorts array elements rather than ordering rows, and its elements are compared with Arrow's +raw total ordering, so `spark.comet.exec.strictFloatingPoint=true` makes it fall back even for a scalar floating-point +element type. Use `spark.comet.expression.SortArray.allowIncompatible=true` to keep it native. ## Optimizing Joins From 9cdd5ea28a7a470662a6ebdafff07ea8875f48e2 Mon Sep 17 00:00:00 2001 From: 0lai0 Date: Wed, 16 Sep 2026 20:40:12 +0300 Subject: [PATCH 5/5] test: drop redundant string interpolator in range partitioning test name --- .../scala/org/apache/comet/exec/CometNativeShuffleSuite.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/spark/src/test/scala/org/apache/comet/exec/CometNativeShuffleSuite.scala b/spark/src/test/scala/org/apache/comet/exec/CometNativeShuffleSuite.scala index 42dae0d186a..23d58b97ab0 100644 --- a/spark/src/test/scala/org/apache/comet/exec/CometNativeShuffleSuite.scala +++ b/spark/src/test/scala/org/apache/comet/exec/CometNativeShuffleSuite.scala @@ -1158,7 +1158,7 @@ class CometNativeShuffleSuite extends CometTestBase with AdaptiveSparkPlanHelper // not strict floating point is on. Neither gate needs the allowIncompatible escape hatch. Seq("true", "false").foreach { strict => test( - s"range partitioning on floating-point uses native shuffle when " + + "range partitioning on floating-point uses native shuffle when " + s"strictFloatingPoint=$strict") { withSQLConf( CometConf.COMET_SHUFFLE_NATIVE_RANGE_PARTITIONING_ENABLED.key -> "true",