perf: slice the child instead of gathering it when unnesting - #5667
Draft
andygrove wants to merge 7 commits into
Draft
perf: slice the child instead of gathering it when unnesting#5667andygrove wants to merge 7 commits into
andygrove wants to merge 7 commits into
Conversation
Measures CometExplodeExec against Spark's GenerateExec across the four dimensions that drive generator cost: fan-out (array length 2, 10, 100), generator variant (explode, posexplode, and their outer forms), element type (bigint, string, struct), and the number of columns replicated alongside the generated one. One in ten rows holds a null array and another one in ten holds an empty array, so the outer variants do different work from the plain ones rather than measuring the same query twice. Each array column gets its own temp view so that no case is charged for scanning a column it does not read.
Inherit the session from CometBenchmarkBase instead of copying the override a fifth time. The copy differed from the base only in using local[5], while silently dropping the base defaults for the vectorized reader, whole-stage codegen, the Comet toggles, and ANSI mode, and carrying a shuffle-partitions setting that no query here shuffles. Also drive view creation and cleanup from one dataset table rather than maintaining the view names in two places, drop generator aliases that no measurement depends on, and attribute the whole-query-total caveat to the harness with a pointer to apache#5363.
Four fixes, each to a measurement that did not isolate what its case claimed to. Count the generated columns instead of writing them. `.noop()` writes `InternalRow`, so the Comet arm was converting every generated row and the Spark arm, whose `GenerateExec` already emits rows, was not: 419K conversions at fan-out 2 against 21M at fan-out 100, scaling with the dimension the group exists to measure. Terminating in an aggregate puts the only row boundary above the final exchange. Exclude `InferFiltersFromGenerate` for both engines. It matches on `outer = false`, so it gave `explode` and `posexplode` 209,714 rows and an extra filter while their outer variants got all 262,144, and the rate was normalized on rows the non-outer arms never saw. Give the struct dataset the same string field as the string dataset. `s1` through `s10` stayed under the writer's dictionary page threshold where 260,000-odd distinct values do not, so the element-type group was also comparing a dictionary-encoded column against a plain one. Equalize the scan between the carried-column cases with an always-true filter over k, s and v. Column pruning drops them from the generator's input, so `explode alone` still does not replicate them, but the scan reads and decodes all four columns in both cases rather than three fewer in one of them. Every counted column is now nullable. `NullPropagation` rewrites `count(c)` to `count(1)` when `c` is not, which would leave the carried columns unreferenced and let pruning drop them before the generator -- the whole dimension.
Addresses the second round of review. The position cases now end in `sum(pos)` rather than `count(pos)`. `pos` is declared non-null, so `NullPropagation` rewrote the count to `count(1)` and no position value was ever read -- which matters here because Spark hands the generator its existing loop index while Comet materializes a parallel List<Int32> through `ListPositionsExpr` and unnests it alongside the values. Every case now declares the aggregate row it must produce, and both engines are run against it untimed before the case is timed. A sink that stops reading what it names still reports a rate; this makes the next one fail loudly instead. Adds an `Explode - nested input` group for the wide, deeply nested shape asked for in review: a customer profile whose event list sits eight struct accessors down, with a second array of four-field structs inside each element, over 100K rows. The requested outer container was a map; that is an array of structs here because Comet has no native generator over maps yet (apache#2837), so the Comet arm of a map case would silently be Spark.
Measures the operator over in-memory batches, so a change to the unnesting kernels is not diluted by the Parquet scan and the aggregate that CometExplodeBenchmark necessarily includes.
Unnesting a single list column pads nothing, so `unnest_list_array` was building an index array holding one i64 per output element and then gathering the child through it, when the indices it built were the contiguous run the elements already sit in. Return a slice of the child instead. Comet takes this path for plain `explode` and for both columns of `posexplode`, whose position array is built with the same per-row lengths; `explode_outer` still gathers, since a NULL or empty row is padded and breaks the run. Guarding on the offset span alone is not enough. Arrow permits a NULL list slot to span elements, which the gather skips and a slice would not, and such a row can cancel out padding elsewhere and leave the totals agreeing with a run that is not the one to take. So the check also rejects a NULL row holding elements, and the view types, whose per-row offsets are independent. Two smaller things on the way past. `predict_output_lens` derived its per-row lengths through `find_longest_length`, which chains `length`, `cast`, `is_not_null` and `zip` to stay generic over list types: four allocating passes to subtract adjacent offsets, once per input batch. Comet only plans `List`, so take that case in one pass and keep the general version as the fallback. And `create_take_indices` appended the repeat indices one element at a time through a builder that has no validity to track; fill the buffer per run instead. Measured with the new native benchmark, on an Apple M3 Max: explode_fan_out/2 -74% explode_fan_out/10 -82% explode_fan_out/100 -90% explode_element_type/bigint -82% explode_element_type/string -91% explode_element_type/struct -91% explode_carried_columns/0 -82% explode_carried_columns/3 -60% explode_outer_with_nulls/bigint -13% explode_outer_with_nulls/string -8% The outer cases keep gathering, so they get only the length and index changes.
andygrove
force-pushed
the
explode-perf
branch
from
September 4, 2026 15:37
1e7d265 to
b111492
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Which issue does this PR close?
N/A — no existing issue. Happy to file one if reviewers would rather track it that way.
Note
Stacked on #5381. The first commit here is that PR's benchmark, which is what
measures the end-to-end effect of the rest. Review the last two commits; this needs a
rebase once #5381 lands. The diff against
mainwill shrink to the native changes then.Rationale for this change
ExplodeExecis a fork of DataFusion'sUnnestExec, and it inherited kernels written to begeneric over every list type DataFusion supports. Comet plans exactly one shape: a
List,depth 1, one column for
explodeand two forposexplode. Three costs in the general path areavoidable at that shape.
The largest is in
unnest_list_array. Unnesting a single list column pads nothing, so the takeindices it builds — one
i64per output element — are the contiguous runoffsets.first()..offsets.last(), and the gather through them reads the child straight throughin order. Rows of a
ListArrayare adjacent by construction, since rowiis[offsets[i], offsets[i + 1]), so the answer is already sitting in the child as one slice. Theoperator was allocating an index buffer the size of its output and copying every element to
reproduce it.
Two smaller ones.
predict_output_lensderives per-row lengths throughfind_longest_length,which chains
length,cast,is_not_nullandzipto stay generic — four allocating passesto subtract adjacent offsets, once per input batch, because
lengthreturnsInt32forListand NULL rows need substituting. And
create_take_indicesappended the repeat indices oneelement at a time through a builder that has no validity to track.
What changes are included in this PR?
unnest_list_arrayreturns a slice of the child when the gather would be a contiguous run.Comet takes this for plain
explode, and for both columns ofposexplode, whose position arrayis built with the same per-row lengths and null mask.
explode_outerstill gathers: a NULL orempty row is padded with a NULL that no slice of the child contains.
The guard is not just arithmetic. Checking that the offset span equals the capacity is not
sufficient, because Arrow permits a NULL list slot to span elements — the gather skips those and
a slice would include them — and such a row can cancel out padding elsewhere and leave the totals
agreeing with a run that is not the one to take.
populated_null_row_falls_back_even_when_the_totals_agreepins that case. The view types are rejected outright, since their per-row offsets are independent
and need not be ordered.
Worth noting for the guard's reach: it accepts NULL rows whose range is empty, which is what
builders and the Parquet reader emit. That matters because
InferFiltersFromGenerateputssize(arr) > 0 AND arr IS NOT NULLbelow every non-outer generator over a column, so a plainexplodein production is usually handed an array column with the NULL rows already filteredout — but when it is not, the rows still contribute nothing and the run still holds.
predict_output_lenscomputes lengths in one pass for the single-Listcase and keepsfind_longest_lengthas the fallback.create_take_indicesfills the buffer per run.The module header said a change to the forked region "either belongs upstream or does not belong
at all". This PR deliberately reverses that: the specializations are for the shape Comet plans,
not upstream's, and the header now says so and says what retiring the fork would cost.
Trade-off
Output batches now alias the input's child buffer rather than owning a compacted copy, and
ListArray::sliceleavesvalueswhole, so the alias is to the child of the whole input batch,not of the chunk. All chunks from one input batch share that buffer and between them fill it, so
there is no amplification when they are all retained; a downstream operator that keeps only some
of them pins all of it. That is bounded by one input batch's expansion, which
pending_inputalready holds materialized, and
BatchSplitStreamabove it already hands out slices of theinput.
How are these changes tested?
Eight new unit tests in
explode.rscover the fast path — child aliasing, a non-zero offset basefrom slicing, empty and dropped-NULL rows, the padded fallback, the populated-NULL-row case
above, and view input — plus equivalence of the fused length computation against
find_longest_lengthacross NULL handling, empty batches, and slicing. The existing chunkingtests already assert that chunked output matches unchunked output exactly, which is what would
catch a fast path that fired when it should not have.
CometGenerateExecSuite(37 tests) andCometExecSuite(144 tests) pass.Measurements
cargo bench --bench explode, added in the second commit, on an Apple M3 Max:explode_fan_out/2explode_fan_out/10explode_fan_out/100explode_element_type/bigintexplode_element_type/stringexplode_element_type/structexplode_carried_columns/0explode_carried_columns/3explode_outer_with_nulls/bigintexplode_outer_with_nulls/stringThe outer cases keep gathering, so they get only the length and index changes. The carried-column
case improves least of the sliceable ones because replicating the passthrough columns is a real
gather that this does not touch.
CometExplodeBenchmarkfrom #5381, same machine,local[1], Spark 4.1 / Scala 2.13. These arewhole-query times including the Parquet scan and the counting aggregate, so they are the diluted
view (Best ms):
The Spark arm is unchanged across every case, which is the control: the two runs are comparable
and the movement is Comet's.