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.

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.

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, 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. array<null> coalesces and stays native.

Serde gates. Refused over NullType-bearing inputs: make_array (single row), array_union (drops entries), array_intersect (returns the other side's entries; reported Unsupported so the codegen dispatcher runs it at every setting, since the Incompatible branch has no dispatcher fallback under allowIncompatible, and array_except's unsupported element types likewise), array_repeat and slice (non-nullable item promised nullable), collect_list/collect_set (nested nullability mismatch), hash/xxhash64 (no Null arm), and map_from_arrays with a literal array beside a per-row one (the native map kernel reads a scalar list through its first row; pre-existing and independent of NullType, found while probing the NullType flavour). The null guards in CometElementAt (ANSI), CometArrayAppend, CometMapFromArrays, CometArraysZip, CometCoalesce and CometSize serialize their children into a predicate copy and a THEN branch, and a stateful child diverges from Spark's single evaluation either way it runs: native CASE evaluates the THEN copy on the rows the predicate selected, and a codegen-dispatched child (any lambda) is one cached kernel whose state both copies share. Any non-deterministic child inside a guard is refused, whichever argument is the nullable one (NullGuard); CometSize builds no guard for a non-nullable child or in legacy mode, where native already answers -1, so those stay native. CometIf, CometCaseWhen and CometCoalesce refuse a NullType result, since native CASE merges its branches' rows through Arrow's merge_n, which cannot build a NullArray with a validity bitmap; native GetStructField returns a scalar for a scalar struct input instead of a one-row array that a CASE result builder slices past, and rebuilds a NullType field as a fresh NullArray, since element_at on an out-of-range index hands over a Null child carrying a validity bitmap. array_union/array_intersect/array_except cast both sides to a deeply-nullable element type, since the native set-op kernel asserts identical nested nullability and a lambda variable arrives nullable where a literal field does not. The coalesce, set-op and non-NullType guard cases are reachable on main; they are included because the sweep found them and the fixes are shared.

Native to_csv. Yields NULL for a row that renders to an empty string and reports itself nullable, as Spark does: Spark hands the row to univocity's writeRowToString with skipEmptyLines on, so a struct with a lone null field (under the default empty nullValue) is NULL, not "". Pre-existing and independent of NullType; the allowIncompatible sweep profile found it on Spark 3.4, whose interpreted StructsToCsv shows the NULL, while Spark 3.5+ crashes in its own generated code on that NULL (nullSafeCodeGen never marks the result null), so the sweep counts the case as invalid in Spark there.

Native shuffle. The row shuffle writer's field-major paths gain the Null struct field case their row-major path already had, so a struct holding a NullType field shuffles through the JVM columnar shuffle instead of panicking, and the writer recreates every NullType-bearing builder after each batch, since NullBuilder::finish keeps its length and a second writer batch used to panic or miscount; the sweep no longer tolerates any failure.

Tests. CometNullTypeCompositionSuite sweeps every NullType producer under consumers, operators and nesting containers, across ANSI, nullable, non-deterministic and cross-input (a stateful argument beside a nullable one) settings, with the optimizer's null and comparison simplifications excluded so no consumer folds to a literal, and under physical profiles that vary the batching and the exchange path (native batches of two rows, 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, with operators that partition, join and run a scalar subquery on the value itself, comparing Comet with Spark with no tolerated failures and enforcing floors on the compared and natively executed counts. A registry check fails the suite if a registered array, map, struct or any-type aggregate serde is reached by no template, so a serde added later cannot stay outside the sweep. CometInMemoryCacheSuite round-trips NullType columns and children through the Arrow cache serializer. Unit tests in CometCodegenSourceSuite, UtilsSuite (including an exhaustive oracle for the coalesce bypass rule), CometJoinSuite, CometColumnarShuffleSuite, test_pyarrow_udf.py, and SQL-file witnesses that pin each fallback reason.

Benchmark. CometNullTypeColumnsBenchmark (separate commit) times the consumed map(id, NULL) projection and broadcast joins with a NullType build side against Spark and the prior fallback path, with an array<null> control, at default and small batch sizes 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 prior path: the gated build sides match it, and the array<null> join runs 2.0x Spark at 64 buffers against 1.1x when it fell back.

How are these changes tested?

Spark 4.1 / Scala 2.13: make test-jvm green on the earlier revision; on the final commit CometNullTypeCompositionSuite, CometSqlFileTestSuite, CometCodegenSourceSuite, UtilsSuite, CometArrayExpressionSuite, CometMapExpressionSuite, CometJoinSuite, CometAggregateSuite and the hash tests all pass, plus pytest test_pyarrow_udf.py. Spark 3.5 / Scala 2.12: CometNullTypeCompositionSuite, CometSqlFileTestSuite, UtilsSuite, CometCodegenSourceSuite pass. Spark 3.4 / Scala 2.12: CometNullTypeCompositionSuite and CometCsvExpressionSuite pass. Disabling any one gate turns the sweep red on the compositions it covers.

Assisted-by: Claude Code (claude-fable-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)