andygrove opened a new pull request, #6910:
URL: https://github.com/apache/datafusion-comet/pull/6910
## Which issue does this PR close?
N/A. Related to the native Iceberg write epic, #5649, where turning native
writes on by default (#5644, #6664) needs end-to-end numbers against Spark.
## Rationale for this change
No existing benchmark measures an Iceberg pipeline end to end, and none of
the Iceberg benchmarks covers complex types:
- `CometIcebergReadBenchmark` scans a single primitive column and sums it.
- `CometIcebergWriteBenchmark` reads Parquet and writes Iceberg, over
primitive columns only. Its copy-on-write `DELETE` case does read Iceberg, but
through `SparkCopyOnWriteScan`, which Comet does not run natively (#5319).
This adds a benchmark of the common Iceberg ETL shape: read a table,
repartition it, write it to another table. It reports how Comet compares with
Spark for each family of column types, with the iceberg-java writer and with
the native writer.
## What changes are included in this PR?
`CometIcebergEndToEndBenchmark` times `INSERT INTO target SELECT /*+
REPARTITION(10, id) */ * FROM source`:
- Five groups of column types, one results table each: primitives (the
`CometIcebergWriteBenchmark` corpus), structs, arrays, maps, and nested types
(an array of structs, a map of arrays, and a struct holding an array and a
map). Nulls appear at every level, and one array or map in ten is empty.
- Three configurations: Spark; Comet with the native scan and native shuffle
but the iceberg-java writer, which is what Comet does by default today; and
Comet with the native writer as well.
- Spark writes the source table once per group, so every configuration reads
the same files, and `read.split.open-file-cost` gives each of its 10 files its
own scan task.
- The repartition hashes a primitive key, which Comet's native shuffle
supports with complex columns carried alongside. A round-robin
`REPARTITION(10)` would run on Comet's JVM shuffle unless
`spark.comet.shuffle.native.partitioning.roundrobin.enabled` is set.
- Before timing, each configuration runs once and is checked. The row count
and an order-independent hash of every row, read back with Comet off, must
match the source. The plan must hold exactly one exchange. The Comet
configurations must run the native scan and the native shuffle, and the third
one the native writer. A stage that does not run natively relabels the row, for
example `(not native: shuffle)`, and prints a warning. Conditions that make the
comparison meaningless throw.
It also moves `capturePlans`, `collectAcross`, `isComet` and
`icebergOnClasspath` out of `CometIcebergWriteBenchmark` and
`CometIcebergSystemFunctionBenchmark` into `CometBenchmarkBase`, so the Iceberg
benchmarks share one copy. The system-function benchmark's classpath probe
changes from `BucketFunction` to `SparkCatalog`, which every Iceberg runtime
the build uses provides.
The benchmark is listed in the Iceberg write contributor guide's testing
table and in the `review-comet-iceberg-write-pr` skill.
## How are these changes tested?
This is a benchmark, so it adds no tests. It was run locally on the default
profile (Spark 4.1, Iceberg 1.11) on an Apple M3 Ultra, with `local[5]` and 2M
rows. Best time in ms, with the speedup over Spark:
| Types | Spark | Comet scan + shuffle | Comet scan + shuffle + native
write |
| ---------- | ----- | -------------------- |
----------------------------------- |
| primitives | 1042 | 806 (1.3x) | 524 (2.0x)
|
| structs | 917 | 606 (1.5x) | 415 (2.2x)
|
| arrays | 1477 | 1045 (1.4x) | 584 (2.5x)
|
| maps | 1536 | 1018 (1.5x) | 566 (2.7x)
|
| nested | 3404 | 2206 (1.5x) | 1172 (2.9x)
|
Two runs agreed to within 0.1x on every ratio.
The checks were also exercised:
- With `spark.comet.scan.icebergNative.enabled=false`, Comet does not engage
at all and the benchmark stops at the first Comet configuration.
- With `spark.comet.shuffle.mode=jvm`, every Comet row is relabelled `(not
native: shuffle)`. The native-writer configuration then measures 1.7x to 2.5x.
The data-generation SQL produces identical checksums on stock Spark 3.5.8
and 4.1.1. The Spark 3.5 profile compiles under `-Pstrict-warnings`, and the
scalafix check passes on the Spark 3.5 and 4.1 profiles.
--
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]
