peterxcli opened a new pull request, #5496:
URL: https://github.com/apache/datafusion-comet/pull/5496
## 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`:
```shell
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:
```shell
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.
--
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]
