andygrove opened a new pull request, #5626:
URL: https://github.com/apache/datafusion-comet/pull/5626
## Which issue does this PR close?
Closes #5625.
**Experimental / RFC.** Off by default, and I am not convinced the gain
justifies the hazard yet -
see the caveats at the bottom. Opening as a draft to get the measurements
and the safety argument
in front of people rather than because I think it is ready to land.
## Rationale for this change
Comet has to do a columnar-to-row conversion before a write because native
write support is still
experimental. That conversion materialises an `UnsafeRow` per row, and
nothing on Spark's write
path needs one:
- `OutputWriter.write(InternalRow)`,
`FileFormatDataWriter.write(InternalRow)` and
`WriteTaskStatsTracker.newRow(String, InternalRow)` are all typed on
`InternalRow`
- `ParquetWriteSupport extends WriteSupport<InternalRow>` and reads fields
through
`SpecializedGetters`
- `BasicWriteTaskStatsTracker.newRow` ignores the row entirely and just
increments a counter
Meanwhile `CometColumnarToRowExec.doExecute` already produces
`batch.rowIterator()` - a reused
`ColumnarBatchRow` that is a zero-copy view over the Arrow buffers - and
then throws it away by
applying an `UnsafeProjection`. For a write that copy is pure overhead: the
writer decodes straight
back out of the row it was just given.
This does not touch what actually makes a write expensive, which is
parquet-mr encoding. Only the
native writer changes that. What it does do is make the interim path cheaper
while native writes
remain experimental, and unlike the native writer it stays Spark-compatible
by construction,
because Spark's own writer still produces the bytes.
## What changes are included in this PR?
`CometColumnarToRowViewExec` returns `batch.rowIterator()` unprojected. It
is deliberately **not**
`CodegenSupport`: whole-stage codegen would generate an `UnsafeRowWriter`
loop and reintroduce
exactly the copy this exists to avoid.
`EliminateRedundantTransitions` plants it under `WriteFilesExec` (planned
writes) or
`DataWritingCommandExec` (when `plannedWrite` is off). The row it hands over
is reused and mutable,
so it is only correct for a consumer that finishes with a row before pulling
the next one. Three
gates keep it there:
- **unpartitioned and unbucketed**, which is what makes `FileFormatWriter`
choose
`SingleDirectoryDataWriter`. The other writers do not qualify: the
required ordering on
partition/bucket columns puts a `SortExec` in between and
`UnsafeExternalSorter` needs
`UnsafeRow`, and `DynamicPartitionDataConcurrentWriter` spills through
`UnsafeKVExternalSorter.insertKV`, which is typed on `UnsafeRow`.
- **one of Spark's own `FileFormat`s**, whose `OutputWriter`s encode each
row on the spot (Parquet
through `ParquetWriteSupport`, ORC through `OrcSerializer` into a
`VectorizedRowBatch`, the text
formats directly). A third-party format is free to buffer the
`InternalRow` it is handed.
- **a schema containing a struct, array or map.** On flat schemas the
projection is a generated
fixed-width copy and the saving is inside the noise, which does not pay
for the hazard above.
Behind `spark.comet.exec.write.rowView.enabled`, default false.
Also adds `CometParquetWriteBenchmark`. Note it sets `spark.shuffle.manager`
explicitly - without
it `isCometLoaded` disables Comet and every "Comet" arm silently measures
Spark, which is #5624.
Each arm asserts through a `QueryExecutionListener` that it planned the
transition its label names,
and warns into the results file if not.
## Measurements
M3 Max, 1M rows, Spark 4.1, release build, best-of-N, versus today's
`CometColumnarToRowExec`:
| schema | uncompressed | snappy |
| --- | --- | --- |
| fixed width (10 cols) | 542 -> 526 ms (declined by the gate; noise) | 553
-> 543 ms (noise) |
| strings | 620 -> 618 ms (declined; noise) | - |
| wide, 50 columns | 3166 -> 3096 ms (declined; noise) | - |
| struct + array + map | 1027 -> 920 ms (**10%**) | 1104 -> 934 ms (**15%**)
|
| struct + array-of-structs + map-of-array-of-structs | 1838 -> 1630 ms
(**11%**) | 1850 -> 1616 ms (**13%**) |
| single struct, depth 1 | 351 -> 315 ms (**10%**) | 350 -> 323 ms (**8%**) |
| single struct, depth 2 | 431 -> 394 ms (**9%**) | 434 -> 382 ms (**12%**) |
| single struct, depth 4 | 575 -> 538 ms (**6%**) | 573 -> 537 ms (**6%**) |
| single struct, depth 8 | 875 -> 816 ms (**7%**) | 881 -> 815 ms (**8%**) |
The declined rows are both arms running the identical plan, so they double
as a noise-floor
estimate: about 0-3%.
Two things worth noting. The gain comes from complex types being present,
not from depth - one
level already captures it, and it flattens out after that. And the native C2R
(`spark.comet.exec.columnarToRow.native.enabled`) is consistently *slower*
than the JVM one on
nested data here, 0.9X against Spark in several groups, which matches its
documented per-batch JNI
cost.
## How are these changes tested?
New `CometWriteRowViewSuite`, 11 tests, registered in both PR build
workflows. The bar for each is
that turning the config on changes nothing observable but the plan:
- byte-identical round trip against the same Comet plan with the config off,
for hand-written
primitives/strings/nulls/nested, for fuzz-generated flat and nested
schemas, and for a
deliberately deep four-level schema with nulls at every level
- the same for ORC and JSON, not just Parquet
- `maxRecordsPerFile` set and unset
- the transition is present exactly once when expected, absent by default,
absent for a flat
schema, absent for partitioned and for bucketed writes, and present for a
complex column sitting
alongside flat ones
Also ran `CometParquetWriterSuite` and `CometNativeColumnarToRowSuite`
alongside it: 67 tests pass.
## Caveats I would want a second opinion on
1. **Is 6-15% on complex-typed writes worth a reused-mutable-row in the plan
at all?** The gates are
argued from Spark's source rather than enforced by anything, and a future
Spark change to a
writer that starts retaining rows would be silent data corruption rather
than a test failure. A
defensive option would be to restrict it to `ParquetFileFormat` only.
2. **The format gate is a package-name prefix check**, which is crude. I
could not find a better
signal for "this `OutputWriter` does not retain the row".
3. Only exercised on Spark 4.1 so far.
--
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]
