grorge123 commented on PR #5526: URL: https://github.com/apache/datafusion-comet/pull/5526#issuecomment-5563761588
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) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
