feat: use DataFusion unnest_outer instead of Comet ListEmptyToNullExpr - #6132
Conversation
sunchao
left a comment
There was a problem hiding this comment.
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/zipwork in the positional outer path, detailed in the inline comment. The current benchmark only exercises single-listexplode. 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
e9239b2d853f3c8484b6816baddfa8bda35605d2has 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)?; |
There was a problem hiding this comment.
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.
There was a problem hiding this comment.
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.
24c7e49 to
470dccf
Compare
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 listproduces one NULL output row, exactly like a NULL list does.
That is Spark's
explode_outersemantics, so Comet no longer needs its own bridge. Beforethis PR the planner wrapped the array child in
ListEmptyToNullExpr, a Comet-onlyPhysicalExprthat rewrote every empty row to NULL so thatpreserve_nullswould pad it.The planner can now ask DataFusion for the semantics directly.
No user-visible behavior changes.
explode_outerandposexplode_outeralready rannatively.
What changes are included in this PR?
Planner (
native/core/src/execution/planner.rs):ListEmptyToNullExprwrapper.NullHandling::PreserveAndExpandEmptywhenexplode.outer, andNullHandling::Dropotherwise. The non-outer arm is unchanged, sincewith_preserve_nulls(false)already mapped toDrop.Operator (
native/core/src/execution/operators/explode.rs):find_longest_length, which servesposexplode, bumps empty rows to length 1. The bump isapplied once after the row-wise maximum rather than once per array as upstream does.
maxis associative and commutative, so
max(max(a, 1), max(b, 1)) == max(max(a, b), 1), andthis leaves the
mapclosure byte-identical to upstream.list_output_lens, Comet's fused single-Listfast path, whichserves
explode. The two must agree element for element, becausepredict_output_lensusesthe result both to place chunk boundaries and as
precomputed_lengthsforbuild_batch.unnest_list_arrayorcreate_take_indices. Upstream carries the wholefeature 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
capacityand 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_lensand thecontiguous-run slice in
unnest_list_array. The docs now say that, so the next person doesnot delete the file and silently regress
explode.How are these changes tested?
Existing coverage is the real gate, and it is unchanged:
explode.sqlandposexplode.sqlcover
explode/explode_outer/posexplode/posexplode_outeracross every primitiveelement type, nested arrays and structs, NULL and empty arrays,
LATERAL VIEW [OUTER], anda map
expect_fallback.CometGenerateExecSuitecovers batch boundaries, sliced inputsfrom limit/offset, and single rows that exceed the batch size.
The native planner test
explode_evaluates_array_once_per_batchalready asserts the fullSpark contract over
[[10, null], [], null, [20]]for all eight combinations of(outer, position, computed), includingpos = [0, 1, null, null, 0].Added in this PR:
longest_length_combines_arrays_before_the_empty_bumppins the multi-array row-wisemaximum, which is the
posexplodeshape and the only one where the bump's placement isobservable. 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_handlingpins the absolute per-row lengths for all threeNullHandlingmodes and checks the fused and general kernels against them, so a mistakemade in both at once cannot pass.
output_lens_handle_a_sliced_inputgains a sliced fixture that also has a validity buffer.contiguous_unnest_covers_empty_rows_and_dropped_nullsgains the outer case, assertingthat the fast path is rejected once rows are padded.
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.evalreturns an empty collection for both a NULL and an empty array, and
GenerateExecsubstitutes an all-NULL generator row, so the two cases are indistinguishable and
posis atrue NULL rather than 0 or -1. The codegen path reaches the same result through an
index == -1sentinel. This is unchanged across 3.4.3, 3.5.8, 4.0.x, and 4.1.1.Performance
native/core/benches/explode.rsgains aposexplode_fan_outgroup. Previously every armbuilt a single
ListUnnest, so nothing measured the positional path at all, and the outerarms held only NULL rows, leaving the empty-row substitution unmeasured. The new
RowMixshape 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:
That is a noise floor of roughly ±20%, with confident p-values, from no code change. The
effect under discussion is one
eq/zippair per batch, which is far below that. An earlierrevision of this description quoted percentages for
explode_outer_with_nulls; those werecollected 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/zippair per batch that the previous code did not, and the previous code additionallyran a whole
ListEmptyToNullExprprojection per batch that this PR deletes. The operator-onlybenchmark 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_outerover a plain column no longer needs thepre-projection, because the wrapper was what made the child a non-
Column. That is onefewer
ProjectionExec. No plan-stability golden file coversexplode.