fix: normalize scalar float sort and window rank keys - #5469
Open
sunchao wants to merge 2 commits into
Open
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Why are the changes needed?
A filter such as
RANK() <= 1must 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 scalarFLOATandDOUBLEkeys, following the nativeWindowGroupLimitExecsupport added in #4870.A rank cutoff can discard a tied row
Consider a table
measurementswhoseDOUBLEcolumnvcontains-0.0,+0.0, and1.0:Spark considers the two zeros equal, so both belong at rank 1. The native comparison instead distinguishes their IEEE-754 representations:
-0.0+0.01.0The query therefore keeps both zero rows in Spark, but only the negative-zero row before this fix.
DENSE_RANKhas the same problem at the cutoff. OnceWindowGroupLimitExechas 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 DESCfollowed byrnk <= 1should 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
NormalizeNaNAndZeroexpression 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 is2. A sort that distinguishes the zero signs can instead produce: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 justORDER BY. An ordinary query such asRANK() OVER (PARTITION BY p ORDER BY id)must still work whenpis aDOUBLEcolumn containing only1.0and2.0. Spark may already have normalizedp; 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
vstill returns its original zero sign or NaN payload. Only the temporary keys used for comparison are normalized. The scope remains scalarFLOATandDOUBLE; the documented limitations for floats nested in arrays or structs and the existingspark.comet.exec.strictFloatingPointfallback 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, andNTILEmust match Spark for single and compound partitions. A filteredRANKquery 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 1query 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.make core,cargo fmt --all -- --check, andcargo clippy --all-targets --workspace -- -D warningspassed. JVM formatting checks and Spark 4.1 Scala style checks also passed.Which issue does this PR close?
Closes #5468.