fix: match Spark ordering in native array extrema - #5403
Conversation
| (array(0.0, double('-0.0'), 1.0)) | ||
|
|
||
| query ignore(array_min signed-zero: Spark +0.0, Comet -0.0) | ||
| query ignore(https://github.com/apache/datafusion-comet/issues/5401) |
There was a problem hiding this comment.
is it any chance to use query expect_fallback so once this fixed we would know the test should be addressed as well, instead of ignoring it
There was a problem hiding this comment.
For cases where fallback is expected, I agree that query expect_fallback(...) is a better regression check than ignore.
I've reworked array_min and array_max locally to follow Spark's ordering natively. In that revision, these signed-zero cases use plain query assertions, so they must execute natively and match Spark in both strict and non-strict modes. The separate non-default-collation tests use query expect_fallback(...), since those cases still require Spark by default.
The rework is ready locally but hasn't been pushed yet, so the current PR diff still contains the old ignore. I'll update the branch once I can publish it.
Which issue does this PR close?
Closes #5401.
Why are the changes needed?
array_minandarray_maxshould select the same element in Spark and Comet. Spark considers positive and negative zero equal and keeps the first equal element. The existing native implementation instead orders negative zero before positive zero, so it can return a different value:ARRAY<DOUBLE>columna[+0.0, -0.0]array_min(a)+0.0-0.0[-0.0, +0.0]array_max(a)-0.0+0.0The sign is observable, for example when the result is converted to a string. These examples describe floating-point values read from a column, not decimal literals that Spark might fold before native execution.
For nested values, the difference can change which whole element is selected, not just the zero sign. Suppose an array of
(key: DOUBLE, payload: INT)structs contains[(-0.0, 2), (+0.0, 1)]. Spark'sarray_maxtreats the keys as equal, compares the payloads, and returns(-0.0, 2). Ordering the zero signs differently selects(+0.0, 1)instead. Nested comparisons also need Spark's nulls-first ordering, which differs from DataFusion's generic scalar comparison.The earlier version of this PR protected strict mode by routing these expressions through Spark-generated JVM code. That left the default native behavior unchanged. This revision fixes native evaluation itself, so correctness does not depend on enabling strict mode.
What changes were proposed in this PR?
Comet now provides Spark-compatible native evaluation for floating-point and nested array extrema. Signed zeros compare equal, all NaN representations compare equal and greater than numbers, and ties keep the first element. Comparison follows the same rules recursively through arrays and structs, including lexicographic and nulls-first ordering. The result preserves the original winning value, including its fields, zero sign, and NaN bits. Ordinary scalar element types continue to use their existing DataFusion implementation.
Both functions remain native in strict and non-strict modes, without the JVM codegen dispatcher or an incompatible-execution opt-in. Non-
UTF8_BINARYstring collations remain a separate compatibility boundary: they fall back to Spark by default, including when the collated string is nested inside an array or struct. That remaining work is tracked in #4496.For nested results, the output path copies the selected elements and releases unused result-buffer capacity left by Arrow's flat-list copy. This addresses cases where small winners otherwise carry buffers sized for much larger losing candidates into downstream operators. The changes are local to array extrema and use existing Arrow facilities; temporary allocation remains, and buffer shrinking is best-effort.
Previously ignored signed-zero regressions become ordinary native assertions, with additional tests for nested ordering, original-value preservation, and result-buffer retention. The compatibility documentation now describes native floating-point support and the remaining collation fallback.
How was this PR tested?
Checks on the current code
cargo fmt --all -- --checkandgit diff --checkalso passed.The native tests compare raw floating-point bits, not just numeric equality. They cover both widths, all 2,197 triples of a special-value set per width, first ties across long-array boundaries, null and empty inputs, sliced buffers, scalar and array inputs, dictionaries, and nested arrays and structs. The allocation regressions cover sparse nested results and small winners alongside large losing values.
In the 8,192-row small-winner probe, the final capacity fix reduced retained child-buffer capacity from 64 MiB to 0 bytes for empty-list winners, and from 64 MiB to 64 KiB for one-
DOUBLEwinners. Both previously failing Comet sorts then passed under the 32 MiB pool. These numbers measure retained result buffers, not peak process memory; the temporary allocation is not eliminated.Performance checks on the current code
Criterion compares Comet with the pinned DataFusion 54.1.0 implementation on ordinary finite values where the results agree, checking equality before timing. These are native-function microbenchmarks, not end-to-end Spark query measurements.
The correctness fix has a measurable cost for long, null-free
FLOATarrays: about 1.5–1.6× DataFusion's latency, confirmed in two current-source runs over a batch of 64 arrays of length 1,024. This cost was already present in the original native rewrite; it is not introduced by the result-buffer follow-ups.array_minarray_maxThe latest capacity-release change was also checked with 16 nested timed cases, repeated twice: four shapes, both operations, and both implementations. Comet's median changes ranged from 6.66% lower to 2.89% higher latency relative to the version immediately before that change. Those fixtures do not characterize deep nesting or long equal prefixes. This PR prioritizes Spark-compatible native results and does not claim an across-the-board speedup.
Earlier validation, before the two allocation follow-ups
The original native rewrite passed the following Spark checks. The Spark tests and full benchmark matrix were not rerun after the two result-copying and capacity changes; validation of the current code is listed above.
The floating-point SQL fixtures compare native execution directly with Spark, with the JVM dispatcher and incompatible-execution opt-in disabled, across strict-mode and Parquet-writer dictionary settings. The collation fixtures separately check the fallback reason with the dispatcher both enabled and disabled. Counts exclude neighboring fixtures and version-gated no-op cases selected by substring matching.
The runtime-NaN test creates negative NaNs after the Parquet scan because Parquet canonicalizes stored NaNs. The dictionary configuration matrix varies the writer setting; it does not establish that these small fixtures contain dictionary-encoded pages. Rust tests separately exercise native dictionary arrays. Generated compatibility documentation was verified on both Spark profiles, and
make formatpassed with Spark 3.4 at that stage.The earlier full Criterion matrix contained 104 timed cases: 26 shapes across both functions and implementations, covering
FLOAT,DOUBLE, anINTcontrol, and nestedARRAY<DOUBLE>values with short/long arrays and different null densities. It showed roughly 1.2× latency for long, sparse-nullFLOATarrays and about 10% higher latency for some short, nullableDOUBLEcases, alongside comparable or lower latency for most other measured shapes. Those are historical observations from the original native rewrite, not measurements of the final copying changes.Validation is focused on these expressions and native result handling; the full Comet test suite and end-to-end Spark workloads were not run.