avantgardnerio opened a new pull request, #2337:
URL: https://github.com/apache/datafusion-ballista/pull/2337
## Purely additive
Nothing in the tree reads the new sketch. `RuntimeStatsExec` grows a second
sketch beside its `TDigest`; the routers, the wire format and the
scheduler-side merge all still read the `TDigest`, and its construction gate is
untouched. Deleting this diff restores the previous behaviour exactly.
Two details are not literally additive, both inside code that has no other
caller:
- `SortKeySketch::ingest` now calls `KllSketch::absorb_sorted_slice` rather
than `absorb_slice`. `SortKeySketch` landed in #2294 and has no in-tree caller
other than the one this PR adds.
- `sketch_batches` moves from inside the T-Digest block to the shared ingest
path, so its guard changes from "at least one non-null value" to "at least one
row". Identical while the construction gate rejects nullable routing
expressions, which is today.
## What this does
`RuntimeStatsExec` builds a `SortKeyCodec` from the first `ORDER BY`
expression and keeps a per-partition `SortKeySketch` next to each
`Mutex<TDigest>`. One evaluation of the routing expression feeds both, so the
second sketch costs a second ingest and not a second `evaluate`. Accessors are
`sort_key_sketch(partition)` and `merged_sort_key_sketch()`, mirroring the
T-Digest pair, and `Drop` logs count / nulls / min / max at `debug!` beside the
existing line.
A `sort_key_sketch_time` metric sits beside `sketch_time` over the same
batches, so any workload that runs the operator prices the replacement against
the incumbent without a separate benchmark.
## Why
`TDigest` is `Float64`-only, single-column, and has no representation for a
NULL. Those three limits are exactly what `RuntimeStatsExec`'s construction
gate enforces today, and they are why range repartitioning cannot route on an
`Int64`, a `Timestamp`, or a nullable column. `SortKeySketch` sketches any
fixed-width key and positions NULLs per `SortOptions::nulls_first`, so adopting
it is what lifts the gate.
Running both at once is the point of this step rather than a cost of it: the
two sketches see the same stream, so the replacement can be held against the
incumbent on real workloads before anything depends on it. On the #2294 thread
@phillipleblanc confirmed there is no need to preserve T-Digest compatibility
once KLL is ready, so the end state is a swap rather than two sketches.
## Evidence
`cargo bench --bench quantile_sketch`, release, n=1M, Ryzen 9 9950X:
| arm | time | vs T-Digest |
|---|---:|---:|
| `tdigest` | 13.737 ms | 1.00x |
| `sort_key_sketch_f64` | 15.933 ms | 1.16x |
| `sort_key_sketch_f64_nulls` | 15.236 ms | 1.11x |
| `sort_key_sketch_i64` | 15.980 ms | 1.16x |
| `sort_key_sketch_timestamp_ns` | 15.944 ms | 1.16x |
| `kll_norm_u64` | 17.092 ms | 1.24x |
And the sorted-input pair, which is why `ingest` takes the sorted path:
| arm | time |
|---|---:|
| `tdigest_sorted` | 2.187 ms |
| `kll_ordered_float_absorb_slice` | 25.229 ms |
| `kll_ordered_float_absorb_sorted_slice` | 7.252 ms |
Sorted input is worth 3.5x to KLL. It is worth 6.3x to T-Digest, which skips
its own sort and appends centroids, so KLL is relatively more expensive on a
sorted stream than on an unsorted one. Both sorted arms use the `OrderedFloat`
representation; there is no sorted arm for the shipped `u64` path yet, so its
sorted cost comes from the end-to-end run below rather than from this table.
### End to end
h2o window Q8 at 1e7 on a local 2-executor cluster, 4 vcores each, 8
partitions, `max_partitions_per_task=4`, release,
`ballista.planner.parallel_window.enabled=true` (not the default), 3
iterations. This is the plan shape that puts two `RuntimeStatsExec` taps in
stage 0:
```
ShuffleWriterExec: partitioning: UnknownPartitioning(8)
RuntimeStatsExec: rows + sketch(routing=v2@3 asc)
OrderedRangeRepartitionExec: routing=v2@3 asc -> 8 sorted partitions
SortExec: expr=[v2@3 ASC NULLS LAST], preserve_partitioning=[true]
RuntimeStatsExec: rows + sketch(routing=v2@3 asc)
DataSourceExec: file_type=parquet
```
Correctness, from the `debug!` lines of both sketches across 62 tap
instances covering 48.95M row observations: every count equal, **zero min/max
divergence**, NULLs counted 0 (the gate guarantees that today).
Cost, from the two metrics on the same taps:
| tap | T-Digest | sort-key | ratio |
|---|---:|---:|---:|
| below `SortExec`, raw parquet order | 467.2 ms | 537.4 ms | 1.15x |
| above `ORRE`, sorted | 84.4 ms | 188.9 ms | 2.24x |
| total | 551.6 ms | 726.2 ms | 1.32x |
+175 ms summed across 12 tap instances of a 2.5s query. The unsorted tap
reproduces the bench's 1.16x. The sorted tap at 2.24x beats the `OrderedFloat`
sorted arm's 3.32x, which is the first measurement of the shipped path on
sorted input.
## Deliberately not here
- **Consumers.** `discover_cuts`, `cut_partitions`, `merge_reports` and the
routers still read the T-Digest. Moving them means promoting cuts from
`Vec<f64>` to `Vec<ScalarValue>`, which pulls in NULL-valued cuts: a typed NULL
boundary makes every consumer's comparison NULL and drops that partition's rows
silently. That needs a cut type carrying the NULL run out of band, and it is
the next PR rather than this one.
- **The wire format.** `QuantileSketchState` is a 6-element T-Digest
`ScalarValue` array. A KLL at `KLL_K = 800` retains ~2400 `u64`, so a naive
`repeated uint64` is roughly 8x the bytes per partition entry. Worth a
report-level pre-merge (one merged sketch per report, `(min, max, count)` per
partition) rather than a straight substitution, since no consumer reads a
per-partition distribution.
- **Lifting the gate.** It has to stay while the T-Digest ingest downcasts
to `Float64Array`.
- **Plan-time sortedness.** `absorb_sorted_slice` decides per batch by
probing `is_sorted()` and silently falling back. `RuntimeStatsExec` could
decide once from `input.output_ordering()` and pick an explicit
`ingest_sorted`, the way `RangeFilterExec` already derives `sorted_on_key`.
That also makes the choice visible in a plan dump, which it is not today.
## Test plan
- `cargo test -p ballista-core --lib`: 317 passed.
- New `sort_key_sketch_agrees_with_tdigest`: 5,000 ascending rows, past
KLL's k=800 level-0 capacity so compaction runs. Extremes asserted equal
exactly; interior quantiles within 1% of the range, against the ~0.4% combined
rank error the two sketches are sized for.
- `execute_populates_sketch_and_row_count` and
`execute_row_count_only_no_sketch` extended to cover the new accessors,
including that row-count-only mode allocates neither sketch.
- clippy `--no-deps --all-targets -D warnings`, `cargo fmt --all`,
`RUSTDOCFLAGS='-D warnings' cargo doc --no-deps`: all clean.
--
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]
