fix: match Spark's float ordering for sort and window keys nested in arrays and structs

[andygrove]

Which issue does this PR close?

Closes #5507.

Part of #6385: the "nested sort keys and rank (#5507)" step.

Rationale for this change

Spark orders floats with SQLOrderingUtil.compareDoubles at every depth of an array or struct: -0.0 equals 0.0, all NaNs are equal, and NaN sorts above every other value. #5469 made Comet normalize scalar FLOAT and DOUBLE sort and window keys so that Arrow's total order agrees with Spark's, but a key that nests floats in an array or struct was compared raw. There -0.0 sorts below 0.0, and a NaN with the sign bit set sorts below -Infinity. Negating a NaN sets that bit on any platform, and on x86-64 every NaN that arithmetic produces has it.

With rows holding -0.0, 0.0, a canonical NaN and a sign-bit NaN, among others:

Query	Spark	Comet before
SELECT id FROM t ORDER BY array(IF(s, -d, d)), id DESC	8, 9, 7, 2, 1, 3, 6, 5, 4	8, 5, 9, 7, 1, 2, 3, 6, 4

The two zeros (ids 1 and 2) and the two NaNs (ids 4 and 5) are peers in Spark, so the tiebreaker orders them; Comet split them and sorted the sign-bit NaN first. RANK, DENSE_RANK, window frames and rank limits over such keys had the same gap. spark.comet.exec.strictFloatingPoint=true made these keys fall back to Spark, unless spark.comet.expression.SortOrder.allowIncompatible was set.

What changes are included in this PR?

• create_normalized_key_expr, which builds the keys of Sort, TopK, Window and WindowGroupLimit, wraps an array or struct key with a float at any depth in NormalizeNestedFloats, as it already wrapped a scalar float key in NormalizeNaNAndZero. Only the comparison key is normalized, so the rows keep their original values, zero signs and NaN payloads included.
• CometSortOrder is compatible for every key type in strict floating-point mode too, except an array or struct key that has a float and whose type can hold a null element or field. Those stay Incompatible there, so they fall back unless SortOrder.allowIncompatible is set. Spark orders such a null below every value whatever the null order, and the native sort and RANGE window frames don't (Native sort orders null elements of array and struct keys by the key's null order, unlike Spark #6476, RANGE window frames over an array or struct key with a null element span the whole partition #6477), so these keys can sort or frame differently from Spark. A key whose type cannot hold a null, such as array(coalesce(x, 0.0D)), stays native. Maps cannot be sort keys in Spark.
• The RANK and DENSE_RANK limit guard from fix: fall back to Spark for RANK and DENSE_RANK limits over nested floating-point keys #6468 is gone: CometWindowGroupLimitExec no longer falls back for a nested float order key, because the native planner now normalizes the key the operator compares.
• Range partitioning needs no change: the native range partitioner only accepts scalar keys, and a nested key goes to the JVM shuffle, which partitions with Spark's own ordering.
• The floating-point compatibility guide, the operator compatibility and tuning guides, the spark.comet.exec.strictFloatingPoint description, the native shuffle contributor guide and the shuffle review skill no longer describe nested keys as a gap, and the guides say which keys strict mode still declines.

How are these changes tested?

• windows/nested_float_order_keys.sql (new, run with the default settings): arrays, structs, arrays of structs and structs of arrays of DOUBLE and FLOAT holding both zeros and both kinds of NaN, as keys of ORDER BY in both directions, TopK, RANK, DENSE_RANK, running sums over the default RANGE frame, and rank limits. Every ORDER BY ends with a unique tiebreaker, so peers must come out in its order. On main its first query returns the wrong order shown above.
  • Two nested-null differences that do not involve floats stay out of the fixture: the queries keep the default null orders, and the running sums leave out the row whose keys hold a null element. Spark orders a null element below every value whatever NULLS FIRST or NULLS LAST says, while the native sort ties nested nulls to that option, and DataFusion's RANGE frame bounds order a null element above every value. Both show up with INT keys too, so they are tracked separately as Native sort orders null elements of array and struct keys by the key's null order, unlike Spark #6476 and RANGE window frames over an array or struct key with a null element span the whole partition #6477.
• windows/nested_float_order_keys_strict.sql (new, strict floating-point mode with SortOrder.allowIncompatible=false, so it applies the shipped policy): keys over the nullable columns can hold a null element or field, so ORDER BY, TopK, RANK, DENSE_RANK, running sums and rank limits over them fall back, and the answers match Spark. Natively, ORDER BY array(d) NULLS LAST puts the row with a null element last where Spark puts it first, and the running sum spans the whole partition. The same queries over coalesce(x, 0.0D) keys, whose types cannot hold a null, stay native.
• The rank-limit unit test floating_sort_keys_preserve_window_group_limit_peers runs its sign-bit, payload and signaling NaNs, zeros and nulls through a one-element list and a one-field struct as well as bare, and checks the same peers and the same bits. It fails at the first list key without the change.
• CometWindowExecSuite: "window group limit: floating-point values nested in the order key" expected the fix: fall back to Spark for RANK and DENSE_RANK limits over nested floating-point keys #6468 fallback. It now expects RANK and DENSE_RANK limits over array(f), array(d) and a struct key to run natively with both zeros as peers, and to fall back in strict mode, where these Parquet columns can hold a null.
• CometExpressionSuite: the two tests that asserted the strict-mode fallback for array and struct sort keys still do, now with the new reason, because the keys they generate are nullable. They also check that opting in with SortOrder.allowIncompatible=true keeps the sort native and matching Spark, over data written to Parquet with a unique id sorted last. A new test sorts on array(d) and named_struct('v', d) over a local relation, where d is not nullable, so strict mode keeps the sort native, and checks that NaN payloads and zero signs come back unchanged, with the zeros and the two NaNs as peers.
• Results on macOS aarch64 with the default Spark 4.1 profile:
  • CometSqlFileTestSuite and CometWindowExecSuite together: 661, including both new fixtures.
  • CometExecSuite and the floating-point tests of CometShuffleSuite and DisableAQECometShuffleSuite: 154. CometExpressionSuite floating-point tests: 26.
  • On Spark 3.5, the two fixtures, the three CometExpressionSuite nested tests and all of CometWindowExecSuite: 73. The two fixtures and those three tests also pass on Spark 3.4 and 4.2.
  • The scalafix check on Spark 3.5 and Scala 2.12.
  • Unit tests in core: 579.

[0lai0]

Approve. The nested float normalization is inserted once in create_normalized_key_expr, so Sort, TopK, Window, and WindowGroupLimit pick it up together, and CometSortOrder is only a serializer. I did not run the Spark SQL suites or windows/nested_float_order_keys.sql.

[sunchao]

Summary

• Prior state and problem: Nested floating-point sort keys used Arrow’s raw ordering, splitting Spark-equivalent zeros and NaNs. Strict mode protected these keys through fallback.
• Design approach: Reuse NormalizeNestedFloats through the shared comparison-key planner.
• Correctness / compatibility analysis: Float normalization matches the reviewed Spark ordering sources across supported versions. Two P2 issues remain: removing strict-mode protection exposes incorrect nested-null results, and the new rank-limit tests conflict with the retained JVM fallback guard.
• Key design decisions: Sharing normalization across operators keeps the implementation small and preserves returned values. Native range partitioning remains restricted to scalar keys.
• Implementation sketch: Wrap float-bearing nested comparison keys, remove CometSortOrder incompatibility reporting, and extend native, Scala and SQL coverage.
• Behavioral changes worth calling out: Nested floating-point sorts and windows now stay native under strict mode. Normalization rebuilds float-bearing buffers while retaining float-free subtrees. No performance benchmark was run.
• Suggested improvements: Preserve protection for affected nullable nested keys and reconcile the WindowGroupLimit guard with the new native-execution assertions.

Reviewed the complete ten-file diff from 62ed9d16d547fed1ac25260a1910893f076ab638 to e9578ca54a04b4fb3cf6e531a3beaa5bd1e179ff, including surrounding code and existing discussion. The PR is non-draft. Routed skills: review-comet-pr, review-comet-expression-pr, and review-comet-shuffle-pr.

Exact-head CI: 32 checks succeeded, 15 were skipped, and two failed. The expressions job fails both configurations of the new SQL fixture, causing Required Checks to fail. Native tests reported 1,907 passes, including the extended peer test. All nine Spark 4.1 SQL matrix rows passed. macOS and other Spark SQL profiles were skipped.

Validation limits: Rebuilt and ran a bounded native reproduction using this checkout’s normalization source and the locked Arrow 59.3.0/DataFusion 55.1.0 versions. Ran matching Spark 3.5.9 reference queries and checked ordering sources for Spark 3.4.3, 3.5.9, 4.0.4, 4.1.3 and 4.2.0. Full Comet integration suites were not rerun locally. Exact-head CI supplied that integration evidence.

[sunchao]

[P2] Retain strict-mode protection for nullable nested keys until their null semantics match Spark. Removing getSupportLevel now admits these keys with spark.comet.exec.strictFloatingPoint=true and spark.comet.expression.SortOrder.allowIncompatible=false. For Parquet t(id INT, d DOUBLE) containing (1,1.0),(2,NULL),(3,3.0), with one shuffle partition, ORDER BY array(d) ASC NULLS LAST, id produces IDs [1,3,2] instead of Spark’s [2,1,3]. SUM(id) OVER (ORDER BY array(d)) produces 6 for every row instead of (id,running) = [(1,3),(2,2),(3,6)]. These silently incorrect results were previously avoided by the strict-mode fallback. The underlying bugs are tracked in #6476/#6477, but exposing previously protected floating-point queries is introduced here. Preserve a guard for affected shapes or fix those comparisons before declaring every key compatible.

Evidence: A freshly compiled probe included this exact checkout’s normalize.rs and used locked Arrow 59.3.0/DataFusion 55.1.0. After normalization, array and struct sorting returned [1,3,2] for ASC NULLS LAST, and the array RANGE window returned sums [6,6,6]. Spark 3.5.9 reference queries returned [2,1,3] and sums [3,2,6] by ID. Base CometSortOrder.getSupportLevel returns Incompatible for these nested DOUBLE keys in strict mode, and CometWindowExec requires successful sort-order serialization.

[andygrove]

Fixed in 6471565. You're right that taking out getSupportLevel exposed these keys to #6476 and #6477 in strict mode. CometSortOrder now returns Incompatible in strict mode for a key that has a float and whose type can hold a null element or field, so array(d) over a nullable column falls back again unless SortOrder.allowIncompatible is set. A key whose type cannot hold a null, such as array(coalesce(d, 0.0D)), stays native, and so does a scalar key. I went by the type rather than the null order because the window case in #6477 is wrong with the default order too.

The new nested_float_order_keys_strict.sql runs your ORDER BY array(d) NULLS LAST and SUM(id) OVER (ORDER BY array(d)) over a table with a null element in strict mode, and checks that both fall back and match Spark, and that the same shapes over coalesce keys stay native. With the guard removed its first query returns the wrong order.

[sunchao]

[P2] Reconcile the JVM rank-limit guard with these native-execution assertions. This plain query requires an entirely native plan, but Spark’s default rank-limit optimization inserts WindowGroupLimit, whose converter still unconditionally rejects nested floating-point order keys for RANK and DENSE_RANK. Consequently, the new fixture fails under both strict-mode settings and blocks required CI. The Rust peer test bypasses this converter, so its success does not establish end-to-end native support. Update the guard and its associated fallback coverage to reflect the newly normalized keys before requiring native execution here.

Evidence: Exact-head run https://github.com/apache/datafusion-comet/actions/runs/36813076101/job/110214256198 reports both nested_float_order_keys.sql configurations failing at line 100 with Expected only Comet native operators, but found Project. Both partial and final WindowGroupLimit nodes report RANK and DENSE_RANK compare nested floating-point values exactly, so they fall back. The responsible guard remains in CometWindowGroupLimitExec.scala:108-127. Focused suite command: ./mvnw test -Dtest=none -Dsuites="org.apache.comet.CometSqlFileTestSuite nested_float_order_keys" after the documented native build.

[andygrove]

Fixed in 6471565. The guard is from #6468, and its reason is gone: the native planner normalizes a nested float order key before WindowGroupLimit compares peers, so I removed it from CometWindowGroupLimitExec. The nested float rank-limit test in CometWindowExecSuite expected the fallback, so it now expects RANK and DENSE_RANK limits over array(f), array(d) and a struct key to run natively with both zeros kept as peers. It also checks that strict mode still falls back for those keys, because these Parquet columns can hold a null. With the guard put back, the fixture fails at the rank-limit query, as it did in CI.

[sunchao]

Summary

• Prior state and problem: Nested floating-point keys split Spark-equivalent zeros and NaNs. Strict mode avoided these discrepancies through fallback.
• Design approach: Reuse NormalizeNestedFloats in the shared comparison-key planner.
• Correctness / compatibility analysis: Normalization matches Spark’s float ordering. One new P2 remains: strict mode now admits null-free, deeply nested keys that fail in native RANGE windows. The two previously reported P2 concerns are fixed.
• Key design decisions: Sharing normalization across operators keeps the implementation small and preserves returned values. Nullable nested float keys retain strict-mode fallback. Native range partitioning remains scalar-only.
• Implementation sketch: Wrap nested comparison keys, recursively check nested nullability, remove the obsolete rank-limit guard, and update tests and documentation.
• Behavioral changes worth calling out: Null-free nested float keys now stay native in strict mode. Normalization rebuilds float-bearing buffers and reuses float-free subtrees. No performance benchmark was run.
• Suggested improvements: Preserve fallback for unsupported nested RANGE comparisons, or implement compatible comparisons, and add regression coverage for the finding below.

Reviewed the complete 13-file diff from 75d7c7afda78cef4894dc7a183d810c7be463aa0 to d0612c90438b460aeb7df6c8a25f7d55fbfb377f. The PR remains non-draft. Routed skills: review-comet-pr, review-comet-expression-pr, and review-comet-shuffle-pr.

Exact-head CI: 35 checks succeeded and 15 were skipped. Required Checks, all nine Spark 4.1 SQL rows, the new SQL/Scala tests, and the native peer test passed. Native CI reported 1,931 passing tests. macOS and other Spark SQL profiles were skipped. Run: https://github.com/apache/datafusion-comet/actions/runs/36918130861.

Validation: Freshly compiled bounded probes using this checkout’s normalization and rank-limit sources with locked Arrow 59.3.0/DataFusion 55.1.0. All 144 sort/rank checks passed, including output-bit preservation. Direct native window probes reproduced the P2 below. Spark 3.5.9 reference queries passed. Reviewed ordering sources across Spark 3.4.3, 3.5.9, 4.0.4, 4.1.3 and 4.2.0. Full local Comet integration builds were not rerun, and the new failure was reproduced through the native pipeline rather than end-to-end JVM execution.

Recommendation: request changes for the new P2.

[sunchao]

[P2] Preserve fallback for deeply nested keys in RANGE windows. With spark.comet.exec.strictFloatingPoint=true and spark.comet.expression.SortOrder.allowIncompatible=false, a Parquet table t(id INT, d DOUBLE) containing (1,1.0),(2,3.0) exposes this through SELECT id, COUNT(id) OVER (ORDER BY array(named_struct('x', coalesce(d, 0.0D))), id) AS running FROM t ORDER BY id. Spark returns [(1,1),(2,2)], but the native window fails with Uncomparable values: List([{x: 1.0}]), List([{x: 1.0}]). This key cannot contain nested nulls, so the new branch declares it compatible. DataFusion’s frame comparator still cannot compare arrays of structs, even after float normalization. The base strict-mode guard kept this valid query on Spark. Could we add a window-specific fallback for these unsupported comparison shapes, or fix their frame comparisons before admitting them?

Evidence: Freshly compiled /tmp/review6475-active-u51_pmld/physical_range.rs includes this head’s normalize.rs and constructs normalized SortExec keys followed by BoundedWindowAggExec with InputOrderMode::Sorted, matching the current planner. With locked Arrow 59.3.0/DataFusion 55.1.0, the null-free array-of-struct RANGE case fails as quoted. Struct-of-array and array-of-array keys also fail. All corresponding ROWS cases and the flat array RANGE control pass. A fresh Spark 3.5.9 run returned [(1,1),(2,2)] and confirmed containsNull=false and struct-field nullable=false. Base CometSortOrder.getSupportLevel rejects this nested DOUBLE key in strict mode. Current CometWindowExec accepts its serialized order, and DataFusion’s ScalarValue::partial_cmp_list returns None when the underlying Arrow comparison rejects struct elements.

[sunchao]

Summary

• Prior state and problem: Nested floating-point ordering split Spark-equivalent zeros and NaNs. Strict mode protected these keys through fallback.
• Design approach: Reuse NormalizeNestedFloats in the shared comparison-key planner.
• Correctness / compatibility analysis: Float normalization matches Spark’s ordering across supported versions. The earlier nullable-key and rank-limit concerns are fixed. The previously reported P2 involving deeply nested RANGE keys remains unresolved.
• Key design decisions: Normalize comparison keys while preserving returned values. Keep nullable nested float keys behind strict-mode fallback and native range partitioning restricted to scalar keys.
• Implementation sketch: Wrap nested comparison keys, recursively inspect nested nullability, remove the obsolete rank-limit guard, and update tests and documentation.
• Behavioral changes worth calling out: Null-free nested float keys now stay native in strict mode. Normalization rebuilds floating-point buffers and reuses float-free subtrees. No performance benchmark was run.
• Suggested improvements: Address the existing RANGE comparison blocker by preserving fallback for unsupported shapes or implementing compatible frame comparisons, with regression coverage.

Reviewed the entire 14-file diff from 33ea8c0dbebcc86abd099ab844c323d260908e4a to 6fa5836c39fba34985f0d3425305e595318fd2c2. The PR remains non-draft. Routed skills: review-comet-pr, review-comet-expression-pr, and review-comet-shuffle-pr.

The existing P2 remains at spark/src/main/scala/org/apache/comet/serde/CometSortOrder.scala:68. With strict mode enabled and incompatible sort orders disabled, COUNT(id) OVER (ORDER BY array(named_struct('x', coalesce(d, 0.0D))), id) over (1,1.0),(2,3.0) is admitted natively. Spark returns counts [1,2], but native execution fails with Uncomparable values: List([{x: 1.0}]), List([{x: 1.0}]). A freshly compiled probe using this head’s normalization source and normalized SortExec followed by BoundedWindowAggExec reproduced the failure. Struct-of-array and array-of-array keys also fail. This existing concern is not duplicated as a new finding.

No additional introduced P1/P2 issues found within this review.

Exact-head CI: 7 checks succeeded, 13 were skipped, and 12 remained in progress, with no reported failures. Native builds/tests and the labeled Spark 4.1 SQL build were still running. The required-check verdict remains pending: https://github.com/apache/datafusion-comet/actions/runs/36997041707.

Validation: Fresh bounded probes with locked Arrow 59.3.0/DataFusion 55.1.0 passed 144 sort/rank checks. Corresponding ROWS windows and the flat-array RANGE control passed. Spark 3.5.9 reference queries confirmed the expected counts and non-nullable nested schema. Reviewed ordering sources for Spark 3.4.3, 3.5.9, 4.0.4, 4.1.3 and 4.2.0. Full local Comet integration builds and suites were not rerun, so the failure was validated through the native pipeline rather than end-to-end JVM execution.

[sunchao]

Summary

• Prior state and problem: Raw Arrow ordering for nested floats split Spark-equivalent zeros and NaNs, affecting sorting, window peers and rank limits.
• Design approach: Reuse NormalizeNestedFloats in the shared comparison-key planner for Sort, TopK, Window and WindowGroupLimit.
• Correctness / compatibility analysis: Normalization agrees with the reviewed Spark ordering sources across supported versions. The three previously reported P2 concerns are addressed: nullable keys retain strict-mode protection, the obsolete rank-limit guard is removed, and unsupported nested RANGE comparisons now fall back.
• Key design decisions: Normalize comparison keys while preserving returned values. Keep the new frame restriction specific to affected RANGE windows, preserving ROWS, fully unbounded frames and CUME_DIST. Native range partitioning remains scalar-only.
• Implementation sketch: Add the shared normalization wrapper, recursive nullability and frame-comparability checks, regression fixtures, and corresponding documentation updates. Reusing the existing normalizer keeps the implementation small.
• Behavioral changes worth calling out: Eligible nested floating-point keys now remain native in strict mode. Normalization rebuilds floating-point buffers and reuses float-free subtrees. No performance benchmark was run.
• Suggested improvements: No additional P1/P2 changes requested.

Reviewed the entire 16-file diff from 33ea8c0dbebcc86abd099ab844c323d260908e4a to 4cf0b1cfdf2cee68b6ad75df73d9b1666d079169. The PR remains non-draft. Routed skills: review-comet-pr, review-comet-expression-pr, and review-comet-shuffle-pr.

No introduced P1/P2 issues found within this review. No substantiated existing P1/P2 blockers remain in the reviewed code.

Exact-head CI: 17 checks succeeded, 13 were skipped, and 8 remained in progress, with no reported failures. The native build passed. Comet integration tests, Rust tests and the labeled Spark 4.1 SQL build were still running. The required-check verdict remains pending: https://github.com/apache/datafusion-comet/actions/runs/37000810143.

Validation: Freshly compiled probes using this checkout’s normalization and rank-limit sources with locked Arrow 59.3.0/DataFusion 55.1.0 passed 144 sort/rank checks, including output preservation. Window probes confirmed the failures excluded by the new guard and successful retained frame and CUME_DIST cases. Spark 3.5.9 reference queries matched expectations. Reviewed relevant sources for Spark 3.4.3, 3.5.9, 4.0.4, 4.1.3 and 4.2.0. Full Comet/JVM integration suites were not rerun locally, so end-to-end validation of the new fallback remains dependent on pending CI.

Review state: Comment pending CI.