andygrove opened a new pull request, #5362: URL: https://github.com/apache/datafusion-comet/pull/5362
## Which issue does this PR close? Related to #5210, which tracks the unnest-related workarounds in the explode planner. ## Rationale for this change `CometExplodeExec` plans onto DataFusion's `UnnestExec`, which emits exactly one output batch per input batch however many rows the unnesting produces. It never consults `datafusion.execution.batch_size` — there is no `batch_size` anywhere in upstream `unnest.rs`, and `UnnestStream` calls `build_batch` once per input batch and returns the whole result. So an 8192-row batch of 100-element arrays comes back as a single 819,200-row batch, and input batch count always equals output batch count. Two consequences: 1. Downstream operators receive arbitrarily large batches. 2. Peak memory scales with input batch size times array length rather than with `batch_size` — the full expansion of an input batch is materialized at once, which is an OOM risk on high-fanout explode. ## What changes are included in this PR? Adds `ExplodeExec` (`native/core/src/execution/operators/explode.rs`), a temporary fork of `UnnestExec` that consumes each input batch in chunks, and points the planner at it instead of `UnnestExec`. `find_longest_length` already computes how many output rows each input row expands into, with null handling accounted for. Prefix-summing it gives *exact* chunk boundaries for the depth-1 unnesting Comet plans, so each `build_batch` call produces at most `batch_size` rows and the oversized intermediate is never materialized. That is what bounds memory, not just output size. Two cases can't be chunked on the input side and are handled by slicing the built batch instead: - a single input row whose array is longer than `batch_size`, since one row is never split across output batches - recursive unnest (`depth > 1`), which Comet does not plan today, but the fallback keeps the operator correct if that changes ### Why a fork The unnesting kernels (`build_batch` and everything it calls) are private to `datafusion-physical-plan`, so they cannot be called from Comet without copying them. They are copied verbatim from DataFusion 54.1.0 (upstream revision `cc7565be1ee97ba8fa2f5d6da373c5e38d81bb13`) and marked as such, so the eventual deletion is mechanical. All Comet-specific behavior lives in `ExplodeExec` / `ExplodeStream` above the copied section. Note that 54.1.0 predates upstream's `NullHandling` enum and still uses `UnnestOptions::preserve_nulls`, which is why the planner still wraps empty arrays with `ListEmptyToNullExpr` for `explode_outer` semantics. That is unchanged by this PR. ### Deleting this fork The fix has been submitted upstream: - apache/datafusion#24383 (issue) - apache/datafusion#24384 (PR) Once Comet moves to a DataFusion release carrying that PR, delete this module and go back to `UnnestExec`. The module docs say so, and #5210 already tracks the related cleanup. I went with a fork rather than an opt-in upstream API because Comet is pinned to DataFusion 54.1.0, so anything landing in 55 or 56 does nothing here until the upgrade. A fork gets the fix now and deletes cleanly. ## Are these changes tested? Yes — six new unit tests in the new module: - `respects_batch_size` — 10 rows x 10 elements at `batch_size=8`, every output batch bounded, values and order preserved - `chunks_input_rather_than_slicing_output` — pins *how* the limit is met: 3 rows of 3 elements at `batch_size=4` must give `[3, 3, 3]` (input chunked per row), not `[4, 4, 1]` (built whole, then sliced). These are indistinguishable by row counts alone but have very different memory profiles, so without this a refactor could silently regress to build-then-slice. - `single_row_exceeding_batch_size_is_sliced` — the unavoidable case - `chunking_preserves_outer_semantics` / `chunking_preserves_non_outer_semantics` — chunked vs unchunked output compared directly for both `preserve_nulls` settings - `multiple_input_batches` — short tail chunk per input batch boundary Full native suite: 163 passed, 0 failed. `cargo fmt` and `cargo clippy --all-targets -- -D warnings` clean. JVM-side suite runs still in progress; I'll report results before marking this ready for review. ## Are there any user-facing changes? Explode now produces more, smaller batches, bounded by `datafusion.execution.batch_size`, and uses substantially less memory on high-fanout arrays. Query results are unchanged. `CometExplodeExec` now displays as `CometExplodeExec` rather than `UnnestExec` in native plan output. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
