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]

This is the deprecated name for the config. Content in main is correct here.

[andygrove]

Actually, there probably shouldn't be updates the native shuffle documentation in the contributor guide as part of the explode_outer PR?

[comphead]

Right, Claude is hallucinating here

[andygrove]

does this TODO still need to be addressed?

[comphead]

I feel we need to keep track DF implementation, and after Comet switch to DF, the current implementation can be drastically simplified

[comphead]

more correctly would be to track #5210 though, I can address it in follow up PR

[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!

[mbutrovich]

This is the call site that runs ListPositionsExpr on the wrapped array. Per the discussion on #5224, ListPositionsExpr::evaluate reuses the input's original offset buffer against a freshly zero based values array, and that panics when the array has a non zero base offset, for example after GlobalLimitExec with a non zero skip.

Before this PR posexplode_outer fell back to Spark, so that path was unreachable by default. This PR removes the allowIncompatible gate, so it becomes reachable by default here.

Can we either pull in the offset rebasing fix from #5224, or add a regression test marked ignore referencing #5224, before this merges?

[mbutrovich]

This TODO is still a placeholder: (TODO: link the Comet tracking issue here). Can you file the tracking issue for removing ListEmptyToNullExpr and this pre-projection once datafusion#19053 lands, and link it here?

[andygrove]

@comphead Yes, the TODO can stay but should link to the actual issue(s)

[mbutrovich]

When existing_nulls is Some, this allocates non_empty and then allocates a second buffer for existing.inner() & &non_empty. Both conditions are already known per row from the has_valid_empty scan above. Can this collapse into one collect_bool pass:

let combined = BooleanBuffer::collect_bool(len, |i| {
    offsets[i + 1] > offsets[i] && existing_nulls.is_none_or(|n| n.is_valid(i))
});

That drops one allocation on the path where the input already carries a null bitmap.

I also looked at NullBuffer::union/union_many (arrow-rs #9692) as an alternative. They don't help here, since one side of the combination is a predicate rather than an existing null buffer, so materializing it first just to union it costs the same allocation this avoids.

[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