feat: support NullType output types in codegen dispatch

[grorge123]

Which issue does this PR close?

Closes #5525.

Rationale for this change

Untyped constructors such as map(), map(k, NULL) and array() leave NullType children in their output type. CometBatchKernelCodegen.canHandle rejected any output type containing NullType, so such expressions fell back to Spark and took their whole operator with them. Only the input side needs that restriction (CometScalaUDFCodegen cannot build a spec for a NullVector); on the output side the kernel just has to emit an all-null NullVector.

Admitting those outputs then exposed a set of downstream assumptions that NullType never arrives, on the JVM IPC paths and in native kernels. This PR closes them and adds a differential sweep so the next one is found here rather than in review. Several of the problems it fixes along the way are also on main; they are marked below.

What changes are included in this PR?

Codegen gate. canHandle accepts NullType in output types (nested included) while still rejecting it in inputs, and rejects duplicate struct field names, which Arrow's StructVector collapses. A NullType child is always declared nullable on both sides of the FFI boundary (Utils.declaredChildNullability), since native kernels that rebuild a list around the input's child fail on the nullability mismatch.

Codegen dispatcher.

• The dispatcher declares its output deeply nullable (map keys stay non-null), as native constructors and Parquet columns do. Two values of one Spark type otherwise reached native with different nested nullability depending on which side produced them, and kernels that compare their inputs' types (set ops, array_remove / array_position, CASE) rejected the pair.
• A kernel whose generated code does not compile falls back to the expression's interpreted eval, as Spark's whole-stage codegen does, instead of failing the query (on main too; for example element_at on an in-bounds literal index with ANSI off).
• Where Spark evaluates an expression through eval rather than generated code (spark.sql.codegen.factoryMode=NO_CODEGEN, beneath a non-leaf CodegenFallback expression, the input of an imperative aggregate but not its FILTER, a generator outside a whole-stage stage), the dispatcher compiles its kernel over eval too. trunc / date_trunc with a non-literal format and array_repeat with a nullable count are dispatched there, since their eval skips an argument that the native kernel evaluates.
• Each occurrence of a dispatched tree that contains a Catalyst Nondeterministic node gets its own kernel and state (DispatchOccurrence); identical occurrences used to share one.
• The dispatcher refuses a non-deterministic user function (an asNondeterministic() ScalaUDF, a non-deterministic Invoke, StaticInvoke or V2 function), reflect, and a subquery inside the dispatched tree, and the operator falls back to Spark. Spark advances such a function's state row by row in one shared object, which a kernel evaluating one expression over a whole batch cannot reproduce, and the native plan is serialized before the operator starts its subqueries. These ran through the dispatcher on main, with wrong results or a "has not finished" error.

JVM IPC paths. Arrow's MinorType.NULL factory rebuilds a NullType map key as nullable, which the IPC reader rejects; Utils.newArrowStreamWriter (now the only way to build an ArrowStreamWriter in main sources, scalastyle-enforced) repairs the key on every writer. VectorSchemaRootAppender loops forever on a NullVector directly under a struct; a broadcast build side with that shape stays on Spark's broadcast (CometBroadcastExchangeExec reports it Unsupported, naming the columns), since shipping it uncoalesced costs every consuming task one IPC stream per buffer, and coalesceBroadcastBatches keeps a defensive bypass on the same predicate. The non-AQE DPP rewrite follows the same gate, so the DPP subquery keeps reusing the join's exchange. array<null> coalesces and stays native.

Serde gates. Over NullType-bearing inputs, the following are routed through the codegen dispatcher (CodegenDispatchFallback) instead of their native kernels, so the projection stays in the Comet pipeline: array_union (drops entries), array_intersect (returns the other side's entries), array_except, array_repeat over an input whose native list item is non-nullable (map_entries; a CreateArray child is exempt), and array(...) over a NullType argument that is not a literal (make_array builds a single row for the batch). The following fall back to Spark: hash / xxhash64 (no Null arm), and collect_list over a struct with a NullType field (the ungrouped final merge fails on nested nullability, and plain nested inputs fail the same way on main). The dispatcher takes NullType as an output but not as an input column, so a gated shape whose NullType side arrives as an upstream column still falls back.

Null guards. CometElementAt (ANSI), CometArrayAppend, CometMapFromArrays, CometArraysZip, CometCoalesce and CometSize serialize a nullable child into a predicate copy and a THEN branch. A non-deterministic child inside such a guard is refused (NullGuard), since the two copies do not see the same rows. arrays_zip checks its arguments with one WHEN each instead of a single AND, so a later argument is not evaluated on rows Spark's generated code skips (an ANSI error there used to fail the query), and falls back where Spark uses its interpreted eval, which evaluates every argument. map_from_arrays checks the guard before its LAST_WIN opt-in. coalesce guards only the arguments it serializes twice, and a one-argument coalesce serializes as its argument (it used to become a CASE with no WHEN, which native rejects, on main too). array_append guards the item only when the array can be NULL. size builds no guard for a non-nullable child or in legacy mode.

Conditionals and set ops. CASE WHEN and coalesce cast each branch of a struct-bearing type to the expression's own type: a native CASE names a merged struct's fields after its ELSE branch where Spark uses the first, which differ under case-insensitive analysis. (Native IF reconciles its branches itself since #6458.) array_union / array_intersect / array_except cast both sides to the set op's own element type, made deeply nullable, since the native kernel compares field names and nested nullability. A cast that only relabels nested field names and nullability is Compatible, so these casts stay native.

Native.

• to_csv yields NULL for a row that renders to an empty string and reports itself nullable, as Spark does (on main too).
• GetStructField returns a scalar for a scalar struct input and rebuilds a NullType field as a fresh NullArray.
• arrays_overlap expands a scalar side to the batch length; with a literal array beside a column it compared only the first row or read past the one-row list (on main too).
• A nested list literal whose inner lists are all empty (a folded array(cast(array() AS array<bigint>))) is built at its declared depth; it came out one level deeper (on main too).
• The row shuffle writer's field-major paths gain the Null struct field case, and the writer recreates every NullType-bearing builder after a batch is written, since NullBuilder::finish keeps its length.
• Native round-robin shuffle falls back when a column it hashes contains NullType, since the native hasher has no Null arm.

Every serde whose support level moved publishes its reason through getUnsupportedReasons, so the generated compatibility pages carry the restriction.

What moves from native to Spark. The PR moves far more queries into Comet than out of it. On the composition sweep and the SQL fixtures (6,901 queries, ANSI on and off), against main: 1,482 go from fallback to native, 129 from an error to native or fallback, and 3 from a wrong answer to native. 18 go from native to fallback:

• 8 broadcast joins whose build side has a NullType directly under a struct. Main runs them when the build side is one buffer and hangs when it is several; shipping the buffers uncoalesced measured slower than Spark's broadcast.
• 4 non-deterministic arguments under a null guard (size, map_from_arrays, arrays_zip), where the guard's two copies do not see the same rows.
• 6 collect_list calls over a struct with a NullType field directly under it, which main handles. The gate is wider than it needs to be: it refuses any NullType under a struct to keep out the shapes that fail on main, where that struct sits inside an array or another struct. Narrowing it can be a follow-up.

Outside that sweep, non-deterministic Scala/Java UDFs, non-deterministic Invoke / StaticInvoke / V2 functions, reflect, and subqueries inside a dispatched expression now fall back to Spark (see the dispatcher above); scala_java_udfs.md documents them.

Benchmark. CometNullTypeColumnsBenchmark (separate commit) times the consumed map(id, NULL) projection and broadcast joins with a NullType build side on Spark, Comet, and Comet with the codegen dispatcher off (the NullType projection falls back and takes its operator with it, as before this PR; it keeps this PR's broadcast gate, so for the struct build side it is not main's native broadcast), with an array<null> control, at the default and a small batch size and with 2 and 64 broadcast buffers. It checks that every arm returns the same rows and profiles plan, allocation, memory and IPC stream counts per arm. Every case is on par with or faster than the dispatch-off arm: the gated build sides match it, and the array<null> join runs 2.0x Spark at 64 buffers against 1.1x with the dispatcher off.

How are these changes tested?

CometNullTypeCompositionSuite sweeps every NullType producer under consumers, operators and nesting containers, across ANSI, nullable, non-deterministic and cross-input settings, with the optimizer's null and comparison simplifications excluded so no consumer folds to a literal, and under physical profiles that vary batching and the exchange path (two-row native batches, the JVM shuffle's bypass and sort-based writers with and without forced spills, native shuffle, AQE, native columnar-to-row) plus one that opts every registered serde into its native kernel through allowIncompatible. It compares Comet with Spark with no tolerated failures, credits a case as native only when its whole plan ran in Comet, and enforces floors on the compared and native counts (28 tests, about 8 minutes on an i7-12700K). A registry check fails the suite if a registered array, map, struct or any-type aggregate serde is reached by no template, and a gate check asserts per gate that the serde is enrolled in the dispatcher, that its reason is published, and that the query keeps its CometProjectExec.

CometCodegenSuite covers per-occurrence kernel state, the refused user functions, reflect and subqueries, and the compile retry. SQL-file fixtures pin the mechanism for every gated shape (expect_dispatch / expect_native), the reason for every remaining fallback, each interpreted context, the ANSI evaluation order of arrays_zip and map_from_arrays, and the struct-name and nullability cases above. There are also tests in CometInMemoryCacheSuite, CometCodegenSourceSuite, UtilsSuite, CometJoinSuite, CometColumnarShuffleSuite, CometNativeShuffleSuite, CometNativePositionalRoundRobinSuite, CometNativeCastSuite, test_pyarrow_udf.py, and native unit tests for nested list literals, GetStructField, to_csv and the row shuffle writer. Each regression test fails with its fix reverted.

On the final commit, rebased onto 351facd26, with the native library in release mode: Spark 4.1 CometCodegenSuite, CometSqlFileTestSuite, CometNativeCastSuite, CometNullTypeCompositionSuite, GenerateDocsSuite and spotless:check, and Spark 3.5 CometCodegenSuite. Before the reflect refusal and the last rebase, on 1b488623a: Spark 4.1 CometSqlFileTestSuite, CometNullTypeCompositionSuite, CometArrayExpressionSuite, CometExpressionSuite, CometTemporalExpressionSuite, CometMapExpressionSuite, CometAggregateSuite, CometJoinSuite, CometExecSuite, CometJsonExpressionSuite, CometCodegenSuite, CometCodegenSourceSuite, CometShuffleSuite, DisableAQECometShuffleSuite, CometNativeShuffleSuite, CometNativePositionalRoundRobinSuite, CometNativeCastSuite, UtilsSuite, GenerateDocsSuite, the native unit tests, cargo clippy and cargo fmt; Spark 3.5 CometSqlFileTestSuite, CometNullTypeCompositionSuite, CometArrayExpressionSuite, CometExpressionSuite, CometAggregateSuite, CometCodegenSuite and CometNativeShuffleSuite; Spark 3.4 CometCodegenSuite. After the reflect refusal, CometCodegenSuite on Spark 4.1, 3.5 and 3.4, and Spark 4.1 CometCodegenSourceSuite, CometNullTypeCompositionSuite and CometSqlFileTestSuite. All passed. Spark's own SQL suites have not been run locally.

The known issues this PR leaves as they are, all present on main, are listed in the latest conversation comments.

Assisted-by: Claude Code (claude-opus-5-5)

[sunchao]

Source review of this pinned change found the two newly admitted broadcast failure paths below. Neither example was executed locally. The supplied snapshot has four action-required workflows and no passing checks; the author-reported local test results were not independently verified.

[sunchao]

Re-reviewed db1768fa2aea540821391c7aef4cff2a4b8c43fe. The two earlier witnesses are fixed in source. Two source-derived P2s remain; neither was executed here.

[P2] Reject duplicate names in newly admitted struct outputs. The output gate now admits transform(array(id), x -> named_struct('a', x, 'a', NULL)) over an INT input. Whole-expression ArrayTransform dispatch skips the inner CreateNamedStruct duplicate-name fallback. Spark retains both fields, but Arrow 18.3.0's default struct policy replaces the first a with the later NullVector. Generated output setup still casts child ordinal 0 to IntVector, so the query fails before rows are written. BASE rejected this Null-bearing output. Please retain fallback for duplicate names recursively, or allocate children without losing their ordinals.

[P2] Keep coalescing safe plain null lists. The new broadcast bypass also rejects plain array<null>, although its ListVector appender grows offsets/validity independently of the Null child and finalizes the child count. An already supported build payload such as IF(rand(17L) < 0.5, array(NULL), array(NULL,NULL)), selected after a join over a repartitioned broadcast build, therefore loses coalescing. Returning B original buffers makes each of P consuming tasks open B compression/IPC streams and schema roots instead of one: B×P setups rather than P. This is an operation-count consequence, not a measured speed ratio. Keep the struct/map protection, but allow the safe plain-list case to coalesce.

[grorge123]

We re-applied the representation-level serde restriction inside canHandle.
Besides that, we found another problem: filter(array(), ...) and map_filter(map(), ...) leave containsNull = false on a NullType child, and downstream native kernels that rebuild a list around that input (map_entries, array_repeat, slice) panic on the nullability mismatch. We fixed it by always declaring a NullType child nullable on both sides of the FFI boundary.

[sunchao]

Re-reviewed 1d0a6f493bbd60453234748d548c85d77348a98c. The previous duplicate-field, map-key schema, and broadcast-bypass issues are addressed in the inspected source. The five additional P2 cases below remain.

Could you add a small microbenchmark for consumed map(id, NULL) projections and the remaining struct/map broadcast bypass, with a plain array<null> control? Compare this head with Spark and safe prior paths using matched versions, settings, and warmup. Please confirm result equality and the executed native/fallback plans, and report throughput, allocations, peak/retained memory, and IPC setup counts for small/default batches and few/many build buffers.

[sunchao]

[P2] Preserve single evaluation of nullable stateful outputs

With codegen dispatch and ANSI enabled, consider element_at(transform(IF(monotonically_increasing_id() % 2 = 0, array(id), CAST(NULL AS array<bigint>)), x -> named_struct('id', x, 'n', NULL)), 1) over a native LONG batch containing id=0,1,2,3. Spark evaluates the transform once per row, retaining the struct for id=2. This gate now admits that Null-bearing output, but the existing ElementAt ANSI conversion serializes its left subtree into both the CASE predicate and the lookup. Native CASE tests all four rows, then reevaluates the transform on the two selected rows. The second selected row receives an odd counter and becomes NULL. Please materialize the left value once or retain Spark fallback for this composition. Current BASE rejects this output. The previous reviewed head admitted it but lacked the duplicated-left CASE, so the newly failing combination is in this increment. Source-derived witness, not executed.

[grorge123]

Fixed: CometElementAt keeps a non-deterministic nullable collection in Spark under ANSI, since the null guard's two copies each hold their own kernel state and the THEN copy only sees the CASE-selected rows. Reproduced the witness first (Comet returned [null] where Spark keeps [[2,null]]), and it now matches. The same guard is built by CometSize (non-legacy mode), CometArrayAppend, CometMapFromArrays and CometCoalesce, so they share the gate (NullGuard); CometSize drops the guard in legacy mode, where native already answers -1. The coalesce case is reachable on main without NullType (coalesce(IF(monotonically_increasing_id() % 2 = 0, array(id), NULL), array(id)) NPEs in columnar-to-row because the result is declared non-nullable); it is included since the sweep found it and the fix is one line on the shared guard.

Tests: expect_fallback queries in element_at_ansi.sql, map_from_arrays.sql and coalesce.sql, plus the non-deterministic dimension of CometNullTypeCompositionSuite.

[sunchao]

[P2] Residual stateful guard case across different arguments

At 79fc84008a7dca16a4c24f28036649c0af2603b0, the original nullable element_at witness is fixed, but the shared guard still misses a cross-input case. This is source-derived, not an executed reproduction. With a native LONG input t containing ordered ids 0, 1, 2, 3 in one partition/batch and codegen dispatch enabled, consider SELECT id, arrays_zip(transform(array(id), x -> named_struct('i', monotonically_increasing_id(), 'n', NULL)), IF(id % 2 = 0, array(id), CAST(NULL AS ARRAY<BIGINT>))) AS z FROM t. The first array is non-nullable and nondeterministic, while the second is nullable and deterministic, so neither matches this predicate. Spark evaluates the first array even on odd rows where the zip is NULL. CometArraysZip.convert instead puts the value-producing kernel inside a CASE whose THEN batch contains only even rows, changing the stateful child's evaluation sequence. The base rejected this NullType-bearing producer. This guard therefore misses a composition newly admitted by the PR. Could the guard preserve one evaluation before filtering or retain fallback when a nullable sibling filters a stateful child, with coverage for nullability and nondeterminism occurring in different arguments?

[grorge123]

Fixed: reproduced the witness first (Comet returned [[4,null],0] and [[5,null],2] where Spark keeps [[0,null],0] and [[2,null],2]), and the non-NullType flavour diverges the same way, so this is the guard's rule being too narrow rather than a NullType path. The witness's stateful child is a lambda, which runs through the JVM codegen dispatcher; its kernel cache is keyed by the serialized expression, so the guard's predicate copy and THEN branch run one kernel instance and share its counter: the predicate consumes it for the whole batch and the THEN branch continues from there. A natively evaluated stateful child gets its own instance per copy, but native CASE evaluates the THEN branch on the rows the predicate selected, so it diverges as soon as the guard filters. NullGuard now refuses any non-deterministic child inside a guard, whichever argument is the nullable one, and CometArraysZip, CometElementAt, CometArrayAppend, CometMapFromArrays, CometCoalesce and CometSize share it; CometSize builds no guard at all for a non-nullable child (or in legacy mode), so size(filter(arr, x -> x < monotonically_increasing_id())) is evaluated once and stays native.

Tests: expect_fallback witness in arrays_zip.sql; CometNullTypeCompositionSuite gains a cross-input sweep that puts a non-nullable stateful producer (including one whose length records the counter) beside a nullable deterministic sibling under every consumer, and a nullable deterministic sweep so the guards' ELSE branches run natively; the size witnesses are in CometArrayExpressionSuite.

[grorge123]

Beyond the five cases above, this revision adds CometNullTypeCompositionSuite, a differential sweep of the admitted shape space: every non-foldable NullType producer (transform(array(id), x -> NULL), map(id, NULL), transform_values(map(), ...), map_entries(map(id, NULL)), named_struct(..., NULL), an all-NULL aggregate) under every consumer, operator and nesting container Spark accepts, across ANSI on/off, a nullable non-deterministic wrapper, and native columnar-to-row on/off, comparing Comet with Spark.

The non-NullType flavour of the same item-field problem (slice(map_entries(map(k, v)), ...), slice(reverse(arrays_zip(...))), the #4789 nested-nullability contract) reproduces on main and is outside this PR's admission. The one native gap the sweep tolerates is also pre-existing: Comet's row shuffle writer has no case for a Null struct field (Unsupported data type of struct field: Null), matched by signature so every other failure stays red.

Assisted-by: Claude Code (claude-fable-5)

[sunchao]

The five earlier concrete witnesses are addressed in the inspected source at 79fc84008a7dca16a4c24f28036649c0af2603b0. A residual cross-input stateful-guard case remains in the existing discussion, and one newly reachable shuffle failure is detailed inline.

This pass used source analysis only. I did not run a Spark/Comet query, test or runtime reproduction. The current-head snapshot has no check runs or workflow runs, so CI validation is unavailable.

[sunchao]

[P2] Do not suppress shuffle regressions newly admitted by this PR

Source-derived, not executed. Over nonempty primitive Parquet t, use SELECT /*+ REPARTITION(3, c) */ c FROM (SELECT transform(array(id), x -> named_struct('v', x, 'n', NULL)) AS c FROM t) with native scan/codegen dispatch, CometShuffleManager, Comet shuffle enabled, shuffle mode auto, spark.comet.shuffle.convertFromSparkPlan.enabled=false, AQE off and JVM columnar-to-row. At the base the producer falls back, and the disabled conversion leaves its exchange in Spark. This PR admits a CometProject, so the array partition key rejects native shuffle but falls through to JVM columnar shuffle. The unsafe-row list writer then sends its struct's Null field to append_field, which reaches exactly the panic waived here. The writer defect is inherited, but its reachability under these settings is new. Please retain fallback for that schema or support Null struct fields, and do not waive failures that passed at the base under the same settings. With conversion enabled the base can also fail, so that is not a valid regression control.

[grorge123]

Fixed: the exact query above does not fail at this head (reproduced under the listed settings: CometProject into CometColumnarShuffle, results match), because a list of structs goes through the writer's row-major append_field, which has a Null arm. The shape that did panic is a top-level struct column with a Null field, e.g. REPARTITION(3) named_struct('v', id, 'n', NULL), which takes the field-major paths that lacked the arm; that was the failure the sweep tolerated. Both field-major paths now handle Null struct fields (every row null, as the row-major path does), and the sweep tolerates nothing: every producer under repartition, shuffle join, group-by and sort compares against Spark.

Tests: CometNullTypeCompositionSuite operator and nesting sweeps with the waiver removed.

[sunchao]

You're right. Correction to my [P2] example: at 79fc8400, the array-of-struct path reaches append_field, which already has a DataType::Null arm. My assertion that this path reaches the unsupported-struct-field panic was incorrect, and I withdraw that specific witness. Thanks for checking it.

I am reviewing the two field-major fixes and removal of the sweep waiver separately at 496c3f7f. This correction is based on the pinned source, not an executed SQL reproduction.

[sunchao]

Re-reviewed 496c3f7f against ef62b463. The broader nondeterminism guard addresses the cross-input case, and the new field-major NullType arms pass the single-batch controls. I found one additional P2 in builder reuse across batches, detailed inline.

I independently reran the exact-source native component harness: 8 tests passed and 3 failed on the same NullBuilder reuse defect. This is not a full Spark, JNI, or shuffle execution. The planner route is source-traced. Four workflows remain action_required, with no head or merge check results in the current snapshot. No fresh benchmark was run.

[sunchao]

[P2] Reset NullType builders before reusing them for another batch

These new arms handle the first batch, but process_sorted_row_partition creates its builders outside the batch loop and builder_to_array calls finish() without replacing them. In the pinned Arrow 58.4.0, NullBuilder::finish() leaves its length unchanged. With the exact current conversion functions, two consecutive two-row batches of struct<v:bigint,n:void> panic on the second finish because the struct has length 2 while its Null child has length 4. The single-batch and typed-null two-batch controls pass. A nested struct fails the same way.

SpillSorter sends an entire destination partition to this native call, so one call can exceed spark.comet.shuffle.jvm.batchSize. The PR now admits producers such as element_at(transform(array(id), x -> named_struct('v', x, 'n', NULL)), 1) over native primitive input. With CometShuffleManager, JVM shuffle mode, spark.comet.shuffle.convertFromSparkPlan.enabled=false, and AQE off, source tracing shows their exchange can enter this writer where the base retained Spark fallback.

Could we reset or recreate Null-containing builders between batches, or retain fallback, and add coverage spanning more than one writer batch? I reproduced the builder failure in an isolated harness using production conversion source and pinned dependencies. The SQL/planner route was source-traced, not executed end to end.

[grorge123]

Fixed: reproduced first with the sort-based writer (spark.shuffle.sort.bypassMergeThreshold=0, spark.comet.shuffle.jvm.batchSize=2, jvm shuffle mode, convertFromSparkPlan.enabled=false, AQE off, 16 rows into 2 partitions): named_struct('v', id, 'n', NULL) panics in StructBuilder::finish with (2 != 4) on the second batch, as you describe, and so do a nested struct and element_at(transform(array(id), x -> named_struct('v', x, 'n', NULL)), 1). The same reuse defect hits every other Null-bearing shape: a top-level NULL column fails the batch's row-count check, map(id, NULL) fails "keys and values have unequal length", and array<null> comes back with the wrong row count. Base control at ef62b463 under the same settings: the top-level NULL column and array(named_struct('v', id, 'n', NULL)) fail there too (row-count check and the same (2 != 4) panic through the row-major arm), the struct shapes hit the field-major unreachable instead, and map(id, NULL), array<null> and the element_at shape stayed in Spark on the base, so those three are admitted by this PR. NullBuilder::finish keeps its length in arrow 58.4.0 (a NullArray owns no buffers to hand over), so process_sorted_row_partition now recreates every builder whose type holds a Null anywhere (contains_null_type) after each batch; the other builders reset on finish and are kept. The bypass writer sends at most one batch per native call, which is why the single-batch controls and the sweep's REPARTITION(3) never saw it.

Tests: null_type_builders_start_every_batch_empty in row.rs drives two batches through the production make_builders / builder_to_array / recreate path for a Null field, a nested one and a top-level Null column; CometColumnarShuffleSuite "columnar shuffle spanning several native writer batches with NullType columns" runs the seven shapes above through REPARTITION(300) with jvm.batchSize=2 and the spill threshold lifted, under both AQE settings, comparing with Spark.

[grorge123]

Rebased onto main and pushed two commits on top of the fix.

The struct/map broadcast bypass is gone. Building the benchmark showed it was slower than the path it replaced: with the bypass in, a map<bigint, null> build side split into 64 uncoalesced buffers over 8 consumer tasks opened 512 IPC streams (one per buffer per task, CometBatchRDD.compute) and ran 0.9x Spark where the fallback path ran 1.2x, a cost that grows with buffers x tasks. The first commit replaces it with a planner gate: a build side with a NullType directly under a struct or map entry stays on Spark's broadcast (CometBroadcastExchangeExec.getSupportLevel reports it Unsupported, naming the columns), which is what happened before the dispatcher admitted NullType outputs. array<null> is insulated by the list and still coalesces natively. The gate and the coalescer's now defensive bypass share one predicate, Utils.hasNullTypeUnderStruct, so both lift together once Arrow's appender can grow a NullVector under a struct; UtilsSuite runs the real appender over every shape and pins the predicate to exactly the hanging ones, and CometJoinSuite asserts the fallback reason and that the plan carries Spark's BroadcastExchangeExec.

The benchmark is the second commit: CometNullTypeColumnsBenchmark (SPARK_GENERATE_BENCHMARK_FILES=1 make benchmark-org.apache.spark.sql.benchmark.CometNullTypeColumnsBenchmark). It covers the cases, arms, checks and dimensions you listed and nothing else, on Spark's default settings. Numbers below are from one run on this head: Spark 4.1 / Scala 2.13, JDK 17, local[1], i7-12700K; Spark's Benchmark harness with its 2s warmup and 2+ timed iterations per arm, all arms in one session.

Cases. Consumed map(c1, NULL) projection and a transform to array<null> control, each over 10M rows; a hinted broadcast hash join (1M-row probe) whose build side projects an array<null> control (coalesces), a map<bigint, null> and a struct with a NULL field (the two gated shapes), with the build sized for 2 and for 64 broadcast buffers (16K and 512K rows: the native scan emits 8192-row batches and the broadcast collects one buffer per batch).

Arms. Spark; Comet (this head); Comet with the codegen dispatcher off, i.e. the prior path where a NullType projection falls back and takes its operator with it; and the two Comet arms again with spark.comet.batchSize=1024.

Result equality and plans. The benchmark computes an order-independent row digest of every arm's result and fails on any difference; all 8 cases report identical results across the 5 arms. It records, per arm, the first non-Comet operator of the final (post-AQE) executed plan: the Comet arm is fully native in every case except the two gated shapes, where it falls back at HashAggregate over Spark's broadcast exactly like the prior-path arm; the prior-path arm falls back at Project for the dispatched projections and at HashAggregate for the array<null> join.

Throughput (best time, ms):

case	Spark	Comet	prior path
map(c1, NULL) consumed	508	514 (1.0x)	497
array<null> control	658	537 (1.2x)	642
join, 2 buffers: array<null> (control, coalesced)	74	51 (1.4x)	48
join, 2 buffers: map<bigint, null> (Spark broadcast)	63	45 (1.4x)	44
join, 2 buffers: struct (Spark broadcast)	58	41 (1.4x)	40
join, 64 buffers: array<null> (control, coalesced)	195	99 (2.0x)	175
join, 64 buffers: map<bigint, null> (Spark broadcast)	272	241 (1.1x)	247
join, 64 buffers: struct (Spark broadcast)	212	190 (1.1x)	190

The gated shapes match the prior path at both sizes; the dispatched map(c1, NULL) projection is within noise of it (stdev 15 ms); the array<null> cases are where staying native pays.

IPC setup counts. Every consuming task decodes every broadcast buffer, so streams = buffers x consumer tasks, both read back from the executed plan and the consuming stage's task count. The array<null> build side coalesces 2 and 64 batches into 1 buffer, so its consumer opens 1 stream at either size; the gated shapes open none (Spark broadcast).

Allocations and memory (one profiled execution per arm after warmup, JVM-wide). JVM allocation for the joins is 42-556 MB on the Comet arms against 62-563 MB on Spark; the two projections allocate 2.2-2.5 GB on every arm (the row conversion of 10M rows). Heap peak is lower on the Comet arms in every case. Retained heap after GC and retained Arrow memory are within GC jitter (Arrow retained is 0 everywhere). Native join build memory, from the join's build_mem_used metric, is 0.3 MB at 2 buffers and 8 MB at 64, identical between default and small batches; Spark's peakExecutionMemory is 1-68 MB.

Small batches. spark.comet.batchSize=1024 costs the row-conversion-heavy projections 0.8x of the default on both Comet arms and changes nothing for the joins. Two observations: the native Parquet scan emits 8192-row batches whatever this setting is, so the broadcast buffer count is set through the build size; and the small-batch projection arms peak at 1.6 GB of heap against 0.64 GB at the default, since the row conversion allocates per batch.

Assisted-by: Claude Code (claude-fable-5)

[andygrove]

Several of the new Unsupported branches sit on serdes that do not mix in CodegenDispatchFallback, so unlike the array_intersect and array_except cases you reason about in the comments, they never reach the JVM dispatcher and instead drop the whole projection back to Spark. CometArrayRepeat is the one I would look at first, because its gate fires for any element that is itself an array Spark typed with containsNull = false, so something like array_repeat(array(c), 2) over a non-nullable column now leaves the Comet pipeline entirely rather than staying native. CometArrayUnion, CometCreateArray, CometSlice, and the NullType result gate in CometIf / CometCaseWhen / CometCoalesce are in the same position. Since CometBatchKernelCodegen.canHandle accepts all of those output types after this change, would mixing in CodegenDispatchFallback keep them in the pipeline instead?

The new restrictions do not reach the generated compatibility pages. CometArrayUnion, CometArrayIntersect, CometArrayExcept, CometSlice, CometArrayRepeat and CometCreateArray each gained an Unsupported branch but still return an empty getUnsupportedReasons(), and CometArraysZip has one that does not mention the new null-guard restriction. hash.scala does this the right way by adding unsupportedNullTypeReason to HashUtils.unsupportedReasons, so it would be good to follow that wherever a support level moved in this PR. CometArrayExcept is the notable one, since its element-type restriction used to at least carry the text inline in the Incompatible note.

CometNullTypeCompositionSuite is now registered in both pr_build_linux.yml and pr_build_macos.yml, and it is large. By my count it expands to 37 tests, and the operator sweep alone runs 8 producers across 32 operator templates against both engines in each of six physical profiles, three of which set spark.sql.shuffle.partitions to 300 over 900 rows. The nesting sweep asserts a floor of 200 compared cases per profile on top of that. How long does the suite take end to end on one runner? If it is long, pruning the profile matrix on the consumer sweeps would still leave the shuffle and batching coverage intact.

[grorge123]

Thanks, all three addressed in one commit on top, rebased onto main (3810936b4).

Dispatcher enrollment. CometArrayRepeat, CometArrayUnion, CometCreateArray, CometIf, CometCaseWhen and CometCoalesce now mix in CodegenDispatchFallback, so their Unsupported cases run through the JVM codegen dispatcher and the plan stays fully native. CometSlice needs no gate after the rebase: #5766's convert no longer serializes a return type, so the mismatch it guarded against cannot happen and slice(map_entries(map(k, NULL)), 1, 1) runs natively. On array_repeat, your example made me look closer: array_repeat(array(c), 2) runs natively on main because native make_array emits a nullable item whatever Spark's containsNull says; what fails is an input whose native item is really non-nullable (map_entries, "ListArray expected data type List(non-null Struct(..)) got List(Struct(..))"). The gate now exempts a CreateArray child, so array(c) stays on the kernel and map_entries(..) goes through the dispatcher. Enrolling coalesce exposed a dispatcher bug: a top-level struct<.., x: null> output failed to allocate (StructVector's constructor builds a NullableStructWriter over the field's children, which has no arm for a Null child); the output vector now adds its children after construction, with a unit test. One boundary: the dispatcher accepts NullType as an output but not as an input column, so a gated shape whose NullType side arrives as an upstream column still falls back. The SQL file tests that asserted a fallback now assert the native plan, and CometNullTypeCompositionSuite checks per gate that the serde is enrolled, the reason is published, and the query keeps its CometProjectExec.

Compatibility pages. Every serde whose support level moved publishes the reason through getUnsupportedReasons, following HashUtils; the runtime note starts with the published text and the suite asserts that, so the two cannot drift apart.

Suite size. It was 37 tests in 10 min 49 s on an i7-12700K, so likely two to three times that on a runner, and the consumer sweeps were the bulk. Each sweep now runs only under the profiles that can change its outcome (a null guard is per row, so batching cannot affect the nullable sweep; a stateful producer only diverges across a batch boundary, so the kernel choice cannot affect that one; a non-deterministic child leaves every kernel, so neither reaches the last), the columnar-to-row-only physical profile is folded into the sort-writer one, and Spark's reference rows are cached across profiles with the same row count. That is 28 tests in about 8 min on the same machine, with the operator and nesting sweeps still under every shuffle path and both batch sizes; nothing removed had ever caught a failure the kept combinations did not.

Rebase notes: CometElementAt takes main's #5544 fix, the NullArray special case in GetStructField moved into #5769's shared child_with_parent_nulls, and the two duplicate-field-name checks are folded into one with main's reason wording.

Assisted-by: Claude Code (claude-fable-5)

[sunchao]

Summary

• Prior state and problem: NullType outputs forced otherwise supported projections back to Spark.
• Design approach: Admit those outputs through codegen, repair Arrow schemas, and gate unsupported native compositions.
• Correctness / compatibility analysis: Compared relevant Spark sources across supported 3.4–4.2 versions. One P2 remains: LAST_WIN bypasses the new stateful-input guard in map_from_arrays. This leaves the earlier single-evaluation concern partially unresolved.
• Key design decisions: Shared nullability and broadcast predicates keep the JVM/native boundaries consistent. Retaining Spark broadcast for unsafe struct/map shapes avoids multiplying IPC streams, while plain array<null> still coalesces.
• Implementation sketch: Updates output-vector allocation, expression dispatch, IPC writers, struct-field extraction and shuffle-builder reuse, with composition tests and a benchmark.
• Behavioral changes worth calling out: More projections remain in Comet, unsupported compositions fall back, and empty CSV renderings become NULL. The benchmark compares Spark, current Comet and dispatcher-disabled execution. Its reported performance was not independently rerun.
• Suggested improvements: Apply the nondeterminism guard before the LAST_WIN compatibility branch and cover that branch with native opt-in enabled.

Reviewed all 61 changed files and six commits from 3810936b40af3562003120742f7d258b234e4dc5 to ba976f0e4bcb978ac9d74267e90497e769118ac4. Routed skills: review-comet-pr and audit-comet-expression. Read existing discussions before assessing findings.

Exact-head CI: label passed. CI, PyArrow UDF Tests, CodeQL, Check PR Title and Delta Contrib Build Gate are action_required, providing no passing build/test results.

Validation: 14 focused native regression tests passed. A disposable native reproduction confirmed the map mismatch, with two controls passing, and standalone Spark 4.1.3 confirmed the expected SQL result. The Scala/JNI execution route was source-traced, not run end to end. JVM suites could not run because Maven bootstrap failed resolving repo.maven.apache.org. The checkout is unchanged.

[sunchao]

[P2] Apply NullGuard before the LAST_WIN branch. With spark.sql.mapKeyDedupPolicy=LAST_WIN and spark.comet.expression.MapFromArrays.allowIncompatible=true, the earlier Incompatible return bypasses this guard and permits duplicated stateful evaluation. Over one Parquet batch containing IDs 0–3, map_from_arrays(IF(monotonically_increasing_id() % 2 = 0, array(id), NULL), transform(array(id), x -> NULL)) should retain {2: NULL} at ID 2, but the native CASE/map evaluation returns NULL there. There are no duplicate keys, so this is unrelated to the documented deduplication incompatibility. The base rejected the transform's array<null> output and retained Spark execution; this PR newly admits the failing composition. Check nondeterminism before the policy branch and add coverage with both settings enabled.

Evidence: Executed the SQL on Spark 4.1.3 using local[1] and single-partition Parquet IDs 0–3: results were [{0: NULL}, NULL, {2: NULL}, NULL], and the optimized plan retained both expressions. A disposable Cargo test reconstructed CometMapFromArrays.convert's native CASE/map tree using exact-head production MonotonicallyIncreasingId and locked DataFusion 55.1.0/Arrow 59.2.0 dependencies. It produced [{0: NULL}, NULL, NULL, NULL], failing the row-2 validity assertion. Source tracing confirms that LAST_WIN returns before NullGuard and allowIncompatible calls convert directly. Reproduction preserved at /tmp/review5526-1790617989-reproduction.rs. Full Comet SQL execution was not available.

[andygrove]

This is a light fully automated review since there are so many PRs open.

Following up on my earlier compatibility-page comment: four serdes whose support level moved in this PR still return an empty getUnsupportedReasons(), so GenerateDocs will not list their new restriction. CometSize (spark/src/main/scala/org/apache/comet/serde/arrays.scala:931) and CometArrayAppend (arrays.scala:57) now return Unsupported(NullGuard.reason) for a non-deterministic child. CometCollectSet and CometCollectList (spark/src/main/scala/org/apache/comet/serde/aggregates.scala:1056 and :1132) now return Unsupported through CometCollectAggregate.nestedNullLevel for any NullType-bearing input. For example, on Spark 4.x, where legacySizeOfNull is off by default, size(filter(arr, x -> x > rand())) over a nullable arr now falls back to Spark, but the size entry on the generated page won't mention it. CometArraysZip and CometMapFromArrays already publish NullGuard.reason, so the same override would cover size and array_append. Hoisting the collect reason into a val would let both aggregates publish it too. The gate check in CometNullTypeCompositionSuite ("NullType gates enroll in the JVM codegen dispatcher and publish their reasons") lists arrays_zip and map_from_arrays but none of these four, which is why it passed. Could you add them there as well? The collect ones would need a lookup through aggrSerdeMap rather than exprSerdeMap.

[sunchao]

Summary

• Prior state and problem: Codegen rejected NullType outputs, forcing otherwise supported projections back to Spark.
• Design approach: Admit these outputs, repair Arrow schemas, and guard native consumers that cannot handle them correctly.
• Correctness / compatibility analysis: Compared relevant Spark sources across supported 3.4–4.2 versions. No additional introduced P1/P2 issues found within this review. The existing P2 LAST_WIN guard-order issue remains unresolved. With MapFromArrays.allowIncompatible=true, LAST_WIN bypasses NullGuard, allowing duplicated stateful evaluation and incorrect results.
• Key design decisions: Shared nullability and broadcast predicates consolidate the Arrow workarounds. Unsafe struct/map broadcast shapes retain Spark execution, while plain array<null> still coalesces, avoiding additional IPC streams per consuming task.
• Implementation sketch: Updates output-vector allocation, expression dispatch, IPC writers, struct-field extraction and shuffle-builder reuse. Adds composition tests and a benchmark comparing Spark, Comet and dispatcher-disabled execution.
• Behavioral changes worth calling out: More projections remain in Comet, unsupported compositions use fallback, and empty CSV renderings become NULL. Benchmark results were inspected but not independently rerun.
• Suggested improvements: Resolve the existing LAST_WIN issue by checking nondeterminism before the policy branch and testing both configuration settings together.

Reviewed all 61 changed files and six commits from 3810936b40af3562003120742f7d258b234e4dc5 through ba976f0e4bcb978ac9d74267e90497e769118ac4. The PR is not a draft. Routed skills: review-comet-pr and audit-comet-expression. Existing reviews, comments and threads were read before assessing findings.

Exact-head CI: label passed. CI, PyArrow UDF Tests, CodeQL, Check PR Title and Delta Contrib Build Gate are action_required, providing no passing build/test results.

Validation: 14 focused native tests passed. A disposable reproduction reconfirmed the existing map mismatch: row 2 becomes NULL instead of {2: NULL}. Both controls passed. The Scala/JNI route was source-traced, not executed end to end. Maven bootstrap failed resolving repo.maven.apache.org, preventing JVM-suite validation. Disposable tests were removed and the checkout is unchanged.

[grorge123]

@sunchao @andygrove thank you both for the reviews. The branch is rebased onto today's main (92b43cfaa), and everything since the last push is in the last commit.

Your points

• LAST_WIN (@sunchao): CometMapFromArrays now runs the null guard before the LAST_WIN branch, so allowIncompatible can no longer hand a stateful child to convert. map/map_from_arrays_last_win_opt_in.sql and the gate test in CometNullTypeCompositionSuite both pin that order. fix: validate map constructor row lengths and null short-circuiting #5846 landed in the meantime and fixes the evaluation order of map_from_arrays itself, so I rebased onto it and dropped my own version of that change. I kept one ANSI fixture from it, for values produced by a lambda the codegen dispatcher evaluates, since the new tests cover native values only.
• Published reasons (@andygrove): size, array_append, arrays_zip, collect_list and collect_set now publish every reason they return, through shared vals. The gate test covers them, looking the aggregates up through aggrSerdeMap. It checks size under spark.sql.legacy.sizeOfNull=false, since Spark 3.x does not guard by default.

What else changed

Both points were instances of a wider pattern, so I went through the rest of the branch looking for more of the same:

• arrays_zip had the evaluation-order problem that fix: validate map constructor row lengths and null short-circuiting #5846 fixed for map_from_arrays: a single AND of null checks could evaluate a later argument on rows Spark's generated code skips, raising an ANSI error there. It now uses one WHEN per argument. There is an ANSI fixture for it, and it fails with the old guard. Spark's interpreted ArraysZip.eval does evaluate every argument, so where Spark would use it, arrays_zip over several arguments now falls back to Spark. Comet tracks those contexts while it serializes: spark.sql.codegen.factoryMode=NO_CODEGEN, beneath a non-leaf CodegenFallback expression (such as the filter that array_compact becomes, even with codegen on), the input of an imperative aggregate such as collect_list (but not its FILTER, which is generated code), and a generator that Spark leaves out of a whole-stage stage. In the same contexts the codegen dispatcher now compiles its kernel over the expression's eval, so a dispatched tree that contains arrays_zip matches too, and array_repeat with a nullable count, whose eval skips the element when the count is NULL, is dispatched (on main too). The same holds for trunc / date_trunc with a non-literal format, whose eval returns NULL for an invalid format without evaluating the date (on main too, behind allowIncompatible). Each context has a fixture that fails without its fix.
• I switched each gate off one at a time and compared results with Spark on the locked dependencies. Three gates turned out to be unnecessary, and I removed them:
  • The NullType CASE gate on If / CaseWhen / Coalesce. Native CASE handles NullType results on DataFusion 55.1.
  • The collect_set gate. The mismatches were only element order.
  • The map_from_arrays literal-beside-per-row gate. The native map expands the scalar.
• collect_list keeps its NullType gate. Without a grouping key, the final merge reads partial state that never went through the planner's nullability cast. Plain nested types fail the same way on main (collect_list(array(id)) over more than one partition), and NullType-bearing nested elements fail on main even in one partition.
• A few guards were broader than needed and are narrowed:
  • coalesce no longer guards its last argument. A one-argument coalesce, which survives the optimizer only when NullPropagation and SimplifyConditionals are excluded, now serializes as its argument; it used to become a CASE with no WHEN, which native rejects (on main too).
  • array_append guards its item only when the array can be NULL.
• array(...) over a NullType argument now goes through the dispatcher unless the argument is a literal. The old check used foldable, but a foldable argument that constant folding did not reduce (element_at(array(NULL), 1)) still arrives per row, and make_array returned one row for the whole batch.
• IF is now serialized as a one-branch CASE WHEN. The native planner coerces CASE branches to one type but built IfExpr without that step, so two branches of one Spark type that differ in nested field nullability at native (native named_struct declares a field built from a literal non-nullable and one built from a column nullable) failed the projection's schema check whenever no row took the THEN branch. Since perf: optimize native CASE WHEN and IF (up to 12x faster) #6350 the CASE planner leaves a branch that already has the coerced type uncast, and evaluates IF and CASE through the same expression. A native CASE also names a merged struct's fields after its ELSE branch where Spark uses the first branch, which differ with case-insensitive analysis (IF(c, s, named_struct('A', id)) with s a struct<a>); array(...), array_append and the set ops then met two differently named structs. IF, CASE WHEN and coalesce now cast each branch of a struct-bearing type to the expression's own type. Comet's own casts are subject to spark.comet.expression.Cast.enabled, which the expressions guide now notes.
• The dispatcher now declares its output deep-nullable, like native constructors and Parquet columns. Two values of one Spark type used to reach native with different nested nullability depending on which side produced them, and kernels that compare their inputs' types (set ops, array_remove / array_position, CASE) rejected the pair. I found these with a sweep that pairs producers of different origin (native constructor, dispatcher, Parquet column, literal) of eight types inside about forty consumers that need one type, run on main and on this branch.
• That sweep also turned up two problems that exist on main:
  • arrays_overlap with a literal array beside a column compared only the first row, or panicked (arrays_overlap(array(id), array(1L)) over a multi-row batch). The kernel now expands the scalar side.
  • A dispatcher kernel whose generated code does not compile failed the query. Spark's own ElementAt codegen, for example, assigns an undeclared null flag when the index is an in-bounds literal and ANSI is off. The kernel now falls back to the expression's interpreted eval, as Spark's whole-stage codegen does.
• A review round found a third one on main: a nested array literal whose inner arrays are all empty, such as a folded array(cast(array() AS array<bigint>)), was built one list level deeper than its declared type, so array_union, arrays_overlap, IF, coalesce and = against it failed on main. This branch newly reached the same bug through arrays_overlap(transform(array(id), x -> array()), array(array())), whose dispatched side main declined. The literal builder now builds an empty child the way it builds a populated one, which keeps the declared depth and also fixes a deeper literal mixing the two (array(array(array()), array(array(array(1))))), which failed to build on main.
• Set ops now cast both sides to the set op's own element type rather than each side's: with case-insensitive analysis Spark accepts array(named_struct('a', x)) beside array(named_struct('A', x)) without a cast, and the native kernel's type check compares field names, so array_union over such a pair failed (on main too, for typed structs; newly reachable here for NullType-bearing dispatched ones). Every side is cast, even one whose Spark type already matches: a native CASE names a merged struct's fields after its ELSE branch where Spark uses the first branch, so array_union(array(IF(id < 0, s, named_struct('A', id))), array(s)) failed too (also on main). A cast that only relabels nested names and nullability is now Compatible, so these casts stay native.
• Native round-robin shuffle now falls back when a column it hashes contains a NullType. The native hasher has no NullType arm, so REPARTITION(n) over a dispatched map(id, NULL) failed (main declined that producer). Positional placement and columns past maxHashColumns are not hashed and stay native.
• The non-AQE DPP rewrite now honours the broadcast gate, so the DPP subquery keeps reusing the join's Spark exchange.
• The multi-batch shuffle test now reaches a second writer batch. It fails when the builder reset is removed; the previous version did not.

Coverage against main

I ran the same 6,901 queries on main (ce455f32d) and on this branch before the rebase: every query the composition sweep generates, with ANSI on and off, plus every query in the SQL fixtures.

main → this branch	queries
fallback → native	1,482
error → native or fallback	129
wrong answer → native	3
native → fallback	18

Of the 18, 12 are deliberate and 6 come from a gate that is wider than it needs to be:

• 8 are broadcast joins whose build side has a NullType directly under a struct. On main they run when the build side is one buffer and hang Arrow's appender when it is several. This branch skips coalescing those shapes so they cannot hang, but uncoalesced buffers cost each consuming task one IPC stream per buffer, which measured slower than Spark's broadcast, so the build side stays on Spark.
• 4 are non-deterministic arguments under a null guard (size, map_from_arrays, arrays_zip). Main builds the same guards and serializes the argument twice. Its answers matched Spark on the sweep's data, but the two copies do not see the same rows, as in the coalesce case under the known issues below.
• 6 are collect_list over a struct with a NullType field directly under it (collect_list(named_struct('a', id, 'b', NULL))), which main handles. The gate refuses any NullType under a struct to keep out the 16 shapes that fail on main, where that struct sits inside an array or another struct (collect_list(array(named_struct('a', id, 'b', NULL)))). Narrowing it to those shapes can be a follow-up.

After each rebase I ran the branch side again. Apart from new fixture queries, the only changes come from upstream: array_min on signed zeros, and float noise in the probe's exact comparison of stddev, corr and regr_*.

Known issues left as they are

All of these are present on main, and none is made worse here:

• Identical non-deterministic expressions that go through the codegen dispatcher share one kernel, so the second copy continues the first one's state. Two identical coalesce(IF(monotonically_increasing_id() % 2 = 0, id, NULL), -1) columns now reach this, where they used to evaluate the argument twice natively and were wrong in both columns.
• Ungrouped collect_list over a nested input with non-nullable fields fails in the final merge (above).
• array_append on Spark 3.x: Spark's generated code evaluates the item even for a NULL array, so an item that throws under ANSI raises in Spark and returns NULL natively.
• array_insert (which array_append becomes on Spark 4.0+): Spark's eval returns NULL for a NULL array or position before it evaluates the item, while its generated code and the native kernel evaluate all three, so where Spark uses eval an item that throws raises natively only.
• The non-AQE DPP rewrite follows the broadcast gate but not spark.comet.exec.broadcastExchange.enabled.
• array_join with allowIncompatible and a nullable non-deterministic nullReplacement serializes the replacement twice, like the null guards above, but has no NullGuard check.
• array_join with a nullable nullReplacement follows Spark's generated code, which skips the array when the replacement is NULL; Spark's interpreted eval evaluates the array first. So where Spark uses eval (for example the input of collect_list), an array argument that raises under ANSI raises in Spark and not natively. This is the same kind of eval/generated-code difference this PR handles for arrays_zip, array_repeat and trunc; I left array_join as it is on main rather than widen the PR further, and can open a follow-up issue for it.
• The map_from_arrays compatibility note still says a NULL key is not detected. Native now rejects it, with a different message.

Testing

On the final commit, with the native library built in release mode:

• Spark 4.1: CometSqlFileTestSuite (601), CometNullTypeCompositionSuite (28), CometArrayExpressionSuite (70), CometExpressionSuite (175), CometTemporalExpressionSuite (37), CometMapExpressionSuite (31), CometAggregateSuite (128), CometJoinSuite (56), CometExecSuite (154), CometJsonExpressionSuite (8), CometCodegenSuite (104), CometCodegenSourceSuite (67), CometShuffleSuite (48), DisableAQECometShuffleSuite (48), CometNativeShuffleSuite (58), CometNativeCastSuite (187), UtilsSuite (11), GenerateDocsSuite (4), the native unit tests for list literals and CASE, cargo clippy, cargo fmt and spotless:check.
• Spark 3.5: CometSqlFileTestSuite (601), CometNullTypeCompositionSuite (28), CometArrayExpressionSuite (69), CometExpressionSuite (170), CometAggregateSuite (126), CometCodegenSuite (103) and CometNativeShuffleSuite (52).

All passed, and each new regression test fails with its fix reverted. I also re-ran CometNullTypeColumnsBenchmark on the final commit. Each case runs the same plans as before, every arm returns the same rows, and every arm's speed relative to Spark is within 0.1x of the numbers I posted earlier, except the dispatcher-off arm of the 16K-row struct join, a run of about 45 ms that has ranged from 1.2x to 1.5x. I have not run Spark's own SQL suites locally. Since this touches the serde, the planner and the native shuffle writer, could a maintainer approve the workflow run and add run-all-spark-profiles and run-spark-4.1-tests?

If the change still looks too broad, or its fallbacks cost more than it is worth, I'm happy to close this PR or split it into smaller ones.

Assisted-by: Claude Code (claude-opus-5-5)

[sunchao]

Summary

• Prior state and problem: The dispatcher rejected NullType outputs, including nested null fields. Supporting them also exposes assumptions in native consumers, Arrow serialization and shuffle writers.
• Design approach: Extend the existing dispatcher and Arrow adapters, repair affected native paths, and retain Spark fallback where compatibility cannot be preserved.
• Correctness / compatibility analysis: The discussed shared-kernel concern remains a P2 blocker. The new routing at spark/src/main/scala/org/apache/comet/serde/conditional.scala:132 exposes it for previously correct inputs. Over one Parquet partition containing id=0..3, select coalesce(IF(id >= 0, monotonically_increasing_id(), CAST(NULL AS BIGINT)), -1L) twice. Spark returns (0,0), (1,1), (2,2), (3,3). An exact-head dispatcher component reproduction returns 0..3 for the first column and 4..7 for the second because both serialized expressions share mutable kernel state. A native control reproducing the base serde's expression tree returns 0..3 for both columns. This therefore materially worsens the existing issue. No additional introduced P1/P2 issues were found within this review, and no duplicate inline finding is proposed.
• Key design decisions: Keeping NullType inputs and duplicate Arrow struct field names gated limits the admission boundary. Reusing the dispatcher and shared schema helpers avoids another execution mechanism. Tracking interpreted evaluation contexts addresses real differences between Spark's eval and generated code.
• Implementation sketch: Add NullVector output handling, align nested schemas and conditional branches, repair null-map IPC schemas, reset shuffle builders, and correct native struct, list-literal and array handling. Regression coverage includes expression composition, operators and Arrow boundaries.
• Behavioral changes worth calling out: More expressions remain in Comet through JVM dispatch, while unsafe broadcast, hashing and aggregate shapes fall back. These choices affect execution cost. The benchmark covers dispatcher and broadcast behavior, but I did not independently rerun its performance measurements. No separate reproducible P1/P2 performance or abstraction concern was established.
• Suggested improvements: Resolve the existing state-sharing regression before merging. Give each expression occurrence independent mutable kernel state while retaining compiled-code reuse, or keep the affected nondeterministic expressions on Spark. Add the two-column counterexample above as a regression test.

Reviewed the entire requested base-relative diff, including all seven branch commits and the Iceberg test deletion, from 11a27c36713779f8fd297e0dd72ac4911dd3e04a to 92d6cfc845d98071d2967d63156edd24ee10165e. The PR is open and not a draft. Existing non-Copilot reviews, comments and threads were considered.

Routed skills: review-comet-pr, review-comet-expression-pr, review-comet-ffi-pr, review-comet-memory-pr, review-comet-shuffle-pr and review-comet-iceberg-write-pr. Relevant Spark source comparisons covered 3.4.3, 3.5.9, 4.0.4, 4.1.3 and 4.2.0.

Exact-head CI: the label check passed. Comet CI and CodeQL both report action_required. There is no passing exact-head build/test verdict.

Validation: 53 focused Rust tests and the base-expression control passed. Fresh Spark 4.1.3 component probes exercised NullType outputs, interpreted array_repeat and nested null-map IPC, and reproduced shared dispatcher state. The generator and dispatcher sources matched the reviewed checkout, with dependency extracts/stubs used for the harness. This was not full Comet/JNI execution. Maven bootstrap failed with UnknownHostException: repo.maven.apache.org, so full JVM, Spark SQL, Python and Iceberg integration suites and benchmarks were not run. Disposable project tests were removed and the checkout is clean.

[grorge123]

@sunchao thank you, you're right. Routing that coalesce to the dispatcher made your two-column query worse than main, so "none is made worse here" in my previous comment was wrong for it. The branch is rebased onto today's main (351facd26), and the fix is in the last commit:

• Each occurrence of a dispatched expression that contains a Catalyst Nondeterministic node (monotonically_increasing_id, rand, uuid, ...) now gets its own kernel. Before serializing such a tree, the serde wraps it in DispatchOccurrence with a per-occurrence id, and the dispatcher strips the wrapper before compiling. The compiled class is still shared, but each occurrence gets its own kernel instance and state. Your query is now a regression test (CometCodegenSuite, "identical non-deterministic dispatched expressions keep their own state"), together with two identical map(monotonically_increasing_id(), NULL) columns. Both fail without the change, over 64 rows in batches of 8.

• While the fix was under review, two more dispatcher problems in the same area came up. Both are also on main, but this branch makes them reachable through more shapes, so the dispatcher now refuses them and the operator falls back to Spark:

  • A non-deterministic user function (an asNondeterministic() ScalaUDF, or a non-deterministic Invoke, StaticInvoke or V2 function) or a reflected method (reflect, whose static target can hold JVM-wide state). Spark keeps the state in one object that every call shares and advances it row by row. Kernels that evaluate one expression over a whole batch either copy that object or consume it column by column. For example, nextCount('x'), nextCount('y') with a stateful Java UDF returned (0, 0) where Spark returns (0, 1). These UDFs ran through the dispatcher on main and now fall back, which scala_java_udfs.md documents.
  • A scalar subquery inside a dispatched expression (my_udf((SELECT max(x) FROM t))). The native plan is serialized before the operator starts its subqueries, so the shipped copy failed with "Subquery ... has not finished". A subquery beside the dispatched expression is unaffected.

  Each refusal has a test that fails without it.

• fix: cast native IF branches to a common type like CASE WHEN #6458 landed in the meantime and reconciles native IF branches itself, naming struct fields after the THEN branch. So I dropped my change that serialized IF as a one-branch CASE WHEN, and the struct branch casts now apply to CASE WHEN and coalesce only. The IF fixtures in if_expr.sql now run against the upstream fix.

Smaller changes in the same commit: a date_trunc fixture for the interpreted-format gate, the hash.sql NullType witness split into one query per function, the round-robin NullType rule in the contributor shuffle guides, and the IF / coalesce audit entries.

Known issues, updated

The first item in my previous list (identical non-deterministic expressions sharing a kernel) is fixed; the rest are unchanged. One more, also on main: Spark skips the right operand of a binary expression when the left one is NULL, but native evaluation computes both, so a stateful right operand advances on rows Spark skips. IF(id = 0, NULL, id) + monotonically_increasing_id() over ids 0..3 in one partition returns NULL, 2, 4, 6 where Spark returns NULL, 1, 3, 5, and arrays_overlap and array_contains behave the same way. This branch adds one more shape that reaches it (an array<void> left operand produced by the dispatcher, which main ran in Spark). Fixing it means changing how native binary expressions evaluate, which is outside this PR.

Testing

With the native library built in release mode, on the commit before its last two changes (the reflect refusal and today's rebase):

• Spark 4.1: CometSqlFileTestSuite (608), CometNullTypeCompositionSuite (28), CometArrayExpressionSuite (70), CometExpressionSuite (175), CometTemporalExpressionSuite (37), CometMapExpressionSuite (32), CometAggregateSuite (128), CometJoinSuite (65), CometExecSuite (154), CometJsonExpressionSuite (8), CometCodegenSuite (108), CometCodegenSourceSuite (67), CometShuffleSuite (48), DisableAQECometShuffleSuite (48), CometNativeShuffleSuite (58), CometNativePositionalRoundRobinSuite (11), CometNativeCastSuite (187), UtilsSuite (11), GenerateDocsSuite (4), the native unit tests for list literals and CASE, cargo clippy, cargo fmt and spotless:check.
• Spark 3.5: CometSqlFileTestSuite (608), CometNullTypeCompositionSuite (28), CometArrayExpressionSuite (69), CometExpressionSuite (170), CometAggregateSuite (126), CometCodegenSuite (107) and CometNativeShuffleSuite (52).
• Spark 3.4: CometCodegenSuite (107).

After the reflect refusal: Spark 4.1 CometCodegenSuite (108), CometCodegenSourceSuite (67), CometNullTypeCompositionSuite (28) and CometSqlFileTestSuite (608), and CometCodegenSuite on Spark 3.5 and 3.4 (107 each). Then on the final commit, rebased onto 351facd26: Spark 4.1 CometCodegenSuite (108), CometSqlFileTestSuite (610), CometNativeCastSuite (190), CometNullTypeCompositionSuite (28), GenerateDocsSuite (4) and spotless:check, and Spark 3.5 CometCodegenSuite (107).

All passed, and each new test fails with its fix reverted. CI on the previous head is still waiting for workflow approval.

Assisted-by: Claude Code (claude-opus-5-5)

[sunchao]

Summary

• Prior state and problem: Dispatcher outputs containing NullType forced their operators back to Spark. Admitting them also exposes downstream assumptions about nested nullability, scalar cardinality and Arrow vectors.
• Design approach: Support NullType outputs while retaining input restrictions, normalize Arrow declarations, and route unsupported compositions through Spark evaluation or fallback.
• Correctness / compatibility analysis: One existing P2 concern remains unresolved, as described in the author's comment. For IDs 0–3 in one partition, arrays_overlap(IF(id = 0, CAST(NULL AS ARRAY<VOID>), transform(array(id), x -> NULL)), array_repeat(NULL, CAST(monotonically_increasing_id() AS INT))) returns [NULL, false, NULL, NULL] in Spark 4.1.3. The exact-head native expression reproduction returns [NULL, NULL, NULL, NULL]: eager evaluation advances the right-hand counter on the null-left row. Compiling the base and head dispatcher gates independently confirms that the array<void> producer is rejected at the base and admitted at this head. Thus this PR makes the wrong-result path newly reachable. Full Comet SQL execution was not available locally, as detailed below.
• Key design decisions: Centralizing IPC schema repair and null-guard checks limits duplicated policy. Per-occurrence dispatcher state fixes the previously discussed shared-counter problem, which the component harness confirms. The broadcast gate avoids the proposed uncoalesced IPC-per-buffer path. No separate reproducible performance or abstraction blocker was found.
• Implementation sketch: The changes span codegen output handling, expression routing, Arrow IPC writers, broadcast selection, native scalar/list handling and shuffle builder reuse, with SQL composition coverage and a microbenchmark.
• Behavioral changes worth calling out: More null-bearing constructors stay in Comet. NullType inputs and selected hash, aggregation and broadcast shapes still fall back. Dispatcher execution now distinguishes Spark's interpreted and generated evaluation contexts.
• Suggested improvements: Address the existing binary-evaluation regression before merging. A targeted fallback or whole-expression dispatch for affected nullable-left/stateful-right compositions can preserve Spark's evaluation sequence without requiring a broad native rewrite. Add the witness above over a single-partition Parquet input to the differential suite.

Reviewed all seven commits and all 99 changed files against base 351facd26ca0febbfe0a4925549e3b89d22747d5, at full head 51e8d6bbfab21cb3fc730ac5ee6f5de4cb50f571. The PR is open and non-draft. Existing discussions were reviewed, excluding Copilot. Routed skills: review-comet-pr, review-comet-expression-pr, review-comet-ffi-pr, review-comet-memory-pr and review-comet-shuffle-pr. Relevant Spark source comparisons covered 3.4.3, 3.5.9, 4.0.4, 4.1.3 and 4.2.0.

Exact-head CI: the label check passed. Comet CI and CodeQL both report action_required. There is no passing build/test verdict for this head.

Validation: 40 focused repository native tests passed, covering arrays_overlap, struct projection/null propagation, CSV and null-bearing shuffle paths. Both bounded reproduction checks passed. Spark 4.1.3 reference execution and dispatcher/Arrow IPC component probes also passed. The JVM harness used exact production sources, extracted helpers and allocator/type stubs, so it does not validate the full JNI pipeline. Maven bootstrap failed with UnknownHostException: repo.maven.apache.org; full JVM/JNI, Spark SQL and Python suites, the native core planner tests and performance benchmarks were not run. Disposable project tests were removed and the checkout is clean.

No additional, non-duplicate introduced P1/P2 issues were found within this review. The findings list is empty because the substantiated remaining concern is already discussed. Recommended review state: request changes for that existing P2 compatibility regression.

[grorge123]

@sunchao thank you. You're right that the eager-argument divergence is newly reachable here even though the native evaluation behind it is on main, so I've fixed it in this PR. The branch is rebased onto 93d1189be, and the fix is in the new last commit.

The rule does not try to model which arguments Spark skips. Several attempts at that kept missing cases: a divisor evaluated first, ln serializing its argument twice, and kernels that serialization builds itself. Instead, it relies on one fact about main. Main's dispatcher refused every NullType result, so an operator whose native plan runs such a kernel fell back to Spark, and so did the native operators above it up to a shuffle. An operator now falls back when both of these hold:

• it evaluates a stateful expression (any Catalyst Nondeterministic node such as monotonically_increasing_id, rand or shuffle, or a non-deterministic Scala UDF, invoked method or V2 function, whatever its arguments) that the dispatcher does not run as a whole; and
• its native plan runs a dispatcher kernel that computes a NullType value, either as its result or anywhere in the tree it runs. A kernel whose payload cannot be read counts as one.

For a kernel with a NullType result, this puts the operator back where main ran it, in Spark. Checking the tree inside a kernel goes further than main did, so a few shapes that main ran natively now fall back too (listed under the costs below). The kernels are read from the serialized plans. That covers:

• the operator's own native block;
• the plan below broadcast, union and coalesce sinks, which exist only over native plans;
• TakeOrderedAndProject's sort keys and projection, which it serializes when it runs;
• the partitioning keys of a shuffle below the operator.

A column the dispatcher runs whole (map(monotonically_increasing_id(), NULL)) stays native, because its kernel evaluates the whole Spark expression, with its own state, row by row. A tree that serialization rebuilds (a decimal promotion) loses that mark and falls back, which only adds fallback. spark_partition_id() and the input-file expressions don't count, because they hold no row state.

Review also found one non-NullType path of the same kind. Main dispatched a relabel-only array<date> cast, but this PR made it Compatible, which moved a non-deterministic child onto native evaluation. That shortcut now applies only to a deterministic child.

Your witness and the other shapes are in misc/nulltype_stateful_argument.sql and misc/nulltype_stateful_shuffle.sql (a seeded shuffle, Spark 4.0+), on a one-partition Parquet table. They are SQL file tests, so each query's answer is compared against Spark, and each shape that should fall back also checks the reason. The other shapes are:

• a NullType value in another column;
• a NullType value in a lower projection;
• below a broadcast join and below a UNION ALL;
• DIV;
• ln;
• four TakeOrderedAndProject shapes;
• DISTRIBUTE BY;
• a typed kernel holding a NullType value;
• native controls.

CometCodegenSuite covers kernels that serialization builds itself (a folded map('k', NULL) literal, a CheckOverflow decimal) with the default optimizer. Locally, I removed each part of the rule in turn, and a fixture failed every time. Those runs are not committed.

What the rule costs. It is conservative, so a few shapes leave native execution where they produced correct results:

• a typed kernel main already dispatched whole, whose tree holds an intermediate NullType value, next to a stateful expression;
• a set op over NullType elements, which this PR now dispatches, next to one.

Results stay correct in both. The composition suite's non-deterministic sweep falls back more often, so its floor of natively executed cases drops from 110 to 60. Most of the cases it loses run a kernel with a NullType result beside the stateful producer, which main ran in Spark; a few are the typed-kernel shape above (the sweep's filter(array(CAST(NULL AS int)), ...) producer).

A correction to my earlier comment. I wrote that the gate and the coalescer's bypass share one predicate and that UtilsSuite runs the real appender over every shape. Neither is accurate. There are two matching predicates, Utils.hasNullTypeUnderStruct on Spark types and Utils.hasNullDirectlyUnderStruct on the Arrow schema. UtilsSuite appends only the shapes that coalesce, while the bypassed shapes never reach the appender, so it cannot tell when an Arrow release fixes the hang.

Testing

With the native library built in release mode, on the commit before its last change (the shuffle detection):

• Spark 4.1: CometSqlFileTestSuite (613), CometNullTypeCompositionSuite (28), CometCodegenSuite (111), CometNativeCastSuite (190), CometCodegenSourceSuite (67), CometExpressionSuite (175), CometArrayExpressionSuite (70), CometMapExpressionSuite (32), CometTemporalExpressionSuite (37), CometJsonExpressionSuite (8), CometAggregateSuite (128), CometExecSuite (154), CometJoinSuite (65), CometNativeShuffleSuite (58), CometShuffleSuite (48), DisableAQECometShuffleSuite (48), CometNativePositionalRoundRobinSuite (11), UtilsSuite (11), GenerateDocsSuite (4), the datafusion-comet native unit tests (590) and the native CASE tests, cargo fmt, cargo clippy and spotless:check.
• Spark 3.5: CometSqlFileTestSuite (613), CometNullTypeCompositionSuite (28) and CometCodegenSuite (110).
• Spark 3.4: CometCodegenSuite (110).

On the final commit: Spark 4.1 CometSqlFileTestSuite (614), CometNullTypeCompositionSuite (28), CometCodegenSuite (111), CometExecSuite (154) and CometExpressionSuite (175); Spark 3.5 CometSqlFileTestSuite (614), CometNullTypeCompositionSuite (28) and CometCodegenSuite (110); Spark 3.4 CometCodegenSuite (110).

All passed. Spark's own SQL suites have not been run locally.

Assisted-by: Claude Code (claude-opus-5-5)

[sunchao]

Summary

• Prior state and problem: The dispatcher rejected NullType outputs, forcing otherwise supported projections back to Spark. Admitting them exposes assumptions in native expressions, Arrow schemas and shuffle writers.
• Design approach: Extend the existing dispatcher’s output handling, repair affected native paths, and preserve Spark evaluation or fallback for unsupported compositions.
• Correctness / compatibility analysis: No introduced P1/P2 issues found within this review. Relevant Spark source comparisons covered 3.4.3, 3.5.9, 4.0.4, 4.1.3 and 4.2.0. Source inspection and component checks support resolution of the previously discussed blockers, including independent dispatcher state and the final commit’s fallback for stateful evaluation beside NullType kernels. No substantiated existing P1/P2 concern remains unresolved.
• Key design decisions: Keep NullType inputs restricted, centralize Arrow schema repair and argument alignment, and isolate mutable state per expression occurrence while retaining compiled-class reuse. The conservative operator fallback avoids modelling every native expression’s evaluation order.
• Implementation sketch: Add NullVector output support, normalize nested declarations, repair map-key IPC schemas, correct native scalar/list/CSV handling, and reset NullType shuffle builders between batches. SQL, composition and boundary tests cover the expanded execution paths.
• Behavioral changes worth calling out: More null-bearing expressions remain in Comet. Unsafe compositions still dispatch or fall back, and empty CSV renderings become NULL. No reproducible P1/P2 performance or abstraction regression was established. Performance benchmarks were inspected but not independently run.
• Suggested improvements: No additional P1/P2 changes requested.

Reviewed full head e32df05a9fee3d4cfb71ac3af752951c64915548 using supplied base 432523fb59efe75b208ae5688eef6012c208a4d6. Coverage includes the entire GitHub PR diff: all eight branch commits and 101 changed files, from merge-base 93d1189beb2a2da1c4e2663de131c61713b51e48. The PR remains open and non-draft. Existing non-Copilot reviews, issue comments, inline comments and threads were considered.

Routed skills: review-comet-pr, review-comet-expression-pr, review-comet-ffi-pr, review-comet-memory-pr and review-comet-shuffle-pr.

Exact-head CI: the label check passed. No build/test check results are reported for this SHA. Combined commit status is pending with zero status entries.

Validation: 40 focused repository native tests passed. Fresh Spark 4.1.3 and Scala/Arrow component probes passed for NullType outputs, independent expression state, interpreted array_repeat, nested null-map IPC and the extracted fallback predicates. The JVM harness used exact production sources and extracted helpers with allocator/type stubs; it does not validate the full Comet/JNI pipeline. Maven bootstrap failed with UnknownHostException: repo.maven.apache.org. Full JVM/JNI, Spark SQL and Python suites, native core planner tests and benchmarks were not run. The checkout remains clean.

[grorge123]

Rebased onto 569eaa59d to resolve the conflicts with main. The only change in behavior comes from #5750, which makes floating-point array_union Incompatible (falling back to Spark) on every Spark version except 4.2.0. This PR mixes CodegenDispatchFallback into CometArrayUnion for its NullType-element case, and that mixin would also have sent the floating-point case through the dispatcher instead of falling back. So CodegenDispatchFallback now has dispatchesIncompatible (default true). CometArrayUnion sets it to false: a NullType-element side still runs through the dispatcher, and a floating-point element type falls back as #5750 intends. QueryPlanSerde skips the dispatcher for an Incompatible result when the flag is false, and GenerateDocs documents those reasons as falling back. #5750's own fallback test in CometArrayExpressionSuite fails when the flag is set back to true. The other conflicts were in SQL fixtures and expressions.md, where I kept main's content and re-added this PR's queries and notes. #4715 now builds a nested list literal whose inner lists are all empty at its declared depth, the same fix this PR had made, so the planner change here is down to its unit test.

Tested locally on the rebased head, with the native library built in release mode:

• cargo fmt, cargo clippy, spotless:check, and the datafusion-comet native unit tests (592).
• Spark 4.1: CometSqlFileTestSuite (620), CometNullTypeCompositionSuite (28), CometCodegenSuite (111), CometNativeCastSuite (190), CometCodegenSourceSuite (67), CometExpressionSuite (176), CometArrayExpressionSuite (75), CometMapExpressionSuite (32), CometTemporalExpressionSuite (37), CometJsonExpressionSuite (8), CometAggregateSuite (128), CometExecSuite (154), CometJoinSuite (65), CometNativeShuffleSuite (65), CometShuffleSuite (48), DisableAQECometShuffleSuite (48), CometNativePositionalRoundRobinSuite (11), UtilsSuite (11), GenerateDocsSuite (4), CometFloatSemanticsSuite (412), SqlFileTestParserSuite (13), CometRangeExecSuite (14), and CometLiteralSuite (4).
• Spark 3.5: CometSqlFileTestSuite (620), CometNullTypeCompositionSuite (28), and CometCodegenSuite (110).
• Spark 3.4: CometCodegenSuite (110).

All passed. Spark's own SQL suites have not been run locally.

Assisted-by: Claude Code (claude-opus-5-5)

[sunchao]

Summary

• Prior state and problem: The dispatcher rejected NullType outputs, forcing otherwise supported operators back to Spark. Admitting these outputs exposes assumptions in native expressions, Arrow schemas and shuffle writers.
• Design approach: Extend the existing dispatcher, repair affected downstream paths, and preserve Spark evaluation or fallback for unsupported compositions.
• Correctness / compatibility analysis: No introduced P1/P2 issues found within this review. Source inspection and focused component checks support resolution of the previously discussed blockers, including shared dispatcher counters and stateful evaluation beside NullType kernels. No substantiated existing P1/P2 concern remains unresolved. Relevant Spark source comparisons covered 3.4.3, 3.5.9, 4.0.4, 4.1.3 and 4.2.0. The rebased array_union routing preserves the upstream floating-point fallback.
• Key design decisions: Keep NullType inputs restricted, centralize Arrow schema repair and argument alignment, and isolate mutable state per expression occurrence while retaining compiled-class reuse. The conservative operator fallback avoids modelling every native expression’s evaluation order. No separate P1/P2 design or abstraction concern was substantiated.
• Implementation sketch: Add NullVector output handling, normalize nested declarations, repair map-key IPC schemas, correct native scalar/struct/CSV handling, and reset NullType shuffle builders between batches. Regression coverage spans expressions, operators and Arrow boundaries.
• Behavioral changes worth calling out: Compared touched paths with release branch-1.1. Intended changes include expanded null-bearing dispatch, corrected evaluation and null handling, and stricter fallback for unsafe user-state, subquery, broadcast, hash, aggregate and shuffle shapes. No unintended release regression was substantiated. Performance benchmarks were inspected but not independently run, and no reproducible P1/P2 performance regression was established.
• Suggested improvements: No additional P1/P2 changes requested.

Reviewed full SHA 90d02ebcebb898a4c6afef1a153116d9f7a29e64, including all eight PR commits and the entire 104-file diff against supplied base 8c93fbb735a8eb6b3b19cde42dc3829825ba8fe4. This includes every file in the merge-base PR diff. The PR remains open and non-draft. Existing reviews, issue comments, inline comments and threads were considered, excluding Copilot.

Routed skills: review-comet-pr, review-comet-expression-pr, review-comet-ffi-pr, review-comet-memory-pr and review-comet-shuffle-pr.

Exact-head CI: the label check passed. Comet CI and CodeQL report action_required. Combined commit status is pending with zero status entries. There is no passing exact-head build/test verdict.

Validation: 40 focused repository native tests and one disposable mixed scalar/column arrays_overlap probe passed. Fresh Spark 4.1.3 and Scala/Arrow component probes passed for NullType outputs, independent dispatcher state, interpreted array_repeat, nested null-map IPC and the extracted fallback predicates. The JVM harness used verified current production sources and extracted helpers with dependency stubs, so it does not validate the full Comet/JNI pipeline. Maven bootstrap failed with UnknownHostException: repo.maven.apache.org. Full JVM/JNI, Spark SQL and Python suites, native core planner tests and benchmarks were not run. Disposable project tests were removed and the checkout is clean.

[mbutrovich]

@grorge123 thanks for all the work on this, and for offering to split it. I'd like to take you up on that. The change #5525 asks for is the first commit (56436fe), at 209 lines. The PR is now 5,334 lines, and I think the automated review rounds are a big part of why. Each round raised more edge cases, many of them bugs that already exist on main without NullType, and each one was fixed here instead of getting its own issue and PR. That's on the review process as much as on this PR. A finding that reproduces on main belongs in a separate issue, not in the scope of the PR where it was found. Could we break this one into PRs that each fix one thing and carry their own tests, landing in this order?

1. Fixes for bugs on main, each in its own PR with an issue. These are the native to_csv, arrays_overlap and GetStructField fixes, the dispatcher's interpreted eval fallback when generated code doesn't compile, per-occurrence kernel state together with refusing non-deterministic user functions, reflect and subqueries, tracking where Spark evaluates through eval (with the arrays_zip, array_repeat and trunc changes), and the struct-branch casts with the one-argument coalesce fix.
2. The non-deterministic null guards. #5781 tracks that bug, and #5867 by @dwsmith1983 fixes it by dispatching those shapes where this PR falls back to Spark. Could the two of you agree on one approach in #5781 before either PR lands?
3. This PR, rebased on the above. It would keep the gate and emitter change plus the fixes that NullType outputs newly make reachable: the IPC map-key repair, the broadcast gate, the shuffle Null builders, the round-robin hash fallback and the native kernel gates. Each shape can be covered with SQL file fixtures, which is where expression coverage is moving (#6615).
4. CometNullTypeColumnsBenchmark and CometNullTypeCompositionSuite as follow-ups.

The comments below are on parts that would stay in this PR or move to the follow-ups.

[mbutrovich]

Should CometNullTypeCompositionSuite run on every PR? You measured about 8 minutes for it locally, and this adds it to the expressions job for every Spark profile in the Linux matrix, plus macOS in pr_build_macos.yml. The SQL file fixtures already cover each gated shape. Could the composition sweep run in the nightly ci.yml schedule instead?

[grorge123]

Agreed. 267b5ce takes the suite out of this PR, and with it the workflow entries, since dev/ci/check-suites.py requires every suite in both workflow files. It comes back as a follow-up, where I'll settle how often it runs along with moving it onto the fuzz matrix you suggest on the suite itself.

[mbutrovich]

This second case ([[[]], [[[1]]]] with every declared field non-nullable, over Int32) has the same input as test_literal_to_list_with_empty_nested_children_non_nullable, which #4715 added to main along with the fix. Could we drop it and keep only the [[], null] case above, which the tests on main don't cover?

[grorge123]

Agreed, 267b5ce drops it and keeps only the [[], null] case.

[mbutrovich]

This deserializes each dispatcher kernel's closure bytes during planning to recover its expression, and runsNullTypeKernel walks every message field of the serialized operator to find the kernels. CometScalaUDF.emitJvmCodegenDispatch already holds the bound expression when it serializes it (CometScalaUDF.scala#L141). Could it record there whether the tree computes a NullType value, so this rule reads that instead of round-tripping through the closure serializer?

The rule exists because of a bug on main: native evaluation computes the right operand of a binary expression even when the left one is NULL, so a stateful operand advances on rows Spark skips. Is there an issue for that? If it's the same problem as #6006, could the scaladoc reference that issue, so it's clear when this rule can be removed? If not, could you file one?

[grorge123]

Good idea. I'll record whether a kernel computes a NullType value while emitJvmCodegenDispatch serializes it, and carry that up to the operator, so the rule no longer deserializes closures or walks every message field. It still has to see expressions the serializer creates, TakeOrderedAndProject's deferred sort keys and projection, and kernels below broadcast, union and coalesce sinks and in shuffle keys, so I'll keep those covered rather than narrow the gate.

On the issue: it's related to #6006 but not the same bug. #6006 tracks expressions that raise an error on rows Spark skips. Here a stateful operand advances on those rows and the answer is wrong without any error, e.g. IF(id = 0, CAST(NULL AS BIGINT), id) + monotonically_increasing_id(). I'll confirm that on main, file it as its own issue linked to #6006, and reference it from the scaladoc. The rule only shields the shapes this PR newly admits; it doesn't fix the typed case.

[mbutrovich]

What do you think about moving this sweep in with the fuzz suites? Only its producers go through the codegen dispatcher. Most of what it found was in native kernels, the shuffle writer and broadcast IPC. Its physical profiles also overlap the shuffle mode and native columnar-to-row matrix that CometFuzzTestBase already runs every test under. Could it reuse that matrix and add only the settings it needs beyond it (two-row batches, the sort-based writer with spills, AQE)? The gate test in this suite also checks the same dispatch decisions as the expect_dispatch SQL fixtures this PR adds. Could those checks live only in the fixtures?

[grorge123]

Makes sense. The suite is out of this PR as of 267b5ce. In the follow-up I'll move the sweep into the fuzz suites, reuse CometFuzzTestBase's matrix and add only the two-row batches, the spilling sort-based writer and AQE. Dispatch decisions will stay in the expect_dispatch fixtures, and the checks that each serde publishes its reasons will move to focused serde tests, since the fixtures don't cover those.

[sunchao]

Summary

• Prior state and problem: Dispatcher outputs containing NullType forced operators back to Spark. Admitting those outputs exposes downstream assumptions about Arrow schemas, nested nullability, batch cardinality and expression evaluation order.
• Design approach: Add NullType output support, centralize IPC schema repair, fix affected native operations and retain Spark fallback for unsupported compositions.
• Correctness / compatibility analysis: No introduced P1/P2 issues found within this review. Checked relevant Spark sources across 3.4.3, 3.5.9, 4.0.4, 4.1.3 and 4.2.0. Compared touched paths with branch-1.1 at 7b7eec69282a7abc33a0e8e687ca594c72f140ec. Expanded dispatch, corrected null handling and conservative fallback are intended changes. No unintended release regression was identified. Previously reported correctness concerns are addressed in the current code.
• Key design decisions: NullType inputs remain restricted. Stateful dispatcher occurrences retain independent state, interpreted contexts preserve Spark’s evaluation order, and unsafe broadcast and hashing shapes fall back. Shared helpers consolidate schema and null-guard policy. Existing maintainer concerns about planning complexity remain open.
• Implementation sketch: The changes span dispatcher generation, expression serialization, Arrow IPC, broadcast selection, shuffle Null builders and native expression handling, with SQL fixtures, composition coverage, documentation and benchmark instrumentation.
• Behavioral changes worth calling out: More NullType-producing expressions stay in Comet, while broadcast builds with direct Null struct/map fields and affected round-robin hashing retain fallback. Benchmark timing and memory effects were not measured during this review, so no quantified performance conclusion is offered.
• Suggested improvements: No additional P1/P2 changes requested. The existing requests to split the PR, simplify planning metadata handling and reconsider composition-suite placement remain open. This review adds no duplicate findings.

Reviewed the full 104-file diff, including all eight branch commits, at 90d02ebcebb898a4c6afef1a153116d9f7a29e64 against 8c93fbb735a8eb6b3b19cde42dc3829825ba8fe4. The PR remains open and non-draft. Read the existing non-Copilot discussion and applied review-comet-pr, review-comet-expression-pr, review-comet-ffi-pr, review-comet-memory-pr and review-comet-shuffle-pr.

Validation: 41 focused native tests passed. Fresh Spark 4.1.3 component probes passed for NullType outputs, independent stateful dispatcher calls, interpreted null short-circuiting and nested null-key map IPC. These probes used verified current production sources with helper stubs. Maven bootstrap failed with UnknownHostException: repo.maven.apache.org, so full JVM/JNI, Spark SQL, Python, native planner and benchmark suites were not run.

Exact-head CI: Comet CI and CodeQL report action_required. Labeling passed. Combined status is pending with zero statuses, so there is no passing exact-head build/test verdict. The checkout is unchanged and no GitHub state was modified.

[grorge123]

@mbutrovich thanks, that's fair, and I agree that a finding which reproduces on main belongs in its own issue. I'll split the PR as you suggest. Here is the plan; let me know if you'd group anything differently.

1. Fixes for bugs on main, each with an issue and its own PR:

   • the native to_csv, arrays_overlap and GetStructField fixes;
   • the dispatcher's interpreted eval fallback when generated code doesn't compile;
   • refusing non-deterministic user functions, reflect and subqueries in the dispatcher. @dwsmith1983 has taken the per-occurrence kernel state in Identical nondeterministic expressions share one codegen dispatcher kernel and continue each other's state #6711 / fix: give each nondeterministic dispatched expression its own kernel #6712, so I'll drop my duplicate implementation and build on that;
   • tracking where Spark evaluates through eval, with the arrays_zip, array_repeat and trunc changes. It goes after the eval fallback, which it reuses;
   • the CASE WHEN / coalesce branch casts and the one-argument coalesce, and the stateful coalesce guard, which serializes its non-final children twice;
   • the set-op argument alignment with the relabel-only cast, and the array_repeat item nullability, if each reproduces with typed inputs.

   I'll reproduce each one on main before opening its issue. Anything that only fails with a NullType output stays in this PR, and any other defect I find to be reachable on main gets its own issue.

2. The null guards: I agree with @dwsmith1983's proposal in size, arrays_zip, map_from_arrays and array_append return wrong answers for a nondeterministic child #5781 to take the dispatcher route from fix: dispatch map lookups with normalized keys and nondeterministic null-guarded children #5867 for size, array_append, arrays_zip and map_from_arrays, so I'll drop those checks from this PR. Narrowing which guards are emitted can be a follow-up.

3. This PR, rebased on the above. It keeps the new admission boundary for NullType outputs and the protections it needs, covered by SQL file fixtures. I've replied to your inline comments on those parts.

4. CometNullTypeColumnsBenchmark and CometNullTypeCompositionSuite as follow-ups.

I've pushed one commit for your comments on the tests (267b5ce): CometNullTypeCompositionSuite and CometNullTypeColumnsBenchmark are out of this PR and its CI workflows, and the duplicate planner case is gone. Otherwise I'll leave this branch as it is until the first group lands, rather than rebasing it again in the meantime.

Assisted-by: Claude Code (claude-opus-5-5)

[sunchao]

Summary

• Prior state and problem: The dispatcher rejected NullType outputs, forcing constructors such as map() into Spark. Admitting these outputs exposes downstream schema, cardinality and evaluation-order assumptions.
• Design approach: Support output-side NullVectors, normalize nested nullability, repair IPC map-key schemas and retain fallback where native execution cannot preserve Spark behavior.
• Correctness / compatibility analysis: No introduced P1/P2 issues found within this review. Compared relevant Spark implementations across 3.4.3, 3.5.9, 4.0.4, 4.1.3 and 4.2.0, including generated versus interpreted evaluation. Previously raised correctness cases are addressed by the current implementation. Comparison with branch-1.1 at 7b7eec69282a7abc33a0e8e687ca594c72f140ec identified intended routing and correctness changes, with no established unintended regression.
• Key design decisions: Keep NullType inputs unsupported, isolate nondeterministic dispatcher occurrences and centralize null guards and IPC repair. The repaired IPC root shares existing vectors. The maintainer’s requests to simplify planning metadata and separate prerequisite fixes remain documented. No reproducible P1/P2 performance regression was established, and performance was not benchmarked.
• Implementation sketch: Changes span dispatcher code generation, expression serde, operator gates, native struct/array/CSV handling, shuffle builders and IPC callers, with focused Scala, Rust, SQL and Python coverage.
• Behavioral changes worth calling out: More NullType-producing expressions can remain in Comet. Unsafe stateful compositions, shared-state user functions, dispatched subqueries and affected broadcast shapes retain Spark fallback. The final commit removes the composition sweep and benchmark while retaining focused regression fixtures.
• Suggested improvements: No additional request meets the specified P1/P2 bar.

Reviewed the entire 99-file base-relative diff, including prerequisites, at 267b5ce6633579558deabd4dd75d2be8c25f2f76 against 8c93fbb735a8eb6b3b19cde42dc3829825ba8fe4. The PR remains open and non-draft. Existing reviews, issue comments, inline comments and threads were considered, excluding Copilot.

Routed skills: review-comet-pr, review-comet-expression-pr, review-comet-ffi-pr, review-comet-memory-pr and review-comet-shuffle-pr.

Exact-head CI: Only the labeling workflow completed successfully. There is no build/test verdict. Combined commit status is pending with zero statuses.

Validation: All 40 focused native tests passed. Spark 4.1.3 component probes passed for dispatcher state isolation, NullType outputs, interpreted array_repeat and nested map-key IPC. These probes use verified production source extracts with surrounding stubs, so they do not validate full JVM/JNI execution. Maven bootstrap failed with UnknownHostException: repo.maven.apache.org. Full JVM/JNI, Spark SQL, Python integration, native-core planner tests and benchmarks were not run.