avantgardnerio opened a new pull request, #2216:
URL: https://github.com/apache/datafusion-ballista/pull/2216
## Summary
Sketch-side prep for the eventual TDigest → KLL swap in `RuntimeStatsExec` —
the TODO on `ballista/core/src/execution_plans/runtime_stats.rs:39`. This PR
isn't the swap itself; it argues (with numbers) that the swap is affordable in
both the URRE and ORRE arms.
- `absorb(&[T])` (iterator) + `absorb_slice(&[T])` — batch-oriented ingest,
amortizes `compact_all`.
- Per-level sortedness invariant — `compact_level` skips re-sort when the
level was left sorted by an earlier promotion; promotions merge-extend in
linear time when the destination is sorted.
- `absorb_sorted_slice(&[T])` — for callers that can guarantee sorted input,
keeps level 0 permanently sorted and skips *every* compaction sort.
- Statistical test coverage against the DataSketches empirical bound
(uniform, clustered, heavy-ties).
The actual `RuntimeStatsExec` integration, wire format for KLL sketches
across the scheduler boundary, and the ORRE plan-shape follow-up are all out of
scope here — separate PRs.
## The two arms
**ORRE arm (ordered input, requires a follow-up plan-shape move):**
Current:
```
Scan ──▶ RuntimeStats ──▶ Sort ──▶ ORRE ──▶ ShuffleWrite
sketches unsorted rows; Sort then does the same work again
```
Follow-up:
```
Scan ──▶ Sort ──▶ RuntimeStats ──▶ DamExec ──▶ ORRE ──▶ ShuffleWrite
KLL<OrderedFloat<f64>>::absorb_sorted_slice
7.3 ms / 1M rows 0.53× TDigest (1.9× faster)
```
`SortExec` is already required by ORRE. Moving it above `RuntimeStats` costs
nothing plan-wise — the same rows get sorted the same way. `DamExec` (from
Ballista's existing operator vocabulary) holds sorted rows until the
scheduler's global cuts arrive. The sketch side is landed here; the planner
move is a separate PR.
**URRE arm (unordered input, ships as-is with this PR's shape):**
```
Scan ──▶ RuntimeStats ──▶ URRE ──▶ ShuffleWrite
KLL<OrderedFloat<f64>>::absorb_slice
25.1 ms / 1M rows 1.8× TDigest
```
## Why swap TDigest for KLL at all
TDigest at `TDIGEST_MAX_SIZE=100` can only sketch a single `f64` column. KLL
is generic over `T: Ord + Clone`, and composes with `arrow::row::OwnedRow` to
cover the full ORDER BY grammar for partition-boundary picking: multi-column
keys, nullable, per-column ASC/DESC, configurable NULLS FIRST/LAST. That
grammar is what parallel window (and later merge-join, order-by execution)
actually needs from the boundaries.
## Numbers (uniform f64, 1M rows, `k=800` for KLL, `TDIGEST_MAX_SIZE=100`)
| variant | time | throughput | vs TDigest |
|---|---:|---:|---:|
| TDigest (baseline) | 13.8 ms | 72 M/s | 1.0× |
| KLL<OrderedFloat<f64>> `insert` (per-row) | 48.3 ms | 21 M/s | 3.5× |
| KLL<OrderedFloat<f64>> `absorb_slice` (this PR, URRE arm) | 25.1 ms | 40
M/s | 1.8× |
| KLL<OrderedFloat<f64>> `absorb_sorted_slice` (this PR + planner follow-up,
ORRE arm) | 7.3 ms | 138 M/s | **0.53× (1.9× faster)** |
| KLL<OwnedRow> `absorb` (multi-column path, this PR) | 80.1 ms | 12 M/s |
5.7× |
The `OwnedRow` path stays a real regression (~5.7×) — the compaction cost
for heap-allocated rows is dominated by cache misses on the `OwnedRow` storage,
not sort-algorithm efficiency. Closing that gap is deferred-optimization item
(2) in the module doc (batch-oriented level-0-borrowed-row ingest), noted as a
TODO on the interim iterator API.
## Test plan
- [x] `cargo test -p ballista-core kll::` — 20/20 pass in 0.7s
- [x] `cargo bench --bench quantile_sketch -p ballista-benchmarks` — numbers
reproducible
- [x] `cargo clippy --all-targets` clean, `cargo fmt --all` applied
Coverage:
- Bit-identical between `insert` / `absorb` / `absorb_slice` on unique-value
streams.
- Bit-identical between `absorb_slice` / `absorb_sorted_slice` on sorted
unique streams.
- DataSketches P99 rank-error bound (`2.296 / k^0.9723`, source:
`datasketches-cpp/kll/include/kll_sketch_impl.hpp:619`) holds at ≤5% fail rate
across 100 seeds on uniform, clustered, and heavy-ties distributions.
- 4-way split + merge produces same quantiles as single sketch within 2ε.
- `debug_assert!` catches unsorted input to `absorb_sorted_slice`.
Not tested here (belongs on the swap PR):
- Actual `RuntimeStatsExec` integration.
- Wire format for KLL sketches across the scheduler boundary.
- End-to-end TPC-H / parallel-window benchmarks.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
--
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]
