apache / apache/datafusion-comet
Feed Spark's write path zero-copy row views instead of materializing UnsafeRow
- 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?
A write from a Comet plan pays for a columnar-to-row transition that materialises an `UnsafeRow`
per row, and none of the write path needs one. `OutputWriter.write`, `FileFormatDataWriter.write`
and `WriteTaskStatsTracker.newRow` are all typed on `InternalRow`, `ParquetWriteSupport extends
WriteSupport` and reads fields through `SpecializedGetters`, and
`BasicWriteTaskStatsTracker.newRow` ignores the row entirely. So the `UnsafeProjection` in
`CometColumnarToRowExec` builds a row that the writer immediately decodes again.
`ColumnarBatch.rowIterator()` already gives a reused `ColumnarBatchRow` that is a zero-copy view
over the Arrow buffers - `CometColumnarToRowExec.doExecute` produces exactly that and then throws
it away by projecting. Handing the writer the view instead skips the copy.
For flat schemas this is not worth doing: the projection there is a generated fixed-width copy and
it measures inside the run-to-run noise of a Parquet write. It becomes worthwhile once a struct,
array or map is present, because the projection then has to build nested `UnsafeRow` /
`UnsafeArrayData` with offset-and-length bookkeeping.
Measured on an M3 Max, 1M rows, Spark 4.1, release build, best-of, comparing against today's
`CometColumnarToRowExec`:
| schema | uncompressed | snappy |
| --- | --- | --- |
| flat (fixed width / strings / 50 columns) | 0-3% (noise) | 0-3% (noise) |
| one struct + array + map | 10% | 15% |
| struct + array-of-structs + map-of-array-of-structs | 11% | 13% |
| single struct, 1 level | 10% | 8% |
| single struct, 8 levels | 7% | 8% |
The gain comes from the presence of complex types rather than from depth - one level of nesting
already captures it.
This does not touch the real cost of a write, which is parquet-mr encoding. Only the native writer
changes that. It is a cheap improvement to the interim path while native writes remain
experimental, and unlike the native writer it is Spark-compatible by construction because Spark's
own writer still does the encoding.
## Describe the potential solution
A `ColumnarToRowTransition` that returns `batch.rowIterator()` unprojected, deliberately not
`CodegenSupport` so that whole-stage codegen does not regenerate the copy, planted by
`EliminateRedundantTransitions` under `WriteFilesExec` / `DataWritingCommandExec`.
The row view is a reused mutable row, so it is only correct for a consumer that finishes with a row
before pulling the next one. That restricts it to:
- unpartitioned and unbucketed writes, which is what makes `FileFormatWriter` pick
`SingleDirectoryDataWriter`. The partitioned and bucketed writers do not qualify - the required
ordering puts a `SortExec` in between and `UnsafeExternalSorter` needs `UnsafeRow`, and
`DynamicPartitionDataConcurrentWriter` spills through `UnsafeKVExternalSorter.insertKV` which is
typed on `UnsafeRow`
- Spark's own `FileFormat` implementations, whose `OutputWriter`s encode each row on the spot,
rather than a third-party format that may buffer the `InternalRow` it is handed
- schemas containing a complex type, per the measurements above
Behind an off-by-default config while it is experimental.
## Additional context
None.
Contributor guide
Research direction
Start with CometColumnarToRowExec, ColumnarToRowTransition, and EliminateRedundantTransitions, then trace how WriteFilesExec/DataWritingCommandExec select FileFormatWriter and SingleDirectoryDataWriter. Done means an off-by-default path uses batch.rowIterator() for complex-schema, unpartitioned and unbucketed Spark FileFormat writes while retaining projection for flat schemas and excluding partitioned, bucketed, and potentially buffering third-party formats.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- scala, spark
- Domain
- backend, data
- Issue type
- Feature
- Difficulty
- 5/5
- Estimated time
- Over a week
- Activity status
- Active
- Clarity
- Clearly specified
- Newbie friendliness
- 38/100