Skip to content

fix: normalize scalar float sort and window rank keys - #5469

Open
sunchao wants to merge 2 commits into
apache:mainfrom
sunchao:dev/chao/codex/fix-window-group-limit-float-ordering
Open

fix: normalize scalar float sort and window rank keys#5469
sunchao wants to merge 2 commits into
apache:mainfrom
sunchao:dev/chao/codex/fix-window-group-limit-float-ordering

Conversation

@sunchao

@sunchao sunchao commented Aug 26, 2026

Copy link
Copy Markdown
Member

Why are the changes needed?

A filter such as RANK() <= 1 must keep every row tied for first place. Native execution can currently drop some of those rows when the ordering column contains signed zeros or different NaN representations. This PR fixes that result mismatch for scalar FLOAT and DOUBLE keys, following the native WindowGroupLimitExec support added in #4870.

A rank cutoff can discard a tied row

Consider a table measurements whose DOUBLE column v contains -0.0, +0.0, and 1.0:

SELECT v, rnk
FROM (
  SELECT v, RANK() OVER (ORDER BY v ASC) AS rnk
  FROM measurements
) ranked
WHERE rnk <= 1;

Spark considers the two zeros equal, so both belong at rank 1. The native comparison instead distinguishes their IEEE-754 representations:

Input value Rank in Spark Rank with the previous native comparison
-0.0 1 1
+0.0 1 2
1.0 3 3

The query therefore keeps both zero rows in Spark, but only the negative-zero row before this fix. DENSE_RANK has the same problem at the cutoff. Once WindowGroupLimitExec has discarded a qualifying row, the window calculation above it cannot recover that row.

NaNs expose the same mismatch. Spark treats every NaN representation as equal and greater than every non-NaN value. For input containing two differently encoded NaNs and 1.0, ORDER BY v DESC followed by rnk <= 1 should keep both NaNs. Arrow's raw total ordering can separate them; a NaN with its sign bit set can even sort below finite values. The regressions in this PR construct distinct NaN encodings directly so the test actually exercises this distinction.

What changes were proposed in this PR?

The fix gives the native sorting and window operators a common representation for comparing scalar floating values. It reuses Comet's existing NormalizeNaNAndZero expression to map every NaN to one canonical NaN and both zero signs to positive zero in the comparison keys. Arrow can then compare those keys using its existing machinery while agreeing with Spark about which values are tied.

This normalization must happen before sorting as well as when assigning ranks. For example, with ORDER BY v, secondary, the rows (-0.0, 1) and (+0.0, 1) are peers and must precede the rows whose secondary key is 2. A sort that distinguishes the zero signs can instead produce:

(-0.0, 1), (-0.0, 2), (+0.0, 1), (+0.0, 2)

Even a corrected peer comparison would see the first peer group split apart. A streaming rank limit could reach the cutoff at (-0.0, 2) and stop before seeing the other qualifying row. The shared comparison-key construction therefore supplies normalized keys to Sort, Window, and WindowGroupLimit, keeping their ordering and peer decisions consistent.

That consistency includes PARTITION BY, not just ORDER BY. An ordinary query such as RANK() OVER (PARTITION BY p ORDER BY id) must still work when p is a DOUBLE column containing only 1.0 and 2.0. Spark may already have normalized p; wrapping it again in Sort while leaving Window unchanged produces different expressions for the same key. DataFusion uses those expressions to recognize partition ordering, so the mismatch causes execution to fail even though the values compare equally. The planner therefore reuses already-normalized expressions and constructs matching keys for Sort, Window, and WindowGroupLimit. This preserves ordinary windows as well as the rank-limit path.

Native range partitioning uses the same rule for its sampled boundaries. An incoming positive zero and a boundary containing negative zero must compare equal, just as they do during sorting. Normalizing both sides keeps the shuffle's comparisons consistent with the operators that consume its output.

The original input values are preserved: selecting v still returns its original zero sign or NaN payload. Only the temporary keys used for comparison are normalized. The scope remains scalar FLOAT and DOUBLE; the documented limitations for floats nested in arrays or structs and the existing spark.comet.exec.strictFloatingPoint fallback policy are unchanged.

How was this PR tested?

The results below are local validation of the pending revision b07d58197; they are not CI results for the published head.

The regressions verify both the returned rows and the execution path. The Spark cutoff tests require native Sort and WindowGroupLimit operators, compare results with Spark, and check that the input and output retain their original floating bits. Native tests cover compound keys, peer groups spanning batches, and agreement between range boundaries and incoming keys. The previously ignored signed-zero SQL regression is also enabled.

Additional regressions protect ordinary windows partitioned by floating columns: RANK, PERCENT_RANK, and NTILE must match Spark for single and compound partitions. A filtered RANK query also checks the Sort → WindowGroupLimit → Window path. A native planner test verifies that both bare and already-normalized partition keys retain recognizable ordering metadata.

With the comparison-key fix removed, the native peer/range regressions and both Spark FLOAT/DOUBLE cutoff regressions fail. In the Spark fixture, unpatched Comet returns 4 rows where Spark returns 8. The fixed runs pass.

On Spark 4.1, the partition-ordering cases were also checked against the library before the key-consistency fix: all six ordinary-window cases failed, while the two WindowGroupLimit controls passed. All eight pass with the fix.

The final review reran all three native planner regressions and the eight Spark 4.1 partition-key regressions; all passed. It also checked the three query shapes behind the Spark SQL CI failures on the earlier revision: a count window partitioned by floating columns, a rank window with computed aggregate/partition keys, and a correlated EXISTS ... LIMIT 1 query that Spark rewrites into a window. Each query matched Spark with native Sort and Window execution asserted. The same probes passed with the pinned base library. These were focused query probes, not reruns of the full CI shards.

Local validation Result
Native workspace unit tests 909 passed; 4 existing tests ignored
Spark 3.4: floating partition-key regressions 6 passed; 2 WindowGroupLimit cases skipped because Spark 3.4 has no such operator
Spark 3.5: floating partition-key regressions 8 passed
Spark 4.0: floating partition-key regressions 8 passed
Spark 4.1: full window suite, all WindowGroupLimit SQL cases, sort fallback tests, and native range shuffle 87 passed

make core, cargo fmt --all -- --check, and cargo clippy --all-targets --workspace -- -D warnings passed. JVM formatting checks and Spark 4.1 Scala style checks also passed.

Which issue does this PR close?

Closes #5468.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

Scalar NaN and signed-zero sort keys split WindowGroupLimit rank peers

1 participant