[andygrove]

Which issue does this PR close?

Closes #5125.

Rationale for this change

GenerateExec runs natively only for explode and posexplode. inline (expanding an array of structs into columns) and stack (the standard unpivot) fall back to Spark and take the surrounding stage with them, often directly above a Parquet scan.

What changes are included in this PR?

• inline / inline_outer: a new inline flag on the Explode proto message. The native planner unnests the array as before, then has ExplodeExec flatten the struct into one column per field, using the struct unnesting that the UnnestExec fork already carries. Every generated column is reported nullable.
• Null struct elements: flatten_struct_cols folds a struct's null mask into each field it flattens, using children_with_parent_nulls from datafusion-comet-common. Arrow leaves field values under a null struct unspecified, and Spark's inline emits a row of nulls for a null element.
• stack: the serde rewrites stack(n, v1, ..., vk) into inline over array(named_struct(...), ...), one struct per output row, padding the last row with null literals of the column type. Each argument is still evaluated once per input row, as in Spark. If an argument cannot be converted, the operator reports that argument's own fallback reason.
• Docs: the expression and operator support pages, understanding-comet-plans.md, and audit notes for the three generators. Inline and Stack behave the same in Spark 3.4.3, 3.5.8, 4.0.1 and 4.1.1.

How are these changes tested?

• New SQL file tests expressions/array/inline.sql and expressions/misc/stack.sql, which assert native execution and match Spark. They cover NULL, empty and NULL-element arrays, NULL fields, the _outer variant, LATERAL VIEW, a struct field named like a carried column, nested and wide field types, arrays built in the query, uneven stack argument counts, NULL literal arguments, and filters and aggregates above the generator.
• A Rust unit test for flatten_struct_cols with a null struct over non-null field values, including a sliced input.
• A CometGenerateExecSuite test that a stack argument which cannot run natively falls back with its own reason.
• Ran locally on the default Spark 4.1 profile: CometGenerateExecSuite (54 passed), the inline, stack, explode and posexplode SQL files, and the explode Rust tests.

[comphead]

Stack will also be supported in DF56

[viirya]

Thanks for picking this up. Getting inline and stack off the fallback path is a nice win, since both tend to sit right above a Parquet scan. Reusing the existing ExplodeExec struct unnesting keeps the native side small, and planning stack as inline over array(named_struct(...)) matches Spark's Stack.eval index math exactly, including the null padding of the last row. Switching the dispatch from nodeName checks to matching on Inline, Stack and ExplodeBase reads much better too.

I left two inline comments. One is about the flatten_struct_cols unit test losing its sliced-input case, which is the shape the contiguous fast path produces in practice. The other is a question about whether liftFallbackReasons is needed here, and about the internal CreateArray / CreateNamedStruct reasons it surfaces.

[viirya]

The second commit dropped the sliced-input case and the non-nullable field from this test, but the PR description still says it covers a sliced input. I think the slice case is worth keeping. The contiguous fast path in unnest_list_array returns values.slice(first, capacity), so for every chunk after the first, flatten_struct_cols receives a StructArray with a non-zero offset. That is the normal production input for inline. The SQL tests fit in one batch and never reach it, and struct_nulls.rs has no slice case either. Could you restore the structs.slice(..) assertion and the non-nullable field?

[andygrove]

Restored the sliced-input case and the non-nullable field in f5cfc51.

[viirya]

I'm not sure this lift does what the comment describes. stackAsStructArray puts the original value expressions into the synthesized tree as the same instances, so a reason recorded on the plus_one UDF in the new test is already on a node under Stack.children. rollUpFallbackReasons walks op.expressions and should find it without the lift. Does the CometGenerateExecSuite test still pass with this line removed?

What the lift adds on top is the generic unsupported arguments for CreateNamedStruct and unsupported arguments for CreateArray reasons from nodes the user never wrote. Those will show up in the extended explain for a plain stack(...). It also runs for explode and inline, whose child is already in the plan. If there is a case that needs the lift, could you scope it to Stack, add a test that fails without it, and keep the synthesized wrappers' generic reasons out of what gets reported? Otherwise dropping it seems simpler.

[andygrove]

You're right: stackAsStructArray reuses the original argument instances, so their reasons are already under Stack.children. I removed the lift, and the plus_one fallback test passes without it.

[rich7420]

Two issues inline.

[rich7420]

Bound this rewrite with stack.supportCodegen; EXPLAIN currently allocates a driver-side struct expression for every requested row.

[sunchao]

Confirmed at f5cfc51e29. A bounded Catalyst probe using stackAsStructArray extracted unchanged from this head expands Stack(10000, 42) from 3 expression nodes to 30,001 nodes, including 10,000 structs. At 51 rows the rewrite already proceeds although Spark's supportCodegen is false. The supported Spark releases all cap stack code generation at 50 rows.

This allocation happens during convert, even before the childOp.isEmpty check, so planning/EXPLAIN for a large valid row count can exhaust driver heap without executing the query. Could we keep larger stacks on Spark before entering this rewrite and add a planning regression case above the bound? This is a method-level reproduction, not a full Comet execution test.

[viirya]

+1 on bounding this. I think the executor side is the stronger reason. The native planner evaluates the synthesized make_array in the ProjectionExec in front of ExplodeExec, so each input batch materializes batch_size * n structs before ExplodeStream gets to chunk the output. Spark's Stack.eval only holds the n rows of one input row at a time. Checking stack.supportCodegen in getSupportLevel would keep the rewrite from being built at all above Spark's 50-row limit, and that still covers every realistic unpivot.

[rich7420]

Pass row counts to RecordBatchOptions; inline(array(_partition,_partition)) fails on unpartitioned Iceberg tables when no columns are carried.

[sunchao]

Confirmed the native failure at f5cfc51e29 with the extracted flatten_struct_cols function and Arrow 59.3.1 from this PR's lockfile. A StructArray::new_empty_fields(2, None) with no carried columns fails with Invalid argument error: must either specify a row count or at least one column. The control using RecordBatchOptions::with_row_count(Some(2)) returns the expected two rows.

The Iceberg case is reachable: CometScanRule explicitly exempts metadata columns such as empty _partition structs from schema validation, and the attribute/array serdes accept that type. Could we preserve the unnested row count when constructing the batch and add a zero-field inline/inline_outer regression? I verified the Arrow failure and traced the serde path, but did not run an Iceberg integration test.

[viirya]

I'd lean toward falling back here rather than threading a row count through flatten_struct_cols. Even with a row count, a zero-column batch still has to leave the native plan and be exported to the JVM, and I'm not sure that path handles a batch with no columns. Returning Unsupported from getSupportLevel when the inline element struct has no fields keeps the previous Spark behavior and costs nothing, since array<struct<>> carries no data. stack cannot hit this because it always has at least one column. Would a plain SELECT inline(array(struct())) FROM t also reach it? Spark accepts struct(), so if Comet converts the zero-field CreateNamedStruct natively, that would make a simpler regression test than the Iceberg case.

[sunchao]

Reviewed f5cfc51e29 with five independent passes covering JVM semantics, Arrow validity and slicing, planner/output mapping, resource behavior, and tests/type gates. The restored sliced-input test and removal of the fallback-reason lift address the earlier comments.

Two P2 issues remain. I added reproduction evidence to the existing threads:

• Bound the stack rewrite: driver planning constructs a struct expression for every requested row, including rows Spark declines to code-generate. Large valid row counts can exhaust the driver heap during EXPLAIN.
• Preserve zero-column row counts: flattening array<struct<>> with no carried columns fails in Arrow batch construction. Unpartitioned Iceberg _partition values can reach this path.

No additional actionable issues found. Compared with branch-1.2, making these generators native is intended, but these cases regress the previous Spark fallback.

Validation: a bounded probe of the exact stack rewrite confirms linear expression-tree expansion. Nine standalone Arrow tests passed, including 9,216 mask/slice combinations and the zero-field failure with an explicit-row-count control. These are focused probes, not full Comet/Spark integration runs. The full native test attempt was blocked by an uncached Iceberg git dependency and a read-only local Cargo cache. Current hosted CI has a cancelled preflight at Mermaid rendering, failed Required Checks, and skipped runtime suites. Requesting changes for the two confirmed issues.

[viirya]

Thanks for the quick turnaround. The restored test now covers the sliced struct and the non-nullable field, and dropping the lift keeps the generic CreateArray / CreateNamedStruct reasons out of the explain. Both of my earlier comments are resolved.

I agree with the two remaining issues @rich7420 and @sunchao raised, and added a note on each thread. Also, CI on f5cfc51e29 was cancelled at Preflight, so the test suites have not run on the latest head yet.