apache / apache/datafusion-comet

Optimize JVM columnar-to-row conversion

Open
#5,119 2 comments 0 reactions 1 assignee Claimed by @peterxcli View on GitHub
enhancement EPIC performance
Dominant language
Scala
Stars
1.3k
Forks
373
Avg merge
2d 4h
Merged PRs (30d)
198

Description

### What is the problem the feature request solves?

The schema-sweep benchmarks in #5120 (follow-up to #5112) show that the JVM `CometColumnarToRowExec` interpreted path loses to the native converter on some schemas, and that all of its losses trace to specific, fixable costs. The interpreted path matters in production: it runs whenever the consumer cannot fuse into whole-stage codegen, and in the broadcast relation build (`relationFuture` in `CometColumnarToRowExec`), which converts every batch of a broadcast table through `rowIterator` plus `UnsafeProjection`.

Current costs in the interpreted path:

- Every value goes through two virtual hops: `ColumnarBatchRow.getX` then `ColumnVector.getX`.
- Each non-null compact decimal value allocates a `Decimal` object (`CometVector.getDecimal`); each decimal with precision > 18 allocates a `byte[]`, `BigInteger`, java `BigDecimal`, scala `BigDecimal`, and `Decimal`. Measured: 392 bytes/row of heap garbage and 58.7 ns/row for a `2 x decimal(38,10), long` schema, vs 137 bytes/row and 70.5 ns/row for the native converter, whose per-value decimal cost is a plain 8 or 16 byte write.
- Null checks are per value even when the batch has no nulls.

In the codegen path, generated accessors fall back to the generic `ColumnVector` interface because no Comet exec node overrides `vectorTypes`, so every per-value accessor is an `invokeinterface`.

Optimizing the JVM path is attractive because it has near-zero per-batch overhead (roughly 0.2us vs roughly 10us plus per-column Arrow FFI cost for the native converter), so improving its per-value costs would make it competitive or winning across the whole batch-size and schema matrix, which also bears on #4440 (whether the native operator is worth keeping) and #5112.

### Describe the potential solution

Sub-tasks, roughly in order of expected impact:

- [x] Replace `UnsafeProjection` with a type-specialized direct converter in the interpreted path: dispatch on column type once per batch, read compact decimals as unscaled longs straight from the Arrow value buffer, copy decimal128 values as 16 raw bytes, and copy string bytes directly, with no per-value object allocation.
- [x] Add a column-at-a-time fast path for all-fixed-width schemas (constant row stride, per-column monomorphic loops, per-column `numNulls() == 0` short circuit), mirroring the native converter's fast path.
- [ ] Override `vectorTypes` on Comet exec nodes so `CometColumnarToRowExec` codegen casts columns to `CometVector` instead of the `ColumnVector` interface.
- [ ] Hoist null checks in `genCodeColumnVector` behind a per-batch `hasNull()` branch.

### Additional context

Benchmark data: #5120 (the schema scenarios were originally proposed in #5118, which is now folded into #5120). Related: #5112, #4440, #3367.

### Status

Sub-tasks 1 and 2 are implemented in #5120 behind the experimental config `spark.comet.exec.columnarToRow.direct.enabled` (default false), with a `minBatchSize` threshold below which the default conversion is used.

Sub-tasks 3 and 4 are skipped for now: both target the codegen path and the expected win is low single-digit ns/row. The `vectorTypes` override cannot promise monomorphic call sites (a column can be dictionary-encoded in one batch and plain in the next, and batches can contain Spark's `ConstantColumnVector` after constant materialization), and the null-check hoist saves only a validity-bitmap bit test behind a well-predicted branch. Benchmark data in #5120 shows the interpreted-path allocation and dispatch costs were the dominant factor, and those are addressed.

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.