andygrove opened a new pull request, #5147: URL: https://github.com/apache/datafusion-comet/pull/5147
## Which issue does this PR close? Part of #5112. ## Rationale for this change #5112 measured roughly 10us of fixed cost per batch in the native columnar-to-row path versus roughly 0.2us for the JVM path, which is what makes the native converter 15.7x slower per row at 32-row batches and 3.7x slower at 8192 rows. This PR attacks the fixed cost only: the per-batch allocations, the redundant re-export of a constant schema, and the JVM objects allocated on the return path. This is a draft — it is a prototype to size the opportunity, not a finished change. See the checklist below for what is left. ## What changes are included in this PR? - **Reuse the Arrow FFI structs.** `NativeColumnarToRowConverter` now allocates the `ArrowArray`/`ArrowSchema` structs once and reuses them for every batch, instead of `exportBatchToAddresses` allocating two structs per column per batch. Those structs were also never closed, so the old code leaked them from the global Arrow allocator for the lifetime of the JVM. - **Export the C schema only when it changes.** The converter caches each column's Arrow `Field` and only exports the C schema when a field differs (for example when a column becomes dictionary-encoded). The native context caches the imported Arrow types and uses `from_ffi_and_data_type` for the other batches, so a steady-state batch exports and imports arrays only. - **Pass the FFI addresses once.** The struct addresses are now stable, so they are registered with the native context in `columnarToRowInit`; the per-batch JNI call takes only `(handle, numRows, hasSchema)`. - **No JVM object allocation on the return path.** `columnarToRowConvert` returns the address of three longs (row buffer, row offsets, row lengths) that the JVM reads with `Platform.getLong`, replacing the per-batch `FindClass` + `NativeColumnarToRowInfo` construction + two `int[]` allocations + two `SetIntArrayRegion` copies. `NativeColumnarToRowInfo` is removed, and the row iterator is reused across batches. - **Skip the per-batch cast pass** (and its `Vec` allocation and per-column `DataType` comparison) when the imported Arrow types are known to match the Spark schema. - **Attribution case in the benchmark:** `CometC2RIsolatedBench` gains an "Arrow FFI export only" case so the remaining fixed cost can be split between the JVM-side export and everything else. ## Results `CometC2RIsolatedBench` (4 columns: long, int, double, string; 1M rows; Apple M3 Ultra, Spark 3.5, release build), ns/row: | batch size | JVM rowIterator + UnsafeProjection | native (main) | native (this PR) | Arrow FFI export only | |---|---|---|---|---| | 8192 | 8.8 | 35.3 | 37.6 | 0.3 | | 512 | 9.0 | 55.7 | 48.0 | 4.7 | | 32 | 19.8 | 334.5 | 163.2 | 85.9 | Solving for a fixed-plus-linear cost model, the fixed cost per batch drops from roughly **9.5us to roughly 4.0us**, and about **2.5us of what remains is the Arrow FFI export and release on the JVM side** (~600ns per column). The per-row cost is unchanged to slightly worse, so the native path is still 4x slower per row than the interpreted JVM baseline at 8192 rows, and the JVM operator additionally fuses into whole-stage codegen. In other words: the small-batch cliff is roughly halved, but this alone does not make the native path competitive. That is the point of the measurement. ## Remaining optimizations Fixed cost (per batch): - [ ] Drop Arrow FFI entirely for this path — pass the raw validity/offset/data buffer addresses in a native scratch array written by the JVM and rebuild `ArrayData` against the cached types. No ownership transfer is needed here because every row is copied out before `convert` returns. Worth about 2.5us per batch based on the export-only measurement above. - [ ] Avoid materializing `ConstantColumnVector` into a fresh Arrow vector on every batch (currently forces a schema re-export too). - [ ] Handle dictionary-encoded columns without the per-batch `cast` in `maybe_cast_to_schema_type`, which fully materializes the column. Per-row cost: - [ ] Make the variable-length path columnar like `convert_fixed_width`: size pass over the offset buffers, prefix-sum row offsets, single buffer resize, then per-column monomorphized write loops (memcpy payloads, null bits from the validity bitmap in words). Today any var-length column forces `write_row_typed`, which resizes and zeroes per row, read-modify-writes the null word per null field, and dispatches per column per row. - [ ] `write_bytes_padded` appends alignment padding one byte at a time. - [ ] Investigate the small per-row regression in this PR at 8192 rows (35.3 → 37.6 ns/row), most likely the two `Platform.getInt` reads per row replacing on-heap `int[]` reads. A length prefix per row in the buffer would remove the offsets/lengths arrays entirely. Structural: - [ ] Revisit the blanket `unsafeRow.copy()` (#3367). Double-buffering the output in Rust and copying only in the broadcast path would restore Spark's normal row-reuse contract without reintroducing #3308. - [ ] Implement `CodegenSupport` so the operator is not a whole-stage-codegen boundary. Note the ceiling: the JVM operator often never materializes an `UnsafeRow` at all because the downstream stage reads columns directly from the vectors, while the native path must always serialize every column of every row. - [ ] Per-batch dynamic fallback to the JVM path below some row-count threshold. - [ ] `convertTime` only times the JNI call, so the per-row cost is misattributed to downstream operators in the Spark UI. If the columnar var-length rewrite still does not put the native path clearly ahead at 8192 rows on a wide, string-heavy schema, the honest conclusion is #4440 (remove the operator) rather than maintaining two implementations. ## How are these changes tested? - `CometNativeColumnarToRowSuite`: 26 tests pass. - Rust `columnar_to_row` unit tests: 18 tests pass. - `NativeUtilSuite` passes. - Benchmarked with `CometC2RIsolatedBench` as above. Note that #5114 proposes disabling this operator by default; this PR is orthogonal to that decision. -- 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]
