avantgardnerio opened a new pull request, #2294:
URL: https://github.com/apache/datafusion-ballista/pull/2294
**This PR is purely additive.** Nothing in the tree constructs any of it,
`RuntimeStatsExec` is untouched, and T-Digest is neither modified nor removed.
`main` behaviour is byte-for-byte identical. It is the foundation for widening
`RuntimeStatsExec` past the single non-nullable `Float64` sketch it accepts
today, split out so the widening itself can be reviewed on its own.
## What it adds
`ballista/core/src/sort_key.rs`:
- **`SortKeyCodec`** is the complete ordering spec for one fixed-width
`ORDER BY` key: its arrow type, its direction, and where its NULLs sort. It
encodes values to an order-preserving `u64` and back. Covers signed and
unsigned integers, `Float32`/`Float64`, and the temporal types that are `i32`
or `i64` underneath. Returns `None` for anything else, which is the signal to
fall back to arrow-row.
- **`SortKeySketch`** pairs that codec with a `KllSketch<u64>` over the
encoded values and a count of the NULLs, and owns `ingest` / `merge` /
`quantile` / `cuts` / `min` / `max`.
`ballista/core/src/kll.rs` gains `min()`, `max()`, `count()`, `at_rank()`,
`Clone` and `Debug`.
`benchmarks/benches/quantile_sketch.rs` gains arms that measure the shipped
path rather than an open-coded stand-in.
## Why
Two things the current sketch cannot express.
### NULLs have no representation
T-Digest has no NULL slot, and no sentinel `f64` can serve as one: `±inf`
collides with real infinities, and there is nothing else outside the value
range. `RuntimeStatsExec::try_new` therefore rejects nullable routing
expressions outright, and `split_batch_by_range` sends NULL keys to partition 0
regardless of what `nulls_first` says (there is a TODO to that effect at
`range_repartition_common.rs:162`).
`SortKeySketch` counts NULLs beside the sketch instead of encoding them. A
NULL has no position among the values, only a side, and `nulls_first` fixes
that side at plan time. Keeping it out of the key also leaves the key 8 bytes
wide instead of 16, which the benchmark prices at 1.19x against 1.58x.
The cost is that a rank over the population is no longer a rank over the
values, so the NULL run has to be stepped over. That remap lives in
`SortKeySketch::quantile` and nowhere else, which is the main reason the type
exists rather than a loose `(sketch, null_count)` pair.
### A sketch can lose its extremes
`tdigest.rs:210-215` reads a batch's extremes from `first()` and `last()`
only, then folds them with `f64::min` / `f64::max`, which are the NaN-ignoring
variants:
```rust
let maybe_min = *sorted_values.first().unwrap();
let maybe_max = *sorted_values.last().unwrap();
if self.count() > 0.0 {
result.min = self.min.min(maybe_min);
result.max = self.max.max(maybe_max);
```
Sorted by `total_cmp`, any batch containing a NaN has NaN at one or both
ends. That element is then discarded by the fold, and because nothing between
the ends is examined, the batch's real extremes go with it:
```
input : [1.0, 5.0, NaN, -NaN, +inf, -inf]
total_cmp : min=NaN max=NaN <- the true extremes
one batch : min=NaN max=NaN
split batches: min=1.0 max=5.0
split, no NaN: min=-inf max=inf
```
The last line is the control: same two-batch split without the NaN, and both
infinities survive. Reproduction is four lines against
`datafusion-functions-aggregate-common` alone.
This matters because `cut_partitions` routes shuffle files by exactly
`[sketch.min(), sketch.max()]`. A file reporting `[1.0, 5.0]` while spanning
`[-inf, +inf]` is routed only to buckets overlapping 1 to 5, and the rest of
its rows are not routed anywhere. There is no error, and since the outcome
depends on which batch arrived first it will not reproduce consistently.
`KllSketch` has no equivalent: `absorb_slice` scans every element using
`Ord` on the key, and `-NaN` / `+NaN` are simply the smallest and largest
`u64`. No shortcut, and no comparison that silently ignores a value.
## Cost
Ryzen 9 9950X, 64 MiB L3. `SortKeySketch` column is the shipped path.
| Rows | Column | T-Digest | SortKeySketch | Ratio |
|---|---|---|---|---|
| 100K | `Float64` | 1.38 ms (72 M/s) | 1.23 ms (81 M/s) | 0.89x |
| 100K | `Float64`, 10% NULL | 1.38 ms (72 M/s) | 1.27 ms (78 M/s) | 0.92x |
| 100K | `Int64` | unsupported | 1.22 ms (82 M/s) | 0.88x |
| 100K | `Timestamp(ns, UTC)` | unsupported | 1.21 ms (83 M/s) | 0.87x |
| 1M | `Float64` | 13.8 ms (73 M/s) | 16.4 ms (61 M/s) | 1.19x |
| 1M | `Float64`, 10% NULL | 13.8 ms (73 M/s) | 15.5 ms (65 M/s) | 1.13x |
| 1M | `Int64` | unsupported | 16.5 ms (61 M/s) | 1.20x |
| 1M | `Timestamp(ns, UTC)` | unsupported | 16.4 ms (61 M/s) | 1.20x |
T-Digest is `Float64`-only, so the integer and temporal rows have no
baseline of their own; those ratios are against T-Digest sketching a float.
Three things to read off it:
- **Column type does not move the cost.** 1.19x, 1.20x and 1.20x for float,
integer and timestamp. Compaction dominates ingest and the sketch only ever
sees `u64`, so the encode difference is noise. One number covers every
fixed-width column.
- **NULLs are a discount.** 1.13x against 1.19x, because a tenth of the rows
are counted rather than sketched.
- **The sign flips with scale.** 11% to 13% faster than T-Digest at 100K,
13% to 20% behind at 1M, as KLL's per-row cost grows with its compaction stack.
`KLL_K` is chosen for worst-case rank-error parity with T-Digest at
`max_size=100` (0.0016 against 0.0021 on a uniform 1M stream), so this is a
matched race rather than a cheaper sketch looking fast. Rerun the parity check
with `KLL_PARITY_CHECK=1 cargo bench --bench quantile_sketch`.
## Why not widen T-Digest instead
Worth stating plainly, because for the numeric and temporal types it is a
real option. You can wrap the routing expression in a `CastExpr` to `Float64`
and keep T-Digest. The precision loss is genuine but small enough not to
matter: an `f64` at 2020s epoch nanoseconds quantizes onto a 256 ns grid, while
T-Digest's own rank error over a day-wide partition is around three minutes.
Nine orders of magnitude apart. Any argument resting on quantile precision is
noise, and the sort_key docs say so explicitly so nobody over-claims it later.
What is left after discounting that:
- NULLs still have nowhere to go, and that is not a precision question.
- The extremes behaviour above is independent of column type.
- Casting is not free either. It needs a `CastExpr` planted at plan time in
the AQE rule, and then `RangeFilterExec`'s halos have to be expressed in cast
units rather than the column's own. That is plan surgery to avoid a code path
that would otherwise already exist.
- Float precision is relative, so the cast does degrade for a narrow spread
at a large magnitude. A partition covering a day is unaffected; one covering
100 µs at epoch-nanosecond magnitudes is past the point where the cast costs
more than the sketch does. A corner, but a real one.
Exactness does pay in one place that is not statistical: `min` and `max` are
exact by construction, tracked outside the compactor so no coin flip can move
them, and they are the two values `cut_partitions` routes files on. There the
error bars are zero, so anything a cast rounds away is error introduced where
none existed.
## Correctness of the arrangement
Arrow's row format is treated as the definition of `ORDER BY` order
throughout, since it is the only encoding that folds column type, `nulls_first`
and `descending` into one memcmp order, and arrow's own sort agrees with it.
The float transform here is the same one `arrow_row::fixed` applies, and both
reduce to `total_cmp`, which is what `ArrowNativeTypeOp::compare` uses.
`integer_keys_order_identically_to_arrow_row` asserts the two induce the
same permutation over a fixture containing ±NaN, ±0.0 and both infinities, so a
divergence fails a test rather than surfacing as misrouted rows.
Following that order is also why there is no NaN handling anywhere in this
PR. NaN has a defined place in `total_cmp`, beyond the infinity of its own
sign, so it encodes to an ordinary key at one end of the `u64` range and
comparisons against it behave normally.
## Testing
25 unit tests in `sort_key`, 21 in `kll`, full `ballista-core` suite green.
The NULL remap is validated differentially against `kll_norm_u128`, the
design this PR prices and rejects. Putting NULLs inside the key needs no remap
at all, which makes it an independent oracle for the arithmetic. The sweep
covers NULL fractions from 0 to 100 of 100 rows, both placements, ASC and DESC,
21 quantiles each.
That oracle earned its keep immediately: it found an off-by-one that eleven
hand-written tests had missed. Rescaling a rank into a fraction of the value
run and letting the sketch multiply it back loses it, since with 99 values rank
59 becomes `59/99` and `59/99 * 99` is `58.999...`, truncating to 58. Fixed by
keeping the subtraction in integers, which is what `KllSketch::at_rank` exists
for.
The worked bit-pattern table in `impl_sortable_float`'s docs is asserted by
`float_key_table_in_docs_is_accurate`, so the hex cannot rot.
## What this does not do
- Does not touch `RuntimeStatsExec`, T-Digest, or any routing code.
- `SortKeySketch` has no wire format yet. That lands with its first
consumer, so the encoding is designed alongside the reader rather than guessed
at.
- The arrow-row tier for multi-column and variable-width keys is measured
here but not built.
## Follow-ups
1. `RuntimeStatsExec` holds a `SortKeySketch` per partition, plus the proto
message.
2. Cuts become sort keys rather than floats through the router, which is
what lifts the `Float64` gate in `ParallelWindowRule` and lets a `Timestamp`
`ORDER BY` plan at all.
3. Multi-column and `Utf8` keys via the arrow-row tier.
--
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]
