feat: support explode_outer - #5192
Conversation
andygrove
left a comment
There was a problem hiding this comment.
Thanks for tackling this. The core approach looks right to me. I traced the null-bitmap logic and it handles the cases that matter: empty rows that are already null are excluded from the fast-path scan, the combined bitmap is existing & non_empty so a null row with garbage offsets stays null, sliced input arrays reconstruct correctly because offsets() is sliced while values() is not, and a zero-row batch takes the fast path. I also checked GenerateExec in Spark and the outer null row does put null in pos as well as col, and the analyzer makes those attributes nullable when outer is set, which matches Field::new("pos", Int32, explode.outer) in the planner.
The gaps I found are mostly around the new pre-projection wiring and the Rust expression itself.
Test coverage for the new pre-projection wiring
The type coverage in explode.sql is really thorough, thank you for that. Two shapes I could not find, and both exercise the new pre-projection specifically.
First, a query that carries the array column through alongside its own explosion, like SELECT id, arr, explode_outer(arr) FROM test_explode_int. That is the only place where the difference between the original array and the null-marked copy is observable, since the passthrough should still show [] for the empty row while the exploded value is NULL. Second, a query with no passthrough columns at all, like SELECT explode_outer(arr) FROM test_explode_int, which drives project_list empty so projections.len() is zero in the planner.
Could you add both? The same pair for posexplode_outer in posexplode.sql would be good too.
Unnecessary unsafe
In native/core/src/execution/expressions/list_empty_to_null.rs (the new_nulls match), NullBuffer::new already does len - buffer.count_set_bits() internally, so the new_unchecked call is doing the same work behind an unsafe block. Could this collapse to the safe version?
let combined = match existing_nulls {
None => non_empty,
Some(existing) => existing.inner() & &non_empty,
};
let new_nulls = NullBuffer::new(combined);Unit tests for ListEmptyToNullExpr
It would be good to have a #[cfg(test)] module in list_empty_to_null.rs. The bitmap combination is the heart of the change and the SQL tests only reach it through a full plan, so a failure there is a lot harder to diagnose. Worth covering the fast path returning the input untouched when no valid row is empty, a mix of empty, NULL, and non-empty rows, empties combined with a pre-existing null bitmap, a zero-row batch, and a sliced input ListArray with a non-zero offset. That last one matters because evaluate rebuilds the array from offsets(), values(), and nulls(), and those have different slicing semantics.
Duplicate field name in the pre-projection
In planner.rs, wrapped_name comes from the child field, so for explode_outer(arr) the pre-projection schema ends up with two fields both named arr. It works today because Column::evaluate only bounds-checks the index and never compares the name, but it makes EXPLAIN output confusing and it would break the moment anything downstream resolves that schema by name. Could the pre-projection column get a reserved name like __comet_explode_outer_arr, with the original child field name kept for the output column name in the second projection? That keeps the final schema unchanged.
Pre-projection when there is no positions column
The comment explains the pre-projection exists so ListPositionsExpr and the array passthrough share one evaluation. That reasoning only applies to posexplode_outer. For plain explode_outer, child_expr appears exactly once in project_exprs, so the extra ProjectionExec is per-batch overhead with nothing to share. Would gating the pre-projection on explode.position work, wrapping inline otherwise?
Related: since this PR makes explode_outer run natively by default where it used to fall back, do you have any numbers? There is no GenerateExec benchmark in the repo today, so even ad-hoc timings in the PR description for an array column with a mix of empty and non-empty rows would help confirm the native path is a win and quantify what the extra pass costs.
Tracking the upstream fix
Once datafusion#19053 is fixed upstream, ListEmptyToNullExpr and the pre-projection become dead weight. Could you file a Comet issue to remove them when that lands, and reference it from the comment in planner.rs? Otherwise this workaround will quietly outlive the bug it works around.
Docs
The blank-line removals in docs/source/contributor-guide/native_shuffle.md look unrelated to this change. I checked and main is already clean under prettier, and current prettier accepts the file either way, so nothing in CI is asking for them. Could you revert that file to keep the diff focused?
In docs/source/user-guide/latest/expressions.md, the new text says "Requires spark.comet.exec.explode.enabled=true". That config defaults to true, so "requires" reads like the user has to opt in. Maybe "enabled by default via spark.comet.exec.explode.enabled" instead?
Which issue does this PR close?
Closes #2838. Fixes the native
explode_outer/posexplode_outergap tracked in that issue and referenced from DataFusion #19053.Rationale for this change
Comet previously routed
GenerateExecwithouter = trueback to Spark (Incompatible) because DataFusion'sUnnestExecwithpreserve_nulls = trueemits one null row for a NULL list but drops rowswhose list is empty. Spark's
explode_outer/posexplode_outermust emit one null row for both cases, so anything containing empty arrays fell back to JVM whole-stage codegen.What changes are included in this PR?
Native:
ListEmptyToNullExpr(native/core/src/execution/expressions/list_empty_to_null.rs) rewrites aList<T>to mark every empty row as null while preserving the original offsets, values, and columnname.
planner.rswraps the array child withListEmptyToNullExprwhenexplode.outeris true, before positions are computed and before the projection feedsUnnestExec.ListPositionsExprinherits themodified null bitmap so
posandvaluestay aligned forposexplode_outer.Serde:
CometExplodeExec.getSupportLevelno longer returnsIncompatibleforop.outer. Unsupported cases (maps, non-deterministic generators, multi-input generators,COMET_EXEC_EXPLODE_ENABLED = false)still fall back to Spark whole-stage codegen through the standard
Unsupportedpath.Tests:
explode_outer with empty array,explode_outer with nullable projected column,explode_outer with mixed null, empty, and non-empty arraysinCometGenerateExecSuite.WHERE id != 4workaround and staleallowIncompatibleConfig:directive inposexplode.sql.sql-tests/expressions/array/explode.sqlcoveringexplode/explode_outer(plusLATERAL VIEWandLATERAL VIEW OUTER) across every primitive element type (boolean, tinyint/smallint/int/bigintat min/max, float and double with NaN / ±0 / ±Inf / NULL, decimal(18,4) and decimal(38,10) at boundaries, string with empty and unicode, binary, date, timestamp), nested
array<array<int>>,array<struct>, NULLs in id and array columns, literal arrays, empty tables, and anexpect_fallbackformapinput.Are these changes tested?
Yes. New
sql-tests/expressions/array/explode.sqlruns throughCometSqlFileTestSuiteand previously ignored tests inCometGenerateExecSuiteare now enabled.Are there any user-facing changes?
explode_outerandposexplode_outernow run natively without requiringspark.comet.operator.GenerateExec.allowIncompatible = true. No behavioral change forexplode/posexplode.