perf: optimize JVM columnar-to-row conversion

[peterxcli]

Which issue does this PR close?

Closes #5119.

This completes the two implementation tasks retained in that issue. It refreshes and productionizes the prototype from #5120 on current main, with expanded correctness coverage and a new same-machine A/B benchmark.

Rationale for this change

The interpreted JVM columnar-to-row path used by CometColumnarToRowExec reads every value through ColumnVector accessors and then writes it through UnsafeProjection. That path is already efficient for simple primitive schemas, but it creates substantial per-value garbage for decimals:

• compact decimals allocate a Decimal object per value;
• decimals with precision above 18 allocate a byte[] / BigInteger / BigDecimal chain;
• the extra allocation increases executor GC pressure even when an idle microbenchmark does not fully expose it.

The benchmark introduced in #5113 originally covered only long, int, double, string, which is close to the existing JVM path's best case. The broader schema analysis in #5112 and #5118 showed that decimal-heavy and all-fixed-width schemas behave very differently. This PR therefore expands the benchmark matrix and adds a direct JVM converter that writes Arrow values into Spark's UnsafeRow representation while reusing its row and buffers.

This approach also avoids crossing JNI and avoids the native converter's per-row defensive copies required by the row-lifetime contract fixed in #3367.

Related behavior and history were cross-checked against #5114, #5115, #3308, #3221, #3266, #3268, #3649, and the production GC discussion in #4440.

What changes are included in this PR?

Direct JVM converter

Adds DirectColumnarToRowConverter, which resolves column types once and writes directly into UnsafeRow storage.

Supported types are:

• boolean, byte, short, integer, and long;
• date, timestamp, and timestamp without time zone;
• float and double;
• UTF-8 string;
• compact and wide decimals.

The converter uses:

• a general row-at-a-time path for schemas containing strings or wide decimals;
• a column-at-a-time, constant-stride path when every column is fixed-width;
• unscaled longs for compact decimals;
• raw big-endian byte copies for wide decimals;
• direct copies into the variable-width row area for strings.

The output is byte-identical to UnsafeProjection, including null-slot zeroing, deterministic padding, multi-word null bitsets, wide-decimal reservations, and Spark's canonical NaN representation. The converter also rejects oversized schemas, batches, and row buffers before allocation.

Integration and fallbacks

Regular execution and broadcast relation builds now share one conversion helper.

Two experimental settings control the feature:

• spark.comet.exec.columnarToRow.direct.enabled=false
• spark.comet.exec.columnarToRow.direct.minBatchSize=128

The optimization is disabled by default. When enabled:

• unsupported schemas fall back to rowIterator plus UnsafeProjection once per plan;
• batches below minBatchSize fall back independently;
• only the JVM operator's non-codegen paths are affected, including broadcast relation builds.

The settings and fallback behavior are documented in the tuning guide.

Benchmark coverage

CometC2RIsolatedBench now compares the existing JVM path, the direct converter, and the native converter across five representative schemas and batch sizes 8192, 512, and 32. It reports both wall-clock time and JVM heap allocation per row. The sink reads raw UnsafeRow slots so benchmark-side wrapper allocation does not distort converter cost.

Known trade-off

The direct path is not universally faster. The mixed primitive/string schema contains no expensive decimal accessors, and its string column prevents the fixed-width path from engaging. UnsafeProjection is already schema-specialized straight-line code for this case, while the general direct path still pays per-field dispatch and row-assembly costs.

For that reason this PR keeps the feature opt-in. It does not add a benchmark-tuned schema heuristic, and the small-batch threshold only addresses per-batch amortization rather than the large-batch mixed-schema regression.

How are these changes tested?

Correctness and repository checks

DirectColumnarToRowConverterSuite compares raw output bytes with UnsafeProjection across supported types, nulls, fixed- and variable-width paths, decimal boundaries, empty strings, multi-word null bitsets, noncanonical float/double NaN payloads, and oversized fixed-width batches.

CometDirectColumnarToRowSuite verifies end-to-end Spark results and plan selection with whole-stage codegen disabled, including mixed types, the fixed-width path, unsupported-schema fallback, and minimum-batch fallback.

Commands run after rebasing onto current main:

make core

./mvnw test -Dtest=none \
  -Dsuites=org.apache.comet.DirectColumnarToRowConverterSuite,org.apache.comet.exec.CometDirectColumnarToRowSuite

make format PROFILES=-Pspark-4.0
python3 dev/ci/check-suites.py
git diff --check upstream/main...HEAD

Results:

• focused Spark 4.1 run: 2 suites, 10 tests, all passed;
• Spark 4.0 formatting/scalafix gate: passed;
• Spark 4.1 compilation, Spotless, and Scalastyle: passed;
• workflow suite registration and diff checks: passed.

Benchmark

Command:

make benchmark-org.apache.spark.sql.benchmark.CometC2RIsolatedBench

Environment:

• Apple M4, 10 CPUs, 24 GiB RAM, AC power;
• macOS 26.5.2;
• OpenJDK 21.0.6 with -Xmx20g;
• Spark 4.1.3 / Scala 2.13;
• Rust 1.95.0;
• native release build with -Ctarget-cpu=native.

Protocol:

• isolated temporary checkouts for unchanged main and the proposed converter;
• identical benchmark instrumentation in both snapshots;
• two runs per snapshot, rejecting the noisy first baseline and retaining the quiet second runs;
• 1,048,576 rows per scenario, with each Spark Benchmark case measured for at least two seconds;
• the same-run JVM control is the primary comparator, avoiding attribution of cross-run JVM drift to the converter;
• results below are the representative large-batch cases at batchSize=8192.

The A/B was recorded at the implementation base (2699f59b7). The subsequent rebase did not change CometColumnarToRowExec, the benchmark, or the vector accessors on main; the current rebased code was recompiled and retested as listed above.

Scenario	Unchanged-main JVM	Same-run JVM control	Direct	Direct vs control	JVM -> Direct allocation
long, int, double, string	9.3 ns/row	9.4 ns/row	12.1 ns/row	0.78x, 29% slower	24.1 -> 8.4 B/row, 65% lower
4 x decimal(12,2), date, 2 x string	31.0 ns/row	29.7 ns/row	26.6 ns/row	1.12x faster	131.6 -> 3.9 B/row, 97% lower
4 x decimal(12,2), date, long	24.5 ns/row	20.1 ns/row	10.4 ns/row	1.93x faster	152.2 -> 24.4 B/row, 84% lower
2 x decimal(38,10), long	70.5 ns/row	67.0 ns/row	37.4 ns/row	1.79x faster	392.1 -> 24.0 B/row, 94% lower
16 x long	35.6 ns/row	30.2 ns/row	20.0 ns/row	1.51x faster	24.5 -> 25.1 B/row, effectively unchanged

The direct converter wins 4 of 5 representative large-batch schemas. Winning cases improve by 1.12-1.93x, while decimal-heavy schemas reduce JVM heap allocation by 84-97%. The mixed primitive/string regression is intentionally reported rather than averaged away and is why the feature remains disabled by default.