Skip to content

feat: use DataFusion unnest_outer instead of Comet ListEmptyToNullExpr - #6132

Merged
comphead merged 1 commit into
apache:mainfrom
comphead:unnest-outer-5210
Sep 24, 2026
Merged

comphead merged 1 commit into
apache:mainfrom
comphead:unnest-outer-5210

Conversation

@comphead

@comphead comphead commented Sep 23, 2026 •

Copy link
Copy Markdown
Contributor

Which issue does this PR close?

Closes #5210.

Rationale for this change

DataFusion 55.1.0, which Comet already pins, carries unnest_outer
(apache/datafusion#22100) as NullHandling::PreserveAndExpandEmpty: an empty input list
produces one NULL output row, exactly like a NULL list does.

That is Spark's explode_outer semantics, so Comet no longer needs its own bridge. Before
this PR the planner wrapped the array child in ListEmptyToNullExpr, a Comet-only
PhysicalExpr that rewrote every empty row to NULL so that preserve_nulls would pad it.
The planner can now ask DataFusion for the semantics directly.

No user-visible behavior changes. explode_outer and posexplode_outer already ran
natively.

What changes are included in this PR?

Planner (native/core/src/execution/planner.rs):

  • Drops the ListEmptyToNullExpr wrapper.
  • Selects NullHandling::PreserveAndExpandEmpty when explode.outer, and
    NullHandling::Drop otherwise. The non-outer arm is unchanged, since
    with_preserve_nulls(false) already mapped to Drop.

Operator (native/core/src/execution/operators/explode.rs):

  • find_longest_length, which serves posexplode, bumps empty rows to length 1. The bump is
    applied once after the row-wise maximum rather than once per array as upstream does. max
    is associative and commutative, so max(max(a, 1), max(b, 1)) == max(max(a, b), 1), and
    this leaves the map closure byte-identical to upstream.
  • The matching change in list_output_lens, Comet's fused single-List fast path, which
    serves explode. The two must agree element for element, because predict_output_lens uses
    the result both to place chunk boundaries and as precomputed_lengths for build_batch.
  • No change to unnest_list_array or create_take_indices. Upstream carries the whole
    feature in the length array, and the existing pad loop already emits the NULL. The
    contiguous-run fast path self-disqualifies for a padded row, because an empty row adds 1
    to capacity and 0 to the offset span.

Deletes native/core/src/execution/expressions/list_empty_to_null.rs (311 lines).

The module docs on the fork were rewritten. They said to delete the fork once Comet moved
to a release carrying apache/datafusion#24384. Comet is now on such a release, but the fork
still carries two paths that were never upstreamed: list_output_lens and the
contiguous-run slice in unnest_list_array. The docs now say that, so the next person does
not delete the file and silently regress explode.

How are these changes tested?

Existing coverage is the real gate, and it is unchanged: explode.sql and posexplode.sql
cover explode / explode_outer / posexplode / posexplode_outer across every primitive
element type, nested arrays and structs, NULL and empty arrays, LATERAL VIEW [OUTER], and
a map expect_fallback. CometGenerateExecSuite covers batch boundaries, sliced inputs
from limit/offset, and single rows that exceed the batch size.

The native planner test explode_evaluates_array_once_per_batch already asserts the full
Spark contract over [[10, null], [], null, [20]] for all eight combinations of
(outer, position, computed), including pos = [0, 1, null, null, 0].

Added in this PR:

  • longest_length_combines_arrays_before_the_empty_bump pins the multi-array row-wise
    maximum, which is the posexplode shape and the only one where the bump's placement is
    observable. The fixture discriminates: a row empty in one array but long in the other, a row
    empty in both, and a row empty in one and NULL in the other.
  • output_lens_substitute_per_null_handling pins the absolute per-row lengths for all three
    NullHandling modes and checks the fused and general kernels against them, so a mistake
    made in both at once cannot pass.
  • output_lens_handle_a_sliced_input gains a sliced fixture that also has a validity buffer.
  • contiguous_unnest_covers_empty_rows_and_dropped_nulls gains the outer case, asserting
    that the fast path is rejected once rows are padded.
  • The chunking-equivalence tests now include empty rows and run under the two modes the
    planner can actually produce.

Both length functions were mutation-tested: disabling the bump in either one fails the new
tests.

Spark behavior was verified against the Spark source rather than assumed. ExplodeBase.eval
returns an empty collection for both a NULL and an empty array, and GenerateExec
substitutes an all-NULL generator row, so the two cases are indistinguishable and pos is a
true NULL rather than 0 or -1. The codegen path reaches the same result through an
index == -1 sentinel. This is unchanged across 3.4.3, 3.5.8, 4.0.x, and 4.1.1.

Performance

native/core/benches/explode.rs gains a posexplode_fan_out group. Previously every arm
built a single ListUnnest, so nothing measured the positional path at all, and the outer
arms held only NULL rows, leaving the empty-row substitution unmeasured. The new RowMix
shape mixes NULL and empty rows, and the positional arms unnest a parallel positions column
alongside the array at short and long fan-out.

I could not obtain a trustworthy measurement. The machine available to me is loaded, and
criterion comparing an identical binary against itself reports:

posexplode_fan_out/dense/2               +20.6%  p=0.00  "Performance has regressed"
posexplode_fan_out/dense/10              -16.4%  p=0.00  "Performance has improved"
posexplode_fan_out/nulls_and_empties/2   -10.2%  p=0.00  "Performance has improved"
posexplode_fan_out/nulls_and_empties/10  -15.8%  p=0.00  "Performance has improved"

That is a noise floor of roughly ±20%, with confident p-values, from no code change. The
effect under discussion is one eq/zip pair per batch, which is far below that. An earlier
revision of this description quoted percentages for explode_outer_with_nulls; those were
collected under the same conditions and are not supportable, so I have removed them rather
than leave them standing.

What can be said without measurement: on the positional outer path this PR performs one
eq/zip pair per batch that the previous code did not, and the previous code additionally
ran a whole ListEmptyToNullExpr projection per batch that this PR deletes. The operator-only
benchmark cannot see that projection, so it understates the change in the planned query either
way. A release run on a quiet machine, or in CI, would settle it.

One plan-shape change: posexplode_outer over a plain column no longer needs the
pre-projection, because the wrapper was what made the child a non-Column. That is one
fewer ProjectionExec. No plan-stability golden file covers explode.

@comphead comphead added run-spark-4.1-tests Run the Spark 4.1 SQL tests on this pull request instead of waiting for the merge queue run-benchmark-check Run the benchmark compile and lint check on this pull request instead of waiting for the merge queue labels Sep 23, 2026
@github-actions github-actions Bot added enhancement New feature or request area:expressions Expression evaluation labels Sep 23, 2026

@sunchao sunchao left a comment

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Reviewed head 24c7e499265808e193c5117a757924a04c3c8fdc against base cccc08b7cee91fc4a3ec4ea4236dbc57906ccdd5 across native behavior, Spark planning, memory and errors, compatibility and design, and test validation.

No correctness defects found. Removing ListEmptyToNullExpr is a sound simplification.

  • Correctness and Spark semantics: NULL and empty arrays produce exactly one outer row with NULL value and NULL position. Plain explode drops both. Computed positional inputs remain evaluated once per batch. Sliced offsets, populated NULL slots, passthrough columns, and output chunking remained correct.
  • Design and complexity: This removes a validity-buffer rewrite and an extra projection for plain-column posexplode_outer. Retaining the existing operator fork is justified by its fused length calculation and contiguous slicing optimizations. The updated documentation explains that accurately.
  • Performance: There is an opportunity to avoid duplicate eq/zip work in the positional outer path, detailed in the inline comment. The current benchmark only exercises single-list explode. A matched local positional probe included the old wrapper/projection costs and suggested a slowdown, but used debug dependencies. It does not establish a release-build regression. A release positional benchmark covering short arrays and mixed NULL/empty rows would settle this.

Validation:

  • Local: 24 native-module tests passed, including 6,144 generated cases checking 163,216 output tuples against an independent oracle. These imported the production operator and position-expression modules directly and used dependency versions matching the committed lockfile, including Arrow 59.3.0 and DataFusion 55.1.0.
  • CI: 39 successful checks and 43 skipped. Logs confirm 1,593 native tests, all 46 generator-suite tests, both explode SQL files, and Spark 4.1 SQL suites passed. The tested merge commit e9239b2d853f3c8484b6816baddfa8bda35605d2 has exactly the reviewed head's source tree. Comet CI, Spark SQL CI.
  • Limits: No local full JVM/native integration run or release benchmark. Older Spark semantics were source-checked.

// Bump empty lists to length 1 so they produce a single NULL-padded output row.
// Runs after the NULL substitution above, which has already set NULL rows to 1,
// so they are not matched here.
length_array = zip(&eq(&length_array, &zero)?, &one, &length_array)?;

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could we apply this eq/zip once after the rowwise maximum? After NULL substitution, max(max(a, 1), max(b, 1)) == max(max(a, b), 1), so this preserves the result while avoiding a duplicate pair for posexplode_outer.

Could the release benchmark also cover the positional path, including short arrays and mixed NULL/empty arrays? The current benchmark constructs only one ListUnnest, so it does not measure this path. A matched local probe that included the old wrapper/preprojection costs suggested a slowdown, but its dependencies were built in debug mode. That is motivation to measure the release path, not evidence of a release-build regression.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Both applied, thank you.

The hoist. The identity holds, so I moved it: max(max(a, 1), max(b, 1)) and max(max(a, b), 1) are both max(a, b, 1). In the code the old form computed max_i max(s_i, 1) and the new one computes max(max_i s_i, 1).

It turned out to be a simplification beyond the saved work. The map closure is now byte-identical to upstream again, so the divergence is a single labeled step after the fold rather than a condition threaded through the loop, and the divergence list in the module docs got more precise.

Your comment also exposed a test gap. Every existing test passed exactly one array to find_longest_length, so nothing covered the fold this moves across. Added longest_length_combines_arrays_before_the_empty_bump, with a fixture chosen to discriminate: a row empty in one array but 2 long in the other (the bump must not inflate it past the other array), a row empty in both (the only row where PreserveAndExpandEmpty differs from Preserve), and a row empty in one and NULL in the other. Worth noting that the per-array form also passes all of these, which is independent confirmation that the two are equivalent rather than just my algebra.

The positional benchmark. Added, and you were right that nothing measured it: every arm built a single ListUnnest, and the outer arms held only NULL rows so the empty-row substitution was unmeasured too. posexplode_fan_out now unnests a parallel positions column alongside the array, over a new RowMix shape that mixes NULL and empty rows, at fan-out 2 and 10.

I could not get a number worth reporting from it. Criterion comparing an identical binary against itself on my machine gives:

posexplode_fan_out/dense/2               +20.6%  p=0.00  "Performance has regressed"
posexplode_fan_out/dense/10              -16.4%  p=0.00  "Performance has improved"
posexplode_fan_out/nulls_and_empties/2   -10.2%  p=0.00  "Performance has improved"
posexplode_fan_out/nulls_and_empties/10  -15.8%  p=0.00  "Performance has improved"

A ±20% noise floor with confident p-values, from no code change. Note that dense/* runs under NullHandling::Drop and explode_outer_with_nulls has one list column, so neither can be affected by this transform at all, which is how I spotted that the readings were noise rather than signal.

Your instinct to distrust the debug-dependency probe was right, and it applies to my own earlier numbers too: the percentages I had quoted for explode_outer_with_nulls in the description came from the same environment, so I have removed them rather than leave an unsupportable claim in place. The benchmark arms are committed, so the measurement is reproducible by anyone with a quiet machine.

On the underlying question of whether the positional outer path got slower: it now does one eq/zip pair per batch that the old code did not, and the old code additionally ran a whole ListEmptyToNullExpr projection per batch that this PR deletes. The operator-only benchmark cannot see that projection, so it understates the change in the planned query in the PR's favour. I did not want to claim a win on that basis without measuring it.

…llExpr`

DataFusion 55.1.0, which Comet already pins, carries `unnest_outer`
(apache/datafusion#22100) as `NullHandling::PreserveAndExpandEmpty`.
That is exactly Spark's `explode_outer` semantics, so the planner can
ask for it directly instead of rewriting empty lists to NULL first.

Closes apache#5210.
@comphead
comphead enabled auto-merge September 24, 2026 00:51
@comphead
comphead added this pull request to the merge queue Sep 24, 2026
Merged via the queue into apache:main with commit 5d59317 Sep 24, 2026
46 checks passed
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

area:expressions Expression evaluation enhancement New feature or request run-benchmark-check Run the benchmark compile and lint check on this pull request instead of waiting for the merge queue run-spark-4.1-tests Run the Spark 4.1 SQL tests on this pull request instead of waiting for the merge queue

Projects

None yet

Development

Successfully merging this pull request may close these issues.

Use DataFusion unnest_outer instead of Comet implementaion

2 participants