fix: order nulls inside nested values first in least

[viirya]

Which issue does this PR close?

Closes #6759.

Rationale for this change

Spark orders a null inside a struct or array before every other value, in both greatest and least. Comet sent nested types without a float leaf to DataFusion's least, which orders such a null after every other value (nulls_first: false), so least could return a different value from Spark.

What changes are included in this PR?

SparkGreatestLeast::handles now accepts every nested type, not only those with a float leaf. For those types, SparkGreatestLeast compares with spark_comparator, which already orders nested nulls first. Scalar non-float types still use DataFusion's greatest and least. The greatest and least entries in the math expression audit are updated to match.

SparkGreatestLeast also reduces constant arguments before comparing row by row: all of them for types without a float leaf, and runs of adjacent ones otherwise, keeping the selected constant itself. The newly routed types keep DataFusion's cost when several arguments are constants.

How are these changes tested?

• New SQL file expressions/math/greatest_least_nested_nulls.sql, covering structs, arrays, and arrays of structs with a null nested value. It fails without this change: for id = 3, Spark returns [null, 3] and Comet returned [3, 1].
• The Rust test handles_floats_and_nested_types_only replaces the previous handles test and checks that every nested type is routed to SparkGreatestLeast.
• The existing greatest, least, and float SQL files still pass.

This pull request and its description were written by Isaac.

[sunchao]

Reviewed the full three-file, one-commit diff at 00ec7d2799b46d3a40158b53cfbc1fd355dde7e4 against b56349697b786ff2ad1c1bcf6ecf45b809af5f30. The PR is not a draft. Applied review-comet-pr and review-comet-expression-pr, with the relevant contributor guidance. The snapshot and live discussion checks contained no existing reviews, conversation comments, inline comments, or review threads.

Summary

• Prior state and problem: Nested values without float leaves used DataFusion’s extrema functions. Its least orders nested nulls last, so Comet could return [3, 1] where Spark returns [null, 3].
• Design approach: Expand SparkGreatestLeast::handles to route every nested result type through the existing Spark-compatible comparator.
• Correctness: Compared Spark’s interpreted and generated Least/Greatest implementations across supported 3.4–4.2 versions, including array and struct ordering. The implementation preserves top-level null skipping, nested nulls-first ordering, lexicographic comparison, and first-argument tie handling. Focused tests reproduced the original failure through base routing and passed at the reviewed head.
• Compatibility analysis: The relevant Spark semantics are consistent across the supported versions. Existing recursive collation checks remain in place. No protobuf, configuration, shim, or return-type contract changes were introduced.
• Key design decisions: Keep scalar non-float inputs on DataFusion’s existing path and reuse spark_comparator for nested inputs. Existing floating-point handling remains unchanged.
• Implementation sketch: The production change broadens the routing predicate and updates its explanatory comments. A Rust regression test, updated routing assertions, and three column-dependent SQL queries cover structs, arrays, and arrays of structs.
• Performance: Scalar non-float execution is unaffected. Newly routed nested inputs use the existing argument normalization, comparator, selection bitmap, and Arrow zip pipeline. Performance assessment was source-based. No benchmark was run, and no evidence-backed P1/P2 performance regression was identified.
• Design: Reusing the established comparison path keeps nested ordering consistent without duplicating comparator logic. The distinction between skipping null arguments and ordering null children is clear.
• Abstraction & complexity: No new abstraction or configuration is added. The existing SparkGreatestLeast and spark_comparator interfaces fit the expanded responsibility.
• Behavioral changes worth calling out: Compared with branch-1.1 at e9efd9f764ee0a59b7898ff028d6985d4a7a28e1, this intentionally corrects non-float nested least results. Earlier floating-point corrections already present in the supplied base are separate from this PR.
• Suggested improvements: No additional change meeting the P1/P2 reporting bar was identified. The running CI jobs still need to establish the integration-test verdict.

Validation: A standalone harness compiled the exact-head expression and comparator modules against cached Arrow 59.3.0 and DataFusion 55.1.0, matching Cargo.lock. All 26 source unit tests and three additional review tests passed. Additional checks covered 1,728 list triples, mixed scalar/column arguments, null and empty values, sliced batches, structs, and arrays of structs. The same additional tests failed through base routing because of the known nested-null ordering bug. git diff --check passed and the checkout remained unchanged.

Exact-head CI: Seven checks succeeded, fourteen were skipped, and two remained in progress. No failed checks were reported. The active jobs were the main CI preflight and the labeled Spark 4.1 native/JVM build in runs https://github.com/apache/datafusion-comet/actions/runs/37680375961 and https://github.com/apache/datafusion-comet/actions/runs/37680397841. This is not a completed integration-test verdict. I did not run a full native/Maven build or the Comet SQL suite locally.

No introduced P1/P2 issues found within this review. No substantiated existing blockers remain.

[comphead]

Thanks @viirya. From reading the code, sending every nested type to spark_comparator matches Spark's ordering of a null inside an array or struct, which comes first in both functions. The inline comments cover a stale audit note, a unit test that does not exercise the fix, and two inert configs in the SQL file. Separately, Comet CI at this head shows Preflight cancelled and Required Checks failed, so the Rust tests and the new SQL file have not run in CI for this commit. Could the workflow be re-run?

[comphead]

Could this PR also update the greatest and least entries in docs/source/contributor-guide/expression-audits/math_funcs.md? Lines 152 and 161 say that only arrays and structs with a floating-point leaf use SparkGreatestLeast and that other types use DataFusion's GreatestFunc and LeastFunc. That no longer holds for nested types without a float leaf, and the least entry should also say it orders a null inside an array or struct first, as Spark does. #6457 added those two lines together with the float routing.

[viirya]

Updated both entries in ad5447e.

[comphead]

nested_nulls_sort_first calls invoke_with_args directly, so it never goes through handles, and pick and spark_comparator are the same as on the base. From reading the code I expect it to pass without this change too. I have not run it. Its two structs are the id = 3 row of the first query in greatest_least_nested_nulls.sql, which does fail without the fix. Could we drop it and keep handles_floats_and_nested_types_only as the unit-level guard for the routing?

[viirya]

Agreed. It never reaches handles, so it passes without the fix. Dropped in ad5447e, and handles_floats_and_nested_types_only stays as the unit-level guard for the routing.

[comphead]

With spark.comet.exec.range.enabled=true, CometExecRule turns range(...) into CometRangeExec and only falls back to the Spark-to-Arrow conversion when that operator declines a range (around line 504 of CometExecRule.scala). So the two sparkToColumnar lines have no effect here, and naming Range in spark.comet.sparkToColumnar.supportedOperatorList is deprecated per its doc in CometConf.scala. Could we drop both lines? The file would then fail if native Range ever stopped taking these ranges, rather than quietly running on a converted leaf.

[viirya]

Dropped both lines in ad5447e. Every query in the file still runs natively through CometRangeExec.

[sunchao]

Reviewed commit ad5447e4a09768ffb8a122b7ae0feadf675c5e31. New P1/P2 findings are detailed inline.

[sunchao]

[P2] Could the expanded routing preserve constant reduction for float-free nested arguments? For an ARRAY<INT> column with multiple constant-array arguments, DataFusion’s execute_conditional previously reduced the constants to one before broadcasting. SparkGreatestLeast::invoke_with_args instead broadcasts every constant and performs a full comparison and zip for each. An 8,192-row batch with eight 16-element constant arrays allocated approximately 1.68 MB through base routing versus 12.94 MB for head greatest and 10.18 MB for head least, with identical results. Spark retains these separate constants after optimization, so ordinary queries reach this path. This introduces repeated batch processing and allocation even for already-correct greatest calls. Please retain constant reduction using Spark’s nested ordering for float-free inputs, or preserve the existing route where its semantics already match.

Evidence: Reproduction input: one ListArray containing 8,192 rows, each row repeating row_index % 21 sixteen times, followed by eight ScalarValue::List arguments repeating constants 0 through 7 sixteen times each. Compare DataFusion GreatestFunc/LeastFunc, selected by base routing, with exact-head SparkGreatestLeast. Output arrays match. An instrumented allocator measured cumulative allocated bytes per invocation: greatest 1,675,836 -> 12,936,608, least 1,675,836 -> 10,184,096. With two constants, both increase from 1,674,716 to 2,890,136 bytes. Harness and compile arguments are saved under /tmp/review6761-ad544-session-621i9yw7/. Run perf-tests perf::nested_constant_cost --nocapture --test-threads=1. Spark 3.5.9 confirmed that SELECT greatest(array_repeat(CAST(id % 21 AS INT),16), array_repeat(0,16), array_repeat(1,16), array_repeat(2,16), array_repeat(3,16), array_repeat(4,16), array_repeat(5,16), array_repeat(6,16), array_repeat(7,16)) FROM range(8192) retains one column-dependent child and eight literal children. The same holds for least. Timing from cached debug dependencies is not used as a release-performance claim.

[viirya]

Agreed, the expanded routing broadcast every constant. 5f06022 merges each run of adjacent constant arguments into one before the row-wise picks, comparing them in argument order with the same Spark ordering. That keeps which of equal arguments is returned, e.g. -0.0 before 0.0, so non-adjacent constants stay separate. With your input (an 8,192-row ListArray<Int32> and eight constant arrays), cumulative allocation is now about 1.69 MB for both greatest and least, against 12.94 MB / 10.18 MB at ad5447e and 1.68 MB through DataFusion's functions, with identical results. The same change also covers nested float arguments, which already took this path. adjacent_constants_merge_in_argument_order covers the merging, float ties, and nested nulls.

[sunchao]

Thanks, the adjacent-constant case is fixed at 5f060225: my reproduction now allocates 1,690,016 bytes versus 1,675,260 through the previous DataFusion route.

[P2] Could we also preserve global constant reduction for float-free nested arguments? merge_constant_runs leaves constant, column, constant unchanged, so both constants still get broadcast and compared across the batch. For native inputs corresponding to the following query, cumulative allocation per 8,192-row invocation is 2,431,384 bytes at this head versus 1,215,772 previously, with identical output. Replacing greatest with least gives the same allocation increase.

SELECT greatest(
  array_repeat(-1, 16),
  array_repeat(CAST(id % 21 AS INT), 16),
  array_repeat(21, 16))
FROM range(8192)

Spark folds the two constant arrays individually and retains their positions around the column-dependent argument. The first-equal concern matters for floats, but equal float-free nested values have the same observable value, so those inputs can still reduce all constants with Spark's ordering before broadcasting. This part of the earlier regression remains unresolved.

Measured with an instrumented native harness using Arrow 59.3.0 and DataFusion 55.1.0, comparing the previous DataFusion functions with the exact-head Comet implementation and asserting equal outputs. These are cumulative allocation measurements, not release timing measurements.

[viirya]

Agreed. fb630b6 reduces all constants of a type without a float leaf to one, wherever they are, since equal values of such types are identical. Float types still reduce only adjacent constants, which keeps the earlier of -0.0 and 0.0. Your constant, column, constant input now allocates 1,215,900 bytes versus 1,215,772 through DataFusion's functions (2,431,384 at 5f06022), with identical output for both greatest and least.

[sunchao]

Reviewed commit 5f06022519ed189fde136a195896020f192835cb. Requesting changes for the new nested-scalar copying regression detailed inline and the unresolved separated-constant allocation regression in the existing discussion: #6761 (comment).

The original adjacent-constant allocation example is fixed. I found no correctness regression after checking Spark ordering, null handling, scalar/column combinations, sliced nested values, type normalization, and collation guards. Against branch-1.1, the nested-null correction is intended. The allocation regressions above are unintended.

Validation: 26 source tests plus 3 additional tests passed in a standalone harness using Arrow 59.3.0 and DataFusion 55.1.0. Current-head CI had 35 successful checks and 15 skipped, with no failures. I did not run the full Comet build locally.

[sunchao]

[P2] Could scalar-run reduction retain the selected original scalar instead of passing it through pick and zip? For nested values, zip copies the entire winning list on every comparison, even when the winner is unchanged and the comparison only examines its first element.

For native inputs corresponding to this query, an 8,192-row invocation allocates 2,052,684 bytes at 5f060225 versus 200,940 through the previous DataFusion route, with identical results:

SELECT greatest(
  array(CAST(id % 21 AS INT)),
  array_repeat(0, 65536),
  array(-1), array(-2), array(-3), array(-4),
  array(-5), array(-6), array(-7), array(21))
FROM range(8192)

The long literal is copied seven times before the final scalar replaces it. Mirroring the signs and using least reproduces the same increase. Comparing the one-row arrays with Spark's ordering and retaining the selected scalar preserves first-equal semantics without those copies. A scratch implementation reduced allocation to 201,324 bytes and returned the same output.

Measured cumulative allocated bytes per invocation with an instrumented native harness using Arrow 59.3.0 and DataFusion 55.1.0. This is an allocation regression, not a release timing claim.

[viirya]

Done in fb630b6. The reduction now compares the one-row arrays with spark_comparator and keeps the selected constant itself, so nothing is copied. Your long-literal input now allocates 201,324 bytes versus 200,940 through DataFusion's functions (2,052,684 at 5f06022). constants_merge_in_spark_order checks that the kept constant is the original one.