fix: match Spark ordering in native array extrema

[sunchao]

Which issue does this PR close?

Closes #5401.

Why are the changes needed?

array_min and array_max should 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:

Value of an ARRAY<DOUBLE> column a	Expression	Spark	Native before this change
[+0.0, -0.0]	array_min(a)	+0.0	-0.0
[-0.0, +0.0]	array_max(a)	-0.0	+0.0

The 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's array_max treats 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_BINARY string 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?

After merging Apache main 525c05b52 in 07c8de2f7, 684 Rust expression-library tests and 35 common-library tests passed, including all 24 array-extrema regressions, from the merged source. cargo fmt --all -- --check, the floating-point guide's Prettier check, and git diff --check for the PR delta against that main commit passed.

The merge retains both guide sections and both sets of benchmark registrations. The array-extrema implementation, its native and Spark tests, and its Spark admission code are unchanged from the preceding PR head. The performance measurements and broader validation below are retained from before this merge; optimized benchmarks, the full native workspace, and Spark suites were not rerun for this merge.

Validation before merging current main

• 662 Rust expression-library tests passed, including all 24 array-extrema tests. The 24 extrema tests also passed in an optimized build.
• Eight constrained-sort checks passed with a 32 MiB memory pool: two nested-data shapes, both extrema operations, and both Comet and DataFusion implementations. Selected values matched the reference results.
• Workspace-wide Clippy passed for all targets with warnings denied; cargo fmt --all -- --check and git diff --check also 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-DOUBLE winners. 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 measurements before merging current main

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 FLOAT arrays: 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.

Operation	Run 1: Comet / DataFusion median latency	Run 2: Comet / DataFusion median latency
array_min	1.625×	1.615×
array_max	1.494×	1.556×

The 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.

Spark version	Directly changed SQL file/configuration cases	Runtime NaN raw-bit test
3.4.3	10 passed; collation cases require Spark 4+	Passed
4.1.3	18 passed, including collation fallbacks	Passed

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 format passed 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, an INT control, and nested ARRAY<DOUBLE> values with short/long arrays and different null densities. It showed roughly 1.2× latency for long, sparse-null FLOAT arrays and about 10% higher latency for some short, nullable DOUBLE cases, 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.

[comphead]

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

[sunchao]

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.

[comphead]

thanks @sunchao I'm spinning up the Claude to make a first round of review, considering the PR size

[andygrove]

Note on this review: this was generated by an LLM (Claude Code) at my request while I worked through a review backlog. I have not verified the individual findings myself. Please treat everything below as suggestions to evaluate rather than as authoritative review feedback, and push back on anything that is wrong or already handled.

This is a good piece of work. Fixing native evaluation rather than routing through the codegen dispatcher is the right call, and the recursive spark_comparator with nulls-first ordering, dictionary decoding, and struct lexicographic comparison looks careful. I particularly appreciate that the description reports the slowdown honestly instead of burying it, and that the tests compare raw bits rather than numeric equality.

The float slowdown is the thing I want to talk about.

1.5x to 1.6x on plain float arrays

float_extrema gives up Arrow's vectorized min/max and does a scalar loop with a per-element null check and two branches. That is a real cost paid by every user on every array_min and array_max over a float column, in order to fix a case that only arises when the array contains both zero signs or multiple NaN encodings.

There is a fast path available here. Arrow's kernel and Spark disagree only when the winning value is a zero or a NaN. For any other winner, all tied candidates are bit-identical, so "keep the first" and "keep whichever Arrow found" produce the same bytes. So you could run Arrow's kernel first, and only when the result is 0.0, -0.0, or NaN fall back to the scalar first-occurrence scan for that one list. Arrays with no zeros and no NaNs, which is essentially all real data, would then run at DataFusion's speed.

Would that work? If there is a reason it does not, it would be good to say so in the code, because a permanent 1.6x on the common path to fix an edge case is a hard trade to justify otherwise.

FixedSizeList input reaches an exec_err

needs_spark_ordering returns true for FixedSizeList as an element type, and spark_comparator handles it. But invoke_with_args matches only DataType::List and DataType::LargeList on the outer array and returns exec_err!("{} does not support type {other}") for anything else. So array_max over a FixedSizeList column would now fail at runtime rather than falling back.

I could not find a path where Comet produces a FixedSizeList here, so this may well be unreachable. If it is, could the match arm say so, along the lines of "Comet only produces List and LargeList; FixedSizeList is unreachable"? And if it is reachable, this needs either handling or a plan-time Unsupported.

Two smaller things

spark_comparator's Dictionary arm calls cast(array, value_type), which materializes the whole decoded values array on every invocation. For a heavily dictionary-encoded column that is a full decode per batch. Is that acceptable, or worth a note that dictionary elements are the slow path?

The Cargo.toml change adds a trailing newline to an unrelated line. Not a problem, just noting it so it does not look accidental in the diff.

[sunchao]

Updated in 9cd863dfa.

I checked the proposed Arrow-first fast path against the pinned Arrow/DataFusion implementation. It is not safe to trigger the corrective scan only when Arrow returns zero or NaN: for array_max([-NaN, 1.0]), Arrow returns 1.0, while Spark returns NaN. The mismatch reproduces for FLOAT and DOUBLE with list lengths 2, 32, and 1,024. I added that counterexample beside float_extrema so a future optimization does not reintroduce it. A safe vectorized path would also have to establish that the input contains no exceptional values, and would need to be measured across null densities and lengths. I have not substituted an unmeasured two-pass algorithm here.

The reported 1.5–1.6× latency applies to the measured long, null-free FLOAT microbenchmark, not every floating array or an end-to-end Spark query. That cost remains explicit in the description; this clarification does not claim a performance improvement.

The FixedSizeList concern does not reach the fallback arm through the supported function signature: invoke_with_args first calls the delegated DataFusion return_type, which rejects an outer fixed-size list. Supporting fixed-size lists as nested elements is a different capability. The code now states that boundary rather than assuming that Comet can never construct such an Arrow array.

Dictionary values are decoded once when the comparator is constructed for a child array/batch, and the comparator retains that decoded array for its index comparisons. This is not a decode per element comparison; I added that scope to the comment. I have not claimed the decode is free or that deep dictionary nesting was benchmarked.

The earlier ignored-case requests are already implemented as native assertions, with separate collation-fallback checks. The final newline in Cargo.toml has no behavioral effect. This update only changes explanatory source comments: runtime code, tests, and the previously reported performance results are unchanged. cargo fmt --all --check and git diff --check pass; no new full Spark build is claimed.

[sunchao]

Updated in 07c8de2f7.

Merged Apache main 525c05b52 in 07c8de2f7 to resolve the branch conflicts without rewriting its history. The resolution retains both the array-extrema and upstream signed-zero guide sections, plus all benchmark registrations from both sides. The extrema implementation, native/Spark regressions, and admission code remain unchanged.

The merged source passes 684 expression-library and 35 common-library unit tests, including all 24 extrema tests. Rust formatting, the guide's Prettier check, and the PR diff check against the merged main commit also pass. The description now distinguishes this fresh validation from the earlier optimized benchmarks, full-workspace checks, and Spark results, which were not rerun for this merge.

[sunchao]

The macOS Spark 4.0 scans job at 07c8de2f7 crashed during native thread cleanup, just after native scan on fake fs passed. This was a JVM SIGBUS, not a ScalaTest assertion failure in the extrema changes.

I downloaded the exact native artifact and verified its ZIP digest against the job log. Symbolication places the saved return address at hdfsThreadDestructor + 0x50, matching the prior crash in #5023 and the libhdfs fix in PR #5036. The HDFS/JNI/storage code, fake-filesystem suite and dependency lockfile are unchanged from merged main 525c05b52. This identifies the known libhdfs teardown failure; it does not make the failed job a passing result. The parent workflow is still active, and I have not claimed a clean macOS rerun.