fix: make columnar-to-row benchmarks exercise Comet

[rich7420]

Which issue does this PR close?

Closes #5624.

Rationale for this change

CometColumnarToRowBenchmark overrides getSparkSession without setting spark.shuffle.manager. With Comet shuffle enabled by default, this disables the extension, so all three arms measure Spark's ColumnarToRowExec despite their labels. The fixed-width and primitive fixtures also contain small integers that trigger the scan safety check.

What changes are included in this PR?

All changes are in CometColumnarToRowBenchmark.scala.

• Set CometShuffleManager before creating the Spark context.
• Disable the unsigned-small-integer safety check for these Spark-written fixtures, which contain no unsigned small integers.
• Run each arm once before timing and inspect the actual .noop() write plan through a query listener. Require the expected Spark, Comet JVM, or Comet native conversion, and write successful verification to the benchmark output.

The check fails before timing if an arm falls back or loses its conversion. Listener synchronization and plan checks are outside the timed cases.

How are these changes tested?

Benchmarks are compiled and linted in CI; execution was checked locally with a temporary driver calling all 10 existing fixture methods with 16,384 rows each on Spark 4.1.3 / Scala 2.13.17 / Java 17. All 30 execution-path checks and benchmark cases completed.

Both fallback paths were checked separately: removing the shuffle-manager setting from a temporary compiled copy, and re-enabling the small-integer safety check for the fixed-width fixture. Each was rejected at the Comet JVM preflight with Spark's ColumnarToRowExec in the captured write plan.

The fork CI run passed preflight, Rust formatting/clippy, benchmark compilation, and Spark 4.0 Scala compilation/lint. Local Spotless, RAT, and whitespace checks also passed.

[sunchao]

Correctness

This fixes the benchmark configuration that allowed all three labeled arms to execute Spark's ColumnarToRowExec. Installing CometShuffleManager before session creation satisfies the extension's default shuffle gate. The unsigned-small-integer override is appropriate for these generated fixtures: the maintained Spark 3.5 and 4.0 Parquet writers annotate ByteType and ShortType as signed INT8/INT16. The change does not alter fixture arithmetic, ANSI settings, schemas, or production fallback defaults.

The new preflight inspects the successful noop write's executed plan. On the maintained Spark 3.5 and 4.0 branches, that callback receives the query execution used to execute the write command. The inherited AdaptiveSparkPlanHelper.collect descends through adaptive plans and query stages, so the assertion is not limited to the outer wrapper. An empty conversion set or any different conversion class fails before timing. Listener callbacks complete before the buffer is read; the listener is removed in finally, and the same per-arm settings are applied to preflight and timed execution with configuration restoration afterward.

No verified P1/P2 issue remains in the changed file. The fixtures cover primitive and nested schemas supported by the existing native conversion guard. This assertion verifies the selected conversion path; it does not compare output values or establish null, overflow, or malformed-input correctness of the conversion implementations.

Validation and limits

Reviewed HEAD 5300abbffcd683f7632e2ba0912db7ad21481597 against BASE 7190df631afe3795914839203c7afe57ea23903c; the PR delta is one file, 67 additions and 19 deletions, from merge-base 93c323466feba127c46c1dfcd5fa4c8b3cc87195. The complete discussion/CI snapshot at 2026-09-05 21:25 UTC has no prior review threads and reports 8 successful / 11 skipped checks.

I verified that the benchmark compile/lint job checked out merge 76b9cf42c846f0e6898fccbe44207b68987b5371, whose parents are this base and head, and that its benchmark source matches the reviewed file. Spark 4.0.4 / Scala 2.13.16 / Java 17 compilation and lint succeeded with -DskipTests; this was not benchmark execution. Local whitespace validation passed. The PR reports a local Spark 4.1.3 run of all 30 cases at 16,384 rows and two negative fallback probes; I did not independently execute those runs. Source compatibility was checked against maintained Spark 3.5 and 4.0. Maintained 3.4 and 4.1 sources are unavailable, so those profiles remain unqualified by this source review.

Performance

The extra execution, listener synchronization, plan traversal, and verification output occur outside the timed callbacks. Each benchmark case retains Spark's own warmup and measurement loop. The added cost is one untimed query per arm and temporary plan references; it does not add traversal or listener registration to every measured iteration.

These remain scan-to-consumer benchmarks. The noop writer consumes the row iterator but discards values, while JVM code generation and native batch conversion have different materialization costs. The existing UDF/aggregate fixture exercises a different consumer path, and the Spark and Comet arms can also differ in scanning, aggregation, and shuffle. The wiring fix therefore supports correctly labeled runs, not an isolated conversion-speed or production-speedup conclusion. No performance measurement was produced in this review.

Design

Capturing the actual write action addresses the relevant failure mode directly: inspecting only the SELECT plan can miss transformations introduced by the sink. Failing before timing also prevents fallback results from being presented under a Comet label. The shuffle-manager setup follows CometBenchmarkBase, and keeping the signed-fixture override in this benchmark makes its assumption explicit. I found no actionable design issue in this scope.

Abstraction & complexity

The small local helper keeps each arm's name, expected operator, and configuration together, and reuses that configuration for verification and timing. It avoids duplicating listener lifecycle code across the three arms without adding a shared benchmark framework or production abstraction. The added structure is proportionate to the verification it provides; no simplification is required for correctness or efficiency.

[andygrove]

Thanks @rich7420