andygrove opened a new pull request, #6566:
URL: https://github.com/apache/datafusion-comet/pull/6566
## Which issue does this PR close?
Part of #6565. This PR does not close it.
## Rationale for this change
`CometSparkToColumnarExec` converts Spark rows and Spark column vectors to
Arrow through `ArrowWriter`, which `CometLocalTableScanExec` and the Comet
cache serializer also use. Most common column types were written one value at a
time:
- decimals through Arrow's `BigDecimal` setter, which allocates a
`BigInteger` and two byte arrays per value
- strings through `setSafe`
- arrays, maps and structs element by element
Fixed-width columns lost the bulk copy from #5442 as soon as a batch held
one null or was dictionary-encoded. #6565 has the per-type measurements.
End to end, on Apple M3 Max, JDK 17, the Spark 4.1 profile and a release
`libcomet`. Each number is the best of 5 runs at `local[1]` over 4M rows, with
Spark's vectorized Parquet reader feeding Comet through
`spark.comet.convert.parquet.enabled`. The cache queries use Spark's default
cache serializer. Query results are identical on both builds.
| Query | Wall time (ms) | `conversionTime` (ms) | Arrow batches |
|---|---|---|---|
| Parquet → TPC-H Q1-style aggregate | 1088 → 548 | 549 → 80 | 977 → 489 |
| Parquet → `sum(length(string))` | 193 → 124 | 73 → 10 | 977 → 489 |
| Parquet → `array<string>`, `struct`, `map<string,string>` | 345 → 225 |
158 → 61 | 245 → 123 |
| Parquet → int, long, double | 158 → 126 | 31 → 11 | 977 → 489 |
| Cache (row output) → Q1-style aggregate | 1185 → 814 | 820 → 485 | 489 |
| Cache (columnar output) → int, long, double | 76 → 64 | 24 → 9 | 800 → 489
|
Per column type, in ns per row with one column and 8192-row batches. These
come from a JMH harness that ran `main`'s `ArrowWriter` and this one in
alternating forks:
| Columnar input | 0% nulls | 10% nulls |
|---|---|---|
| int | 0.55 → 0.55 | 2.37 → 0.85 |
| long | 1.02 → 1.04 | 4.10 → 1.42 |
| boolean | 2.00 → 0.38 | 2.32 → 1.01 |
| decimal(9,2) | 34.5 → 1.5 | 37.7 → 1.9 |
| decimal(18,2) | 49.9 → 1.6 | 55.2 → 1.9 |
| decimal(38,10) | 60.9 → 6.0 | 56.2 → 7.3 |
| string | 14.9 → 1.5 | 14.2 → 2.0 |
| int, dictionary-encoded | 2.42 → 0.63 | 2.58 → 1.06 |
| string, dictionary-encoded | 15.9 → 5.9 | 13.0 → 6.7 |
| struct<int,long,double,date> | 34.4 → 3.7 | 37.3 → 5.5 |
| map<string,string> | 229 → 22.4 | 189 → 22.4 |
| array<int>, Parquet reader layout | 50.4 → 9.2 | 28.3 → 12.5 |
| array<string>, Parquet reader layout | 86.8 → 14.0 | 75.4 → 15.0 |
| array<struct<int,string>>, Parquet reader layout | 172 → 25.2 | 144 → 67.7
|
"Parquet reader layout" means a child slot for every null or empty
collection, which is how Spark's nested Parquet reader lays out its vectors.
The worst case for the array path is arrays of zero to two elements with half
the rows null, so that nearly every row is its own run of children. That case
is roughly even (0.9-1.1x).
On row input, decimal(9,2) and decimal(18,2) go from 105-157 to 8.5-9.5 ns
per row, and decimal(38,10) from 86 to 21. The other row types are unchanged.
The machine was shared with other jobs while I measured, so take the numbers
as indicative.
## What changes are included in this PR?
All of the columnar fast paths produce exactly what the row path writes for
the same rows. That includes the fields of a null struct coming out null, and a
decimal past 18 digits that overflows its precision still failing as Spark's
`getDecimal` does.
- **Decimals.** The unscaled value is written directly, as a sign-extended
long up to 18 digits. Past 18 digits it is written from its two's-complement
bytes, which are range-checked in place for Spark's vectors and unsafe rows.
The `Decimal` that `getDecimal` would build is only built for a value that does
not fit.
- **Fixed-width columns.** These are copied in bulk from Spark's vectors
whether or not they hold nulls. Off-heap vectors are copied straight from their
memory. Dictionary-encoded and other vectors are read value by value, skipping
null rows, because Spark 3.4's bulk getters decode ids under nulls. Validity is
written eight rows to a byte, and booleans are packed the same way.
- **Strings and binary.** Offsets and validity are written in one pass, and
the bytes are copied in one block (or one run of adjacent rows at a time). A
dictionary-encoded column decodes each id once per input batch and copies the
bytes for later rows. That cache is dropped at every input batch, because
Spark's readers reuse a vector across row groups whose dictionaries differ.
- **Arrays and maps.** Offsets are written in one pass and the elements are
written in runs of adjacent rows. Spark's nested Parquet reader leaves a child
slot for each null or empty collection, so a batch's elements are rarely
contiguous.
- **Structs.** These are written field by field. The fields of a null struct
are then nulled, unless a field is an array or a map, in which case a struct
with nulls takes the row path as before.
- **`SparkColumnarArrowReader`.** It now fills each Arrow batch across Spark
batches, so the 4096-row batches of Spark's Parquet reader and the 10000-row
batches of its cache become 8192-row batches. `ArrowWriter.writeColumns`
appends rather than overwriting its row count. Pulling the next source batch no
longer counts as conversion time.
- **Row input.** `CometSparkToColumnarExec` and `CometLocalTableScanExec`
count input rows once per batch instead of through a per-row iterator callback.
Two dispatch tests in `CometArrowStreamSuite` asserted that
dictionary-encoded vectors and slices under 32 rows go through Arrow's
per-value setters. They now assert that they don't.
## How are these changes tested?
- A new `CometArrowWriterSuite` writes columnar slices in uneven appended
chunks into an undersized root, and checks that every vector matches what the
row path writes for the same rows, including fields under null parents. It
covers:
- 17 primitive types, plain and dictionary-encoded, with the ids under
null rows out of range
- 11 nested types, including child vectors laid out backwards
- on-heap and off-heap vectors
- 0%, 20% and 100% nulls
It also checks decimals at every storage width, a wide decimal overflow,
nulled struct fields, and dictionaries that change between the input batches
the reader combines.
- A new `CometArrowStreamSuite` test fills Arrow batches from input batches
that reuse one set of vectors.
- `CometArrowWriterSuite`, `CometArrowStreamSuite` and
`CometStringWriterSuite` pass on Spark 4.1 and on Spark 3.4 / Scala 2.12.
- On Spark 4.1, these also pass: the `SparkToColumnar` and `LocalTableScan`
tests in `CometExecSuite`, `CometInMemoryCacheSuite`,
`CometMapExpressionSuite`, `CometArrayExpressionSuite` and
`CometTaskMetricsSuite`. All of them except `CometTaskMetricsSuite` also pass
on 3.4.
- The semantic scalafix check passes on the 3.5 profile.
- Outside the repo, a harness compared this writer with `main`'s over about
11,000 cases. It covered Spark's nested Parquet reader layouts, dictionaries,
off-heap vectors, slices, appends and decimal precision boundaries, and caught
each of the bugs I injected on purpose.
--
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]
