feat: support explode_outer

[comphead]

Which issue does this PR close?

Closes #2838.
Closes #5224

Fixes the native explode_outer / posexplode_outer gap tracked in that issue and referenced from DataFusion #19053.

Rationale for this change

Comet previously routed GenerateExec with outer = true back to Spark (Incompatible) because DataFusion's UnnestExec with preserve_nulls = true emits one null row for a NULL list but drops rows
whose list is empty. Spark's explode_outer / posexplode_outer must 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:

• New ListEmptyToNullExpr (native/core/src/execution/expressions/list_empty_to_null.rs) rewrites a List<T> to mark every empty row as null while preserving the original offsets, values, and column
  name.
• planner.rs wraps the array child with ListEmptyToNullExpr when explode.outer is true, before positions are computed and before the projection feeds UnnestExec. ListPositionsExpr inherits the
  modified null bitmap so pos and value stay aligned for posexplode_outer.

Serde:

• CometExplodeExec.getSupportLevel no longer returns Incompatible for op.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 Unsupported path.

Tests:

• Un-ignored explode_outer with empty array, explode_outer with nullable projected column, explode_outer with mixed null, empty, and non-empty arrays in CometGenerateExecSuite.
• Dropped the WHERE id != 4 workaround and stale allowIncompatible Config: directive in posexplode.sql.
• Added sql-tests/expressions/array/explode.sql covering explode / explode_outer (plus LATERAL VIEW and LATERAL VIEW OUTER) across every primitive element type (boolean, tinyint/smallint/int/bigint
  at 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 an expect_fallback for map input.

Are these changes tested?

Yes. New sql-tests/expressions/array/explode.sql runs through CometSqlFileTestSuite and previously ignored tests in CometGenerateExecSuite are now enabled.

Are there any user-facing changes?

explode_outer and posexplode_outer now run natively without requiring spark.comet.operator.GenerateExec.allowIncompatible = true. No behavioral change for explode / posexplode.

[andygrove]

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?

[comphead]

Thanks @andygrove for the review. Addressed the comments in 60fe5c0

[andygrove]

I dug into the ListPositionsExpr concern from my earlier review and filed #5224 for it.

Short version: ListPositionsExpr panics when its input ListArray has a non-zero offset base. It builds a fresh values array numbered from zero but reuses the input's original offset buffer, so ListArray::new unwraps an InvalidArgumentError from Arrow. GlobalLimitExec with a non-zero skip produces exactly that shape, since LimitStream does batch.slice(self.skip, ...).

This is pre-existing and not something you introduced. It reproduces on main today with plain posexplode. I had guessed the CometFilter that Spark inserts above the limit via InferFiltersFromGenerate would reset the offsets and keep the non-outer path safe, but it does not. All rows pass the predicate, so Arrow's filter returns the input arrays untouched and the slice survives.

What this PR changes is the blast radius. posexplode_outer falls back today, so it is safe by default. Once it runs natively it hits the same panic. Here is a test that shows the difference. It passes on main and fails with this PR:

test("posexplode_outer over limit with offset") {
  withSQLConf(
    "spark.sql.adaptive.enabled" -> "false",
    "spark.sql.leafNodeDefaultParallelism" -> "1",
    CometConf.COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED.key -> "true",
    CometConf.COMET_EXEC_EXPLODE_ENABLED.key -> "true") {
    Seq((1, Array(1, 2, 3)), (2, Array(4, 5)), (3, Array(6)), (4, Array(7, 8)), (5, Array(9)))
      .toDF("id", "arr")
      .createOrReplaceTempView("t")
    val df = spark.sql(
      "SELECT id, posexplode_outer(arr) FROM (SELECT id, arr FROM t LIMIT 4 OFFSET 1)")
    checkSparkAnswerAndOperator(df)
  }
}

With this PR the plan becomes CometExplode directly over CometGlobalLimit -1, 1 and the query fails with:

org.apache.comet.CometNativeException: called `Result::unwrap()` on an `Err` value:
InvalidArgumentError("Max offset of 9 exceeds length of values 6")

I verified the same failure on main by setting spark.comet.operator.GenerateExec.allowIncompatible=true, which is the behavior this PR makes the default.

One note on the test. spark.sql.leafNodeDefaultParallelism = 1 is required rather than cosmetic. With the default parallelism each partition produces a one-row batch, LimitStream discards whole batches instead of slicing, and the bug is masked. AQE off just keeps the plan readable.

I would suggest the Scala form over a posexplode.sql entry here. The SQL harness does support -- Config:, but the reproduction depends on how many files the INSERT writes and how the shuffle reader batches them, and that is harder to pin down from a .sql file.

Could you either pick up the fix from #5224 as a prerequisite and include this test passing, or add it as ignore referencing #5224? The fix itself is small, rebasing the offsets to zero so they line up with the newly built values array:

let base = offsets[0];
let rebased = OffsetBuffer::new(offsets.iter().map(|o| o - base).collect::<Vec<_>>().into());

I lean toward fixing it in this PR since this is the change that makes it reachable by default, but I am fine either way as long as it is tracked rather than silent.

[comphead]

It would be great having this test, so we can also validate DF native implementation when the time comes

[comphead]

@andygrove are you thinking on having a fix for both outer and regular explodes in this PR?

[andygrove]

Thanks for the updates. Everything from my last round is addressed, and addressed well.

• The unsafe NullBuffer::new_unchecked is gone in favor of the safe NullBuffer::new(combined).
• list_empty_to_null.rs now has a seven-test module covering both fast paths, mixed empty/null/non-empty rows, empties on top of a pre-existing bitmap, a zero-row batch, sliced input with a non-zero offset, and return_field nullability.
• The pre-projection column is named __comet_explode_outer_<name>, so the duplicate field name is gone.
• The pre-projection is gated on explode.position, so plain explode_outer no longer pays for a ProjectionExec it cannot share.
• Both query shapes I asked about are in explode.sql and mirrored in posexplode.sql, plus the two new batch-boundary tests with COMET_BATCH_SIZE = 4. Nice addition.
• The native_shuffle.md churn is reverted and the expressions.md wording now reads "Enabled by default".

Four things left.

CI: scalafix

The four Lint Java jobs are failing in Run scalafix check, and it is this PR. Removing the op.outer branch from CometExplodeExec.getSupportLevel left Incompatible imported but unused in spark/src/main/scala/org/apache/spark/sql/comet/operators.scala at line 61. The only other match in the file is a string literal. .scalafix.conf enables RemoveUnused, so dropping Incompatible from that import list should clear all four jobs.

The Spark SQL Tests (Spark 3.5) / spark-sql-sql_core-3 failure is not yours. It failed in Setup Spark after 1m36s while the other six 3.5 shards ran the same step and passed. That one just needs a re-run.

#5224 is now reachable by default

To answer your question above: yes, I would fix both in this PR rather than split it. native/core/src/execution/expressions/list_positions.rs is unchanged here, so it still builds a fresh values array numbered from zero while reusing the input's original offset buffer. A ListArray with a non-zero offset base panics inside ListArray::new. Today posexplode_outer falls back so users are safe. Once it runs natively it hits the panic. The fix is small and it covers plain posexplode at the same time:

let offsets = list.offsets();
let base = offsets.first().copied().unwrap_or(0);
let total_len = (*offsets.last().unwrap() - base) as usize;
// ... build values as today ...
let rebased = OffsetBuffer::new(offsets.iter().map(|o| o - base).collect::<Vec<_>>().into());
let result = ListArray::new(
    element_field,
    rebased,
    Arc::new(Int32Array::from(values)),
    list.nulls().cloned(),
);

Pair it with the posexplode_outer over limit with offset test from my earlier comment. If you would rather keep the fix separate, please add that test as ignore referencing #5224 so it is tracked rather than silent.

The TODO placeholder is still literal

planner.rs still says (TODO: link the Comet tracking issue here). I searched and no such issue exists yet. Could you file one for removing ListEmptyToNullExpr and the pre-projection once datafusion#19053 lands, then drop the link in place of the placeholder? A TODO: link shipped in a comment leaves nothing pointing at the workaround.

Benchmark numbers

Still missing. This change flips explode_outer and posexplode_outer from falling back to running natively by default, and there is no GenerateExec benchmark in the repo. Even ad-hoc timings in the description for an array column with a mix of empty and non-empty rows would confirm the native path is a win and quantify what the extra pass costs.

[mbutrovich]

First pass, thanks @comphead!

[comphead]

java.lang.RuntimeException: Error executing SQL 'SELECT id, explode(array(rand(0))) FROM test_explode_int WHERE id = 1' nondeterministic expressions are only allowed in Project, Filter, Aggregate or Window, found:
  explode(array(rand(0))),col
  in operator Generate explode(array(rand(0))), false, [col#93426].; line 1 pos 0;

However the test passed in pure Spark

[andygrove]

is this a real Spark rule? I could not find any mention of it in Spark source

[comphead]

Claude is ingenious sometimes.

[andygrove]

I built this branch locally and ran benchmarks so we have numbers on the record for the outer variants.

There was no synthetic-data benchmark for GenerateExec, only the TPC-DS micro query in CometTPCDSMicroBenchmark, which needs generated TPC-DS data. So I added an explodeExecBenchmark to CometExecBenchmark and ran that. I am happy to push it as a separate PR, or you are welcome to fold it into this one.

Setup

Apple M3 Max, JDK 17.0.10, Spark 4.1.3 / Scala 2.13 (the repo default), release native build with RUSTFLAGS="-Ctarget-cpu=native", local[5], 10M input rows written to snappy Parquet.

Two data shapes:

• mixed: one row in five holds a NULL array, one in five holds an empty array, the rest hold 1 to 3 elements. This is the shape the outer variants exist to handle and the one that drives ListEmptyToNullExpr down its slow path.
• all non-empty: every array holds 1 to 3 elements, so the outer cases take the ListEmptyToNullExpr fast path.

The benchmark checks findFirstNonCometOperator on the executed plan before timing, so none of these numbers are accidentally measuring a silent fallback. All six cases planned fully natively.

Two independent runs, best time in ms:

query	shape	Spark	Comet	Comet relative
explode	mixed	164 / 152	146 / 145	1.1X / 1.0X
explode_outer	mixed	168 / 169	122 / 122	1.4X / 1.4X
posexplode_outer	mixed	165 / 161	171 / 169	1.0X / 1.0X
explode	all non-empty	182 / 171	91 / 95	2.0X / 1.8X
explode_outer	all non-empty	166 / 158	89 / 95	1.9X / 1.7X
posexplode_outer	all non-empty	177 / 169	129 / 133	1.4X / 1.3X

What I take from this

explode_outer is a clear win. 1.4X on the mixed shape and 1.7X to 1.9X when the arrays are dense. That was the open question on this PR and I think it is answered.

ListEmptyToNullExpr is not costing anything measurable. On the dense shape, plain explode with no wrapper and explode_outer with the wrapper come out the same, 91 and 95 against 89 and 95. So the extra per-batch scan is effectively free even when it finds nothing to do. On the mixed shape explode_outer is actually faster than plain explode, 122 against 145. I believe that is DataFusion's preserve_nulls = false filtering cost rather than anything this PR introduced, since it shows up on the non-outer path.

posexplode_outer on the mixed shape shows no speedup. Comet's best time came out marginally behind Spark in both runs, 171 and 169 against 165 and 161. Spark's stdev is 16 to 18ms there so the gap itself sits inside the noise, but the direction was consistent across both runs and there is clearly no gain.

The pos branch looks like where the time goes. Comparing within the same run on the same data, explode_outer at 122ms against posexplode_outer at 169ms, and 89 to 95ms against 129 to 133ms on the dense shape. That is roughly plus 40ms either way, about 40% on top. ListPositionsExpr::evaluate fills a Vec<i32> with a scalar push per element in a nested loop, which is a fair amount of per-element work for what is just a repeated ramp. That is pre-existing rather than something you added here, and it also affects plain posexplode.

I do not think any of this blocks the PR. explode_outer is the headline feature and it delivers. But it seemed worth knowing that posexplode_outer is at parity on the shape that matters, and that the positions builder rather than the empty-to-null pass is the thing to look at if we want to improve it. Would you be up for filing a follow-up issue for ListPositionsExpr so it does not get lost?

Caveats on reading the table

Cases land in the 90 to 180ms range and Spark's stdev ranges from 3 to 25ms, so anything under about 10ms of difference is noise. The two shapes use different Parquet files, so cross-shape comparisons carry some scan-cost difference as well. The within-run, same-file comparisons are the trustworthy ones, which is why I leaned on the explode_outer against posexplode_outer delta rather than on absolute numbers.

[mbutrovich]

Two things worth doing before merge: the optimization I listed in the first review, and swap the planner.rs TODO for a link to #5210, which already exists and already explains why (apache/datafusion#22100 shipped NullHandling::PreserveAndExpandEmpty upstream, just not in a release yet). The ListPositionsExpr loop is worth a five-minute check against Vec::extend before filing the perf follow-up, but that's not blocking.

[mbutrovich]

I pushed my suggested changes so we can try to get this into 1.0.

[andygrove]

LGTM. Thanks @comphead and @mbutrovich

[mbutrovich]

Removing my "Requested changes" with a token "Approval" now that @andygrove took a look at my changes.

[andygrove]

Removing my "Requested changes" with a token "Approval" not that @andygrove took a look at my changes.

*now

[comphead]

THanks @andygrove and @mbutrovich for the review. Added 1 more commit to remove Claude's creativity on non-existent rules and also point to non-deterministic support in generators which added starting from Spark 3.5. eb19459