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.

[sunchao]

Correctness

Reviewed f7fd38b85083365f8312839b5d07b27c3263dca2 against 36146a87bf9ca9ca9e211b4372628ed2f9d8c8c6. I found no new P1/P2 issue in the nine-file change.

The existing non-codegen JVM path reads a ColumnarBatchRow and feeds it through UnsafeProjection. This adds an experimental direct writer for supported schemas. The shared conversion helper applies the same policy to partition execution and broadcast relation construction. The flag defaults to false, batches below 128 rows retain the existing projection, and an unsupported field keeps the whole schema on that fallback. Whole-stage codegen and native C2R do not use the new writer.

I compared the row layout and conversion contracts against the maintained Spark 3.5 and 4.0 branches. Null bitsets and unused slot bytes are cleared, strings retain eight-byte alignment and deterministic padding, and wide decimals reserve 16 bytes even when null. The decimal sign trimming preserves the minimal signed representation. Compact dictionary decimals use the dictionary accessor, while wide dictionary decimals use the existing decoded-value path. Dates and timestamps retain their already evaluated integer values, so this conversion adds no separate ANSI or timezone calculation. Buffer-size arithmetic is checked before narrowing or allocation.

The reused row has the same copy-before-retaining contract as the prior projection. The hash relation builders consume and copy its bytes before advancing. The fixed-width path owns a batch buffer, and the general path copies strings and decimals into its row buffer.

Validation

• An independent Java component probe compiled the exact converter and all eight vector source files, then passed 32 cases and 1,704 byte comparisons against Spark's row writer. It covered real Arrow plain and dictionary vectors, compact and wide decimals, nulls, noncanonical NaN payloads, long Unicode strings, growing and shrinking batches, multiword null bitsets, and retained row copies. It used cached Spark 4.1.3 and Arrow 18.3.0 dependencies with a small test-only field-type bridge. This was not a full Spark/JNI or repository build.
• The Spark 4.1/JDK 17 shuffle shard passed 498 tests, including all four new integration tests for mixed/null values, fixed-width conversion, small-batch fallback and unsupported-schema fallback. Its checkout was merge 9a7934c, whose tree matches the reviewed head.
• The JVM build passed at that same tree with tests skipped. The execution shard subsequently passed 922 tests, including all six new converter unit tests, at the same verified merge checkout. Current-head CI is now complete with 23 successful checks and 14 skipped checks. Skipped jobs are not counted as executed validation.
• Maintained Spark 3.4, 4.1 and 4.2 source branches were unavailable locally. The Spark 4.1 runtime and CI evidence above do not replace that canonical-source comparison.

Performance

The expanded isolated benchmark compares the direct writer with a same-run JVM projection control across five schemas and batch sizes 8192, 512 and 32. It prepares Arrow batches outside the timed cases and measures JVM allocation separately.

The author's reported 8192-row results support the decimal-heavy motivation: compact decimals improve from 20.1 to 10.4 ns/row and wide decimals from 67.0 to 37.4 ns/row. The mixed primitive/string case slows from 9.4 to 12.1 ns/row, about 29%, despite allocating less. These are author measurements of conversion alone, not independently reproduced query-level gains. The measured inputs also do not establish performance for null-heavy or dictionary data.

The fixed-width path trades a retained buffer for the whole batch and eager conversion for fewer per-value calls. That matters for large batches and consumers that stop early. The default-off setting, small-batch fallback and explicit experimental scope fit the schema-dependent evidence.

Design

The two conversion modes are straightforward: fixed-width batches are written column by column at a constant row stride, while variable-width rows use a reusable buffer. Sharing the selection logic between the two non-codegen entry points keeps their fallback behavior consistent. Constructing the fallback projection lazily also avoids its setup when every batch uses the direct path.

Keeping Spark's projection as the unsupported-schema fallback limits the compatibility surface. The configuration and tuning documentation explain how to opt in and where the optimization applies. I found no design change that needs to be made before merge.

Abstraction & complexity

The specialized writer adds substantial type-specific code, but each branch corresponds directly to Spark's physical row representation. Separate fixed-width loops avoid adding a per-cell strategy abstraction or generating another schema-specific class. The decimal-heavy measurements provide a concrete reason for that specialization.

The main maintenance obligation is preserving byte-level compatibility as types or vector representations evolve. The byte comparisons, explicit supported-type gate and existing fallback address that obligation within this change. I found no additional actionable simplification.

[andygrove]

I checked this out and byte-compared the direct converter against UnsafeProjection end to end rather than only through the unit suite, reading raw UnsafeRow.getBytes off queryExecution.toRdd with whole-stage codegen disabled so the path this PR changes actually runs. Eight schema shapes came back byte-identical: compact and wide decimals, strings, longs, a partition column, nulls at several strides, empty and multibyte strings, the decimal precision boundary at 18 and 19 and at 38, NaN, both infinities and negative zero. I also ran the broadcast relation build, which hands the reused row to BroadcastMode.transform on the driver and which nothing in the PR exercises, and a broadcast join over decimal and string build-side columns returned 500 rows, 500 distinct, identical with the flag off and on.

I also confirmed the flag-off path is the original code verbatim and that useDirectConverter short-circuits on the config before calling supportsSchema, so at the default settings this changes neither behaviour nor work done.

The part I could not get a test to reach is the three instanceof branches on vector class. DirectColumnarToRowConverterSuite builds every batch through CometArrowConverters.rowToArrowBatchIter, which only produces plain vectors, and it constructs ConstantColumnVector by hand so that case is covered. I tried to get a dictionary vector out of Parquet with four distinct values over 4000 rows and the native scan still handed the converter CometPlainVector for every column, decimals included. So writeWideDecimal's non-CometPlainVector branch and the dictionary side of compactDecimalValue are exercised by inspection only. Could the suite build an Arrow vector with a DictionaryEncoding directly and run a compact and a wide decimal through it? CometDictionaryVector.getLongDecimal looks right to me from reading, so I expect it to pass.

Two things worth adding to the tuning guide next to the flag. The fixed-width path converts the whole batch in setBatch, so an early-exit consumer like a LIMIT pays for all 8192 rows where the default path is lazy per row. And batchBuffer is retained per partition at fixedSize * numRows, roughly 6.7 MB per task for a hundred fixed-width columns, on heap and outside Comet's memory accounting. Both look like reasonable trades for the throughput, they just are not something a reader of the docs would discover before turning the flag on.

Last one. The compact and wide decimal fast paths skip the input.changePrecision(precision, scale) check that UnsafeRowWriter does, where a false result makes Spark write NULL instead of the value. I take it the assumption is that a Comet vector always carries a value already at the declared precision and scale, which I think holds after schema adaptation. Worth stating as a comment on compactDecimalValue and writeWideDecimal, since it is the one place the two writers can disagree and it is invisible from the code as written.