feat(core): Replace TDigest with KLL - #2337
Merged
avantgardnerio merged 30 commits intoAug 19, 2026
Merged
Conversation
RuntimeStatsExec carries a SortKeySketch beside its TDigest, built from a SortKeyCodec over the first ORDER BY expression. Both sketches observe the same stream from one evaluation of the routing expression, so either can be held against the other on any workload rather than only in a unit test. Nothing consumes the new sketch yet. The routers, the wire format and the scheduler-side merge still read the TDigest, and the Float64 non-nullable construction gate stays for its sake — that gate is what the swap lifts, since SortKeySketch sketches any fixed-width key and gives NULLs a position per nulls_first. SortKeySketch::ingest takes KllSketch::absorb_sorted_slice, worth 3.5x on sorted input (25.2ms -> 7.3ms per 1M rows, release), falling through to absorb_slice when the input is not sorted. Deciding that from the input's declared ordering instead of probing each batch is a follow-up. sketch_batches moves to the shared ingest path so deleting the TDigest cannot take the counter with it, and sort_key_sketch_time prices the replacement against the incumbent in situ on any workload. h2o window Q8 @1e7, release, 3 iterations: 62 taps compared, every count equal and zero min/max divergence. Sketch cost is 1.15x the TDigest on the unsorted tap and 2.24x on the sorted one, +175ms summed across 12 tap-instances of a 2.5s query. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
avantgardnerio
changed the title
`SortKeySketchState` carries a KLL compactor stack over one ORDER BY key: `k`, the NULL count, the exact extremes as `repeated ScalarValue`, and the levels as one Arrow IPC stream — one row per level, a single `List<Struct<expr_0>>` column, items ascending. Arrow IPC rather than a packed blob because the arrow schema states what a key is, so a multi-column key becomes struct siblings instead of a second payload shape needing a discriminant field. The alternatives are recorded on the message: a packed blob with a key-family tag, and delta + varint, which lands within 10% of the information-theoretic floor for a sorted sample and so buys only ~24% on Float64 keys. Dense Int64 keys measured ~4x, which is the case worth revisiting for, and adopting it bumps BALLISTA_PROTOCOL_VERSION rather than needing a per-message version field. The key's direction and NULL placement stay out of the message. They live once per report in the `order_by` tag, so `try_from_proto` takes SortOptions from the caller and reads the key's type from the payload's own schema. `RuntimeStatsReport` gains a report-level merged sketch and `RuntimeStatsPartitionEntry` gains its exact key range and NULL count, filling the `MinMaxState` TODO. No consumer reads a per-partition distribution — `merge_reports` folds them all for global cuts and `cut_partitions` needs only each partition's extremes — so merging on the executor keeps a report's size independent of the operator's partition count. Neither field is populated yet. `KllSketch` exposes `k()`, `levels()` and `from_parts()`, all generic in `T`, so no key representation reaches its signatures. `from_parts` returns `Option` rather than a `Result` to keep `kll.rs` free of its DataFusion dependency; it refuses a stack KLL could not have produced and sorts every level, so sorted-before-serialization is enforced on both sides rather than carried as a wire flag. Measured at 816 retained items over 1M rows: 7,460 bytes encoded against 6,528 raw, so 1.14x for framing. Per task at K=8 that is ~7.7 KB against 16.6 KB of per-partition T-Digests today. Round trip asserts every answer, not the byte count: both extremes exactly and a 101-point quantile sweep. Mutation-checked — dropping the top level fails it, and so does flattening the stack into one level, which preserves every item, the count and both extremes while corrupting only the weights. Empty, NULLs-only and garbage-payload cases covered. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
`stats_to_report` fills the report-level merged sketch from `merged_sort_key_sketch()`, and each partition entry's key range and NULL count from that partition's own sketch. Both executor task paths and both scheduler status paths already move the whole report, so nothing else needed plumbing. `SortKeySketch` gains `value_min()` / `value_max()`, distinct from `min()` / `max()` on purpose. Those answer about the least and greatest *row*, so they report a typed NULL when NULLs sort at that end, which is the wrong answer for a router: a NULL bound makes the comparison NULL and silently drops every row it was meant to select. The per-partition range is a value range. The scheduler decodes the sketch on arrival and logs what it rebuilt. Reconstructing it there is the point — a byte count proves the field crossed, where a count and a range prove it survived. The key's direction and NULL placement come from the report's `order_by` tag, which is the only place they live. h2o window Q8 @1e7, release, 2 executors, K=8, `max_partitions_per_task=4`: task=1 total_rows=3951424 sketches=8 key_ranges=8/8 sort_key_sketch={bytes=6280 k=800 count=3951424 nulls=0 min=1.0000059022230063 max=99.99997064917741} task=0 total_rows=6048576 sketches=8 key_ranges=8/8 sort_key_sketch={bytes=7176 k=800 count=6048576 nulls=0 min=1.0000105194802908 max=99.99996897925054} Each sketch's count equals its task's row count and the two sum to 10,000,000, the query's full input. `key_ranges=8/8` is counted rather than assumed, so a partition entry whose range never got filled in would show up here instead of routing no files in silence. Payload is 6.3-7.2 KB per task against 16.6 KB of per-partition T-Digests. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
`cuts` ranked over the population and reported a typed NULL wherever a
boundary landed inside the NULL run. Every consumer then dropped that
partition in silence: `RangeFilterExec` and `PerPartitionFilterExec` compare
against a boundary in SQL's three-valued logic, and `cut_partitions` computes
`NULL + halo`. Ranks are now taken among the values, so no rank a NULL could
answer exists and the bad boundary is unrepresentable rather than filtered
out afterwards.
Sizing still targets the population, so a run shorter than one partition's
share costs no balance at all — it shares the lowest partition with the values
below the first cut. A longer run is indivisible, since NULLs cannot be told
apart and no split of them is reproducible read-side, so it fixes partition
0's size and only the values above it can balance.
Three floors on one rank, each carrying its own case:
population_rank
.saturating_sub(null_count) // step over the run
.max(even_rank) // split the values above it evenly
.max(cut + 1) // keep boundaries distinct
Dropping the second gives `[60, 1, 13, 26]` where `[60, 13, 13, 14]` is
available: the swallowed boundaries bunch at the lowest values while the ones
past the run keep ranks measured over rows the run already took, leaving the
top partition holding the difference. Dropping the third makes the first two
boundaries both name the minimum, since a KLL rank is a cumulative-weight
threshold and ranks 0 and 1 coincide.
NULLs values K cuts partition sizes
10 1..=90 4 [15, 40, 65] 24 / 25 / 25 / 26
60 1..=40 4 [1, 14, 27] 60 / 13 / 13 / 14
90 1..=10 4 [1, 4, 7] 90 / 3 / 3 / 4
`quantile` is untouched and still answers NULL inside the run, which is the
honest answer about the value at a rank over every row. The two notions of
rank now differ deliberately: one describes the distribution, the other names
boundaries a router can compare against.
`nulls_last` with NULLs is refused rather than mis-cut — the run sits above the
values there, so boundaries pull down from the maximum and need their own
expression. h2o Q8 declares NULLS LAST but observes no NULLs, so it takes the
allowed path.
Tested as a property over five shapes rather than by pinning integers: every
boundary non-NULL, boundaries non-decreasing, every row accounted for, and
partitions above the run within one row of each other. Mutation-checked by
removing each floor.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
`cuts` refused a nulls_last key with NULLs rather than mis-cutting it. The
run sits above the values there, so every adjustment reverses: boundaries pull
down from the maximum and the run takes the top partition instead of the
lowest. Same shape, ceilings instead of floors.
nulls_first rank = max(pop - n, even_up, cut + 1)
nulls_last rank = min(pop, even_down, values - (K - 2 - cut))
nulls_first NULLs values K cuts partition sizes
false 60 1..=40 4 [14, 27, 40] 13 / 13 / 13 / 61
false 90 1..=10 4 [4, 7, 10] 3 / 3 / 3 / 91
The even split is now gated on the run actually taking a partition, which it
does only when longer than one partition's share. Ungated it over-spaces the
top boundaries even with no NULLs at all: at K=16 over 100 values it computed
rank 94 where the population rank was 93, so the two null placements
disagreed on identical NULL-free input. Caught by the cross-check test rather
than by reasoning.
Tests: the nulls_last mirror as a property over five shapes, and agreement
between the two placements at every K when nothing is NULL. Mutation-checked
— reusing the nulls_first arithmetic for nulls_last gives `[0, 13, 13, 74]`,
an empty first partition against a run partition swollen by everything the
mirror should have spread, and dropping the guard breaks the agreement test.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Bounds were `ScalarValue` at the surface and `f64` internally, so a `Timestamp` or `Int64` routing key was refused even though nothing about the filter needed a float. Now `ScalarValue` throughout: `WidenedBound` carries it, the predicate builds a `Literal` of whatever type arrived, and the fast path compares with `PartialOrd`. The binary search addresses the array by index via `ScalarValue::try_from_array` rather than reading a `Float64Array`'s values buffer, which costs one `ScalarValue` per probe — 13 for an 8192-row batch, against a linear pass over the whole thing — and works for any ordered type without a per-type dispatch list to keep in step with `sort_key.rs`. Halo widening is typed arithmetic (`ScalarValue::add` / `sub`), so a halo of a type the key cannot be widened by is refused rather than coerced. A `Float64` halo against a `Timestamp` key is a planner bug, and silently coercing a duration is how a nanosecond bound acquires a 256 ns grid. A zero halo is treated as widening by nothing, before any arithmetic. Without that, `Float64(0.0)` — which the scheduler passes for every consumer with no halo — would refuse every non-Float64 key. `as_f64` is gone. Its `Float64(None)` arm rejected a NULL bound, which cannot occur now that `cuts` ranks among the values. Read loose, write tight: this widens what the receiver accepts and leaves the writers alone. Routing anything other than Float64 to it comes later. Not widened here, deliberately: `sorted_on_key` still requires ascending, so descending input takes the mask path. Reversing the search is a separate concern from type width. `non_float64_bounds_are_rejected` becomes `non_float64_bounds_filter_on_both_paths` — an `Int64` key with a zero `Float64` halo selecting `[5, 12)` through the binary search and through the mask, asserting `sorted_on_key` either way so both are exercised. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
A NULL key is absent from a comparison rather than failing one: arrow's kernels leave its mask entry NULL and `filter_record_batch` reads that as exclude, so a nullable routing column lost every NULL row. Not a wrong aggregate — missing rows. `slow_filter` now takes the mask's null entries from the ordering, which is where a NULL's position actually lives. Which partition claims the run comes from the bound, not a partition number: only a globally-outermost partition is unbounded at the end the run occupies, and that survives the scheduler slicing bounds down to a task. So `takes_nulls = nulls_first ? lo.is_none() : hi.is_none()`, and RFE needs no global partition identity. Where the run sits is a fact about the declared order, so construction refuses a nullable key whose input declares no ordering on it — no ordering at all, or an ordering on some other expression. Defaulting there would hand the run to whichever end the default named, by accident rather than by decision. A non-nullable key has no run to place, so it is unaffected. Nothing else in the operator learns about NULLs. Cuts cannot be NULL, halo arithmetic never sees one, and `null_count == 0` short-circuits, so a non-nullable key pays nothing. This matters beyond range partitioning. DataFusion computes a window over the NULL peer group rather than dropping it — for `RANGE BETWEEN 3 PRECEDING AND CURRENT ROW`, NULL rows come back with `count(*) = 2` over the two of them — so the run being indivisible is a correctness requirement for parallel windows, not only a reproducibility argument. Splitting it would hand each BWAG a partial peer group. The fast path keeps its bail on `null_count > 0`, with a TODO recording what the null-capable version buys (sorted nulls stay contiguous with the selection, so the slice stays zero-copy where this copies) and what it costs (plan-declared null placement becomes load-bearing for slicing, which the bail avoids). Tested over both placements at K=3: the run lands in exactly one partition, index 0 under nulls_first and K-1 under nulls_last, with the values unaffected by it riding along; plus the refusal above. Mutation-checked — restoring the drop gives `[0, 0, 0]` against `[3, 0, 0]`, and ignoring `nulls_first` puts the run at the wrong end. Q8 unaffected, as expected: `slow_batches=0` throughout and RFE's elapsed_compute is 378-570us against 459-689us before. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
`split_batch_by_range` sent NULL keys to partition 0 regardless of the sort order, with a TODO admitting it and a test pinning the placeholder. It now takes `nulls_first` and sends the whole run to partition 0 or K-1 accordingly, which is the same bit `RangeFilterExec` reads on the other side when it asks which of its bounds is unbounded. Write and read agree because both derive the answer from the ordering rather than from a number either of them was told. The run goes to one partition, not spread across several: a NULL has no position among the values, only a side, and NULLs cannot be told apart, so no split of the run could be reproduced read-side. Both scatter paths now thread the whole `PhysicalSortExpr` instead of the bare routing expression. The key and its null placement come from `order_by[0]` either way, and carrying them as one value means they cannot be handed in from different places. URRE and ORRE keep rejecting a nullable routing expression. That gate is not waiting on this — the scatter can place a run now — it waits on cut discovery, which reads a T-Digest with no NULL slot and would therefore size partitions from a population missing them. The stale comments claiming otherwise are corrected. `split_routes_nulls_to_partition_zero` becomes `split_routes_the_whole_null_run_to_the_end_it_occupies`, asserting both placements over the same batch. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
`discover_cuts` returns `Vec<ScalarValue>`, URRE and ORRE carry it, and `split_batch_by_range` takes `ScalarValue` boundaries. The values still come from the T-Digest, wrapped in `ScalarValue::Float64`, so this is the pipe widened with the same water flowing: Q8's output is unchanged and any difference is the routing change alone. Routing no longer downcasts to `Float64Array`. A row's partition is now the count of boundaries its key is at or above — the half-open convention counted rather than searched — via one `gt_eq` per boundary, which compares whatever the key happens to be. That trade is measured and it is a loss: O(n·K) against the O(n·log K) binary search it replaced, `scatter_split_time` 45ms against 24-36ms on Q8 at K=8. About 20ms of a ~2.5s query, paid for the type generality that lets a non-Float64 key route at all. Recorded at the call site along with the two ways back, including one that measured *worse* (arrow `cast`/`add` accumulation, 87 to 123ms — those kernels allocate an array per boundary, 24 per batch at K=8, replacing an in-place increment). NULL rows still route by `nulls_first` rather than by their comparison result, which is absent rather than false. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
`MergedRuntimeStats` carries `ScalarValue` cuts and extremes, `RangeRepartitionRouting` carries them to the adapter, and `cut_partitions` takes them along with `ScalarValue` halos. Values still come from the T-Digest, wrapped in `ScalarValue::Float64`, so the routing decisions are unchanged and Q8 is unaffected. Halo widening in `cut_partitions` is now the same typed arithmetic `RangeFilterExec` uses, with the same zero shortcut so a `Float64(0.0)` halo cannot refuse a key of another type. Rather than duplicate it, `widen_below` / `widen_above` move to `range_filter` as `pub(crate)` and both callers share them — halos are functionally RFE's, so the definition belongs where the meaning does. `raw_bounds_from_cuts` stops converting f64 to `ScalarValue`, since it now receives what it used to build. `downstream_halos` returns the halos as RFE already stores them, which deletes `scalar_to_f64` and its "only f64 halos are implemented" refusal. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
`discover_cuts` reads `merged_sort_key_sketch()`, `merge_group` decodes one
`SortKeySketch` per report rather than N×K per-partition T-Digests, and
`cut_partitions` takes each file's extremes from the per-partition `key_min` /
`key_max` the executor already ships. Types were widened in the two commits
before this, so this changes only the source.
Which makes the cuts move, and by the right amount. h2o Q8 at 1e7, K=8:
T-Digest 13.354 25.726 38.111 50.480 62.879 75.238
KLL 13.388 25.674 38.199 50.544 62.852 75.403
Every difference is under 0.2 on a [1, 100] range, inside the two sketches'
combined rank error. The query returned exactly 10,000,000 rows on each of
three iterations, as it must — moving a boundary redistributes rows, it does
not create or destroy them.
`merge_group` refuses a report that carries a sketch with an empty `order_by`
tag, since the key's direction and NULL placement live only there and a sketch
merged under the wrong ordering produces a plausible distribution of nothing.
`stats_to_report` sets the tag exactly when a sketch exists, so the invariant
holds by construction rather than by hope.
`cut_partitions` also learns the case the gates still make unreachable: a file
with a null count and no value range is all NULLs, and belongs wholly to the
partition the run occupies rather than to whatever an overlap check would say
about a range it does not have. `nulls_first` travels with the cuts on
`MergedRuntimeStats` and `RangeRepartitionRouting`, so the file router and the
read-side filter cannot disagree about which end that is.
Fixtures move with the source: reports carry a merged sketch and a sort tag,
partition entries carry key ranges, and the corrupt-wire test now corrupts the
payload the merge actually reads.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Nothing reads it. `RuntimeStatsExec` keeps one sketch, the wire keeps one sketch message, and `datafusion-functions-aggregate-common` leaves both Cargo.toml files. `RuntimeStatsPartitionEntry.sketch` becomes `reserved 3` rather than being reused, so a later field cannot inherit a decoder's expectations of the old one. `QuantileSketchState`, `sketch_to_proto` and `sketch_from_proto` go with it. The construction gates go too, since they existed only for the T-Digest's sake. `try_new` now refuses exactly what `SortKeyCodec` cannot encode, so any fixed-width key is sketchable in either direction, nullable or not. The two tests asserting the old refusals become one asserting a nullable `Float64` and an `Int64` are accepted while a `Utf8` key is still refused by name — the replacement gate is pinned rather than merely absent. `sort_key_sketch_time` becomes `sketch_time`: it no longer has a sibling to be distinguished from. Earlier commit messages quote the longer name. The differential test could not survive a deletion that leaves one implementation. It becomes `the_sketch_sees_every_row_through_the_operator`, still driving 5,000 rows past KLL's k=800 level-0 capacity so compaction runs, and asserting the count, the exact extremes, and that `cuts(8)` yields seven real non-decreasing boundaries. h2o Q8 @1e7, release: 10,000,000 rows on each of three iterations, 2.533s average. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
The executor↔scheduler proto changed shape: `QuantileSketchState` is gone, `RuntimeStatsPartitionEntry.sketch` is reserved, and reports now carry a `SortKeySketchState` alongside per-partition key ranges. A version-1 executor talking to a version-2 scheduler would send a sketch field the scheduler no longer reads and omit the ones it now requires. Strict equality means the mismatch is refused at registration rather than surfacing later as a stage that cannot compute its cuts. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Both operators refused a non-Float64 or nullable routing expression. What the sketch encodes is now the only restriction, so any fixed-width key routes in either direction, nullable or not, and a variable-width key is refused by name rather than by type equality. The nullable half was waiting on cut discovery, which used to read a T-Digest with no NULL slot and would have sized partitions from a population missing the run. That is gone: the run is counted beside the values, sized into the cuts, scattered to the end `nulls_first` names, and read back from there. Which makes the whole chain reachable for the first time. Every piece has been in place and untestable end to end — the NULL-aware cuts, the scatter's placement, the filter's claim on the run, and `cut_partitions` routing a NULLs-only file — because nothing could construct a plan that produced one. Doc corrections that went stale earlier in the chain: URRE's "the impl hardcodes Float64 downcast internally" stopped being true when `split_batch_by_range` started comparing generically, and both operators' constructor docs still promised `Float64`. The three tests asserting the old refusals become two asserting acceptance — a nullable `Float64` and an `Int64` route, a `Utf8` key is refused with "no sort-key encoding". Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
`halo_from_bound` cast every frame bound through `f64`, so `RANGE 3 PRECEDING` on an `Int64` key produced a `Float64` halo — which `RangeFilterExec` now refuses rather than coerces, and which rounds a nanosecond offset onto a 256ns grid at 2020s epoch magnitudes. The bound is the halo. This converts its units into the key's where that means something, and otherwise leaves it alone: an interval is already the delta type a halo is, and casting it to a timestamp would be nonsense. Subtracting an interval from a timestamp yields a timestamp, which is what a widened bound has to be. Whether the pair works is then checked rather than predicted — the widening is attempted on a probe of the key's type, which is the exact operation the filter will perform. A pair that does not typecheck declines the rewrite, so the query runs unparallelized instead of failing at execute time. Zero is exempt: it widens by nothing and `RangeFilterExec` short-circuits before the arithmetic, so `CurrentRow` pairs with any key. That also means `ScalarValue::new_zero` failing for an exotic type must not decline — it means "not known to be zero", so it falls through to the probe. The halo extraction moves below the routing type's derivation, since it now needs it. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
The rule declined every ORDER BY that was not `Float64`, so none of the widening below it could be reached from a query. It now declines only a key the sketch cannot encode, which is the same condition `RuntimeStatsExec`, `URRE` and `ORRE` apply — one definition of "routable", asked in four places rather than four different definitions. Declining, not failing: a `Utf8` key leaves the query on the non-parallel path, as an unsupported frame shape already does. `no_rewrite_on_non_float64_order_key` becomes `rewrites_an_int64_order_key`, and a `Utf8` column joins the test schema so the decline path stays covered by something rather than being asserted about a case that no longer declines. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
`split_batch_by_range` counted one `gt_eq` pass per boundary and accumulated in a scalar loop — O(n·K), which is 256 passes over every batch at the TPC-H `target_partitions=256`. Encode the key column and the boundaries into arrow's row format once per batch, then binary search the boundaries per row: O(n·log K), and `rows.row(i)` is a slice view so no per-row `ScalarValue` is built. `SortField` carries the `SortOptions`, so the encoding places a DESC key's bytes inverted and a NULL's bytes at the end `nulls_first` names. The `null_target` branch and the `is_null` check go, and a DESC key now reads its boundaries in the order the sketch produced them. No boundaries returns the batch itself rather than a `take_arrays` copy of every row. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…ng it
RFE read NULL placement off its child's declared ordering, but it also
advertises `maintains_input_order` with no `required_input_ordering`, so
`EnsureRequirements` was free to sink an unrelated sort beneath it. With a
nullable key the parallel-window rewrite always died on the scheduler:
Context("EnsureRequirements", Internal("RangeFilterExec: routing_expr is
nullable but the input declares no ordering on it, so which partition
holds the NULL run is unknown"))
The child under the narrow filter had become `SortExec: expr=[id ASC]` —
the query's own ORDER BY, pushed down past it. With a non-nullable key the
same pushdown silently cleared `sorted_on_key`, so the binary-search slice
never ran.
`InputOrder` now carries what the cuts' producer said. `Ordered` is
required of the input, so nothing can be planted between that reorders the
rows the placement was stated against; `Unordered` states the run's end
without demanding an order, leaving an unordered range-repartition
upstream legal. `sorted_on_key` still comes off the child, since that is a
claim about the child.
Renames `routing_expr` to `filter_expr`: this operator filters, and the
routing vocabulary belongs to the ops that redistribute rows.
Adds end-to-end coverage over the oracle (DataFusion), rule-off and
rule-on, across nullable and non-nullable keys, NULLS FIRST/LAST, all four
halo shapes, and DESC on its serial-path gate — plus a check that the
rewrite fired, since an all-serial plan would agree on every answer.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…n't say Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
`cuts` blended two rules per boundary and let an extremum pick between
them. Measured over 4.5M (K, nulls, values, cut) combinations:
* nulls_first: the population term never won its `max`. Dead in every
case, because `nulls_outgrow_a_partition` is exactly the condition
that forces the even split higher.
* nulls_last: the population term won its `min` 25,275 times, and every
one made the split worse. 5796 cases change, 5796 improve, 0 regress.
K=3 over 2 NULLs and 3 values gave sizes [0, 2, 3] — an empty
partition beside a double one — where the even split gives [1, 1, 3].
Both branches become one rule: an even split among the partitions the run
leaves, or the population rank shifted past the run.
The rest is readability. Names the magic numbers, unfuses `- 2` into
`last_cut_idx - cut_idx`, turns the closure into a loop, and makes the
clamps mirror — each branch derives its own-side bound from the cuts on
that side, then clamps against the far end of the rank space. `at_ranks`
returns `max` for any rank at or above the count, so the added
`.min(sketch_cnt)` is output-neutral.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
avantgardnerio
marked this pull request as ready for review
August 18, 2026 21:46
Contributor
Author
|
@phillipleblanc FYI |
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
phillipleblanc
approved these changes
Aug 19, 2026
phillipleblanc
left a comment
phillipleblanc
left a comment
Contributor
There was a problem hiding this comment.
Looks good! One minor edge case that we might want to fix first.
Empty cuts mean no value existed for a boundary to name. That happens when every key was NULL and when the stage produced no rows at all — both valid degenerate range partitionings where one partition takes everything. The scheduler read them as a missed sketch and failed the job instead. `SortKeySketchState` already carried the NULL count across the wire, so surfacing it on `MergedRuntimeStats` is enough to tell an all-NULL key from a sketch that never arrived. The zero-row case had a second failure: it returned before parking any routing, leaving the downstream `RangeFilterExec` with no boundary to resolve its bounds from. Only a single-partition stage has no boundary, so only that one keeps its early return. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Range repartitioning routed on a T-Digest, which is
Float64-only, single-column, and has no representation for a NULL. Those three limits were enforced as construction gates on four operators and one planner rule, so a query ordering by anInt64, aTimestamp, or any nullable column could not be range-repartitioned at all.This replaces it with the
SortKeySketchfrom #2294 and removes the gates. Not purely additive: the T-Digest is deleted,QuantileSketchStateleaves the proto, andBALLISTA_PROTOCOL_VERSIONgoes to 2.What can be routed now
Float64nulls_firstRangeFilterExecfast path still ASC-only)f64, cast from any frame boundOne definition of routable,
SortKeyCodec::try_new, asked in four places instead of four different type checks that could drift apart.NULLs get a partition the same way a value does
A NULL has no position among the values, only a side, and NULLs are indistinguishable from one another. So a NULL run cannot be split across partitions and have the split be reproducible on the read side. That is not a special case: a repeated value cannot be split either, and cuts have always been allowed to repeat and produce a dead partition rather than break a run apart.
Sizing therefore targets the population, NULLs included, and only repairs when the run is genuinely too big to fit one partition:
[15, 40, 65][1, 14, 27][1, 4, 7]A run shorter than one partition's share costs no balance at all: it shares the lowest partition with the values below the first cut. A longer one fixes that partition's size, which nothing can improve, and the remaining partitions come out within one row of each other.
One rank per boundary, one of two rules picked by whether the run outgrows a partition, then a floor:
Without the even split the top partition keeps everything the run displaced:
[60, 1, 13, 26]where[60, 13, 13, 14]is available. Without the floor the first two boundaries both name the minimum, because a KLL rank is a cumulative-weight threshold so ranks 0 and 1 coincide.nulls_lastis the mirror:minwith ceilings instead ofmaxwith floors.Every cut is a real value, structurally
cuts()ranks among the values, so there is no rank a NULL could answer. This is not a filter applied afterward, it is unrepresentable.That matters because a NULL boundary fails silently in every consumer.
RangeFilterExecandPerPartitionFilterExecturn a cut intoBinaryExpr(key, Lt, Literal(cut)), and arrow's comparison kernels propagate nulls into the mask whilefilter_record_batchreads a null mask entry as exclude. The rows do not error, they disappear.cut_partitionswould separately computeNULL + halo.Worth being precise about the mechanism, because "SQL semantics" is the wrong name for it: arrow has two comparison families and the predicate path uses the null-propagating one. Arrow's ordering comparisons (the row format,
ArrowNativeTypeOp::compare) give NULL a defined position. The comparison kernels aBinaryExprreaches for do not. SoRangeFilterExecnever asks a NULL to fail a comparison. The partition that claims the run tests for the run directly:Which partition claims the run, without a partition number
RangeFilterExecsees a task-local slice of the bounds, so it cannot know its global partition index. It does not need one: only a globally-outermost partition has an unbounded end, and that survives the scheduler slicing bounds down to a task.The scatter derives the same answer from the same bit (
nulls_first ? 0 : K-1), so write side and read side agree without either being told a partition number, and there is no third place for them to disagree.Construction refuses a nullable key whose plan states no order for it. Without a stated order there is no fact about where the run belongs, and defaulting would hand it to whichever end the default named, by accident.
Halo arithmetic: generalized, and skipped when it is a no-op
Halos are now typed, using
ScalarValue::add/sub, shared betweenRangeFilterExecandcut_partitionsso the two cannot drift on what a halo means. No coercion: aFloat64halo against aTimestampkey is a planner bug, and silently coercing a duration is how a nanosecond bound acquires a 256 ns grid.Two things make that livable rather than restrictive:
A zero halo widens by nothing, and is short-circuited before any arithmetic. The scheduler passes
Float64(0.0)for every consumer with no halo at all, so without this a zero halo of one type would refuse a key of every other type. This is what makes non-window range repartitioning type-agnostic: no halo means no arithmetic means the key type never has to pair with anything.A halo is a delta, not a value of the key's type. For a numeric key those coincide; for a temporal one the delta is an interval. Rather than enumerate which pairs work,
halo_from_boundattempts the exact operation the filter will perform:A pair that does not typecheck declines the rewrite, so the query runs unparallelized rather than failing at execute time.
RANGE INTERVAL '1' DAY PRECEDINGon aTimestampkey is now expressible end to end. Previously every bound was cast throughf64.Wire format
SortKeySketchStatecarriesk, the NULL count, the exact extremes asrepeated ScalarValue, and the compactor stack as one Arrow IPC stream: one row per level, a singleList<Struct<...>>column, items ascending.IPC rather than a packed blob because the arrow schema states what a key is. A multi-column key becomes struct siblings rather than a second payload shape needing a discriminant field, and the key's in-memory representation stays out of the wire format. Bulk items go to IPC, individual scalars go to
ScalarValue: 816 retained items would pay ~10 bytes each asScalarValue, where two extremes pay ~20 bytes total and stay readable in a log line.Levels are sorted before serialization, so a decoder takes ascending order as given rather than being told per level. Direction and NULL placement are deliberately absent: they live once per report in the
order_bytag every consumer already reads.Payload size
Per sketch is larger. Per task is smaller, because the size no longer scales with partition count:
No consumer reads a per-partition distribution.
merge_reportsfolds them all together for global cuts, andcut_partitionsneeds only each file's extremes to route it. So one merged sketch per report is lossless for every consumer that exists, and the executor is where the merge is cheapest since the task already holds every partition's sketch when it builds the message.Measured: 7,460 B encoded against 6,528 B raw at 816 retained items, so 1.14x for IPC framing.
Measurements
h2o window Q8 at 1e7, release, 2 executors, K=8,
max_partitions_per_task=4.Correctness. 10,000,000 rows on every run. Cuts moved by less than the two sketches' combined rank error, which is what a different algorithm approximating the same distribution should look like:
Sketch ingest (
cargo bench --bench quantile_sketch, n=1M):tdigestsort_key_sketch_f64sort_key_sketch_f64_nullssort_key_sketch_i64sort_key_sketch_timestamp_nsFuture directions
Delta encoding is worth ~3.7x on integer keys. Sorted levels make it available. Measured over 816 items:
Floats are near their information-theoretic floor already: 816 sorted samples from a continuum carry about
816 x (log2(range/816) + 1.44)bits, which is ~4.7 KB against a best measured 4,966 B. There is no clever encoding waiting to be found there. Dense integers are a different story, and that is exactly the key family this PR adds. Adopting it bumpsBALLISTA_PROTOCOL_VERSIONrather than needing a version field in the message.Range repartitioning on strings. Outside a window there is no halo, so there is no arithmetic, so nothing about the routing needs a numeric key. Two things block a
Utf8key today:SortKeyCodechas no fixed-width encoding for it (the arrow-row tier is the documented path, priced at 3.75x in the bench), andRangeFilterExecstill gates onnumeric || temporal, which is stricter than its zero-halo path requires. Neither is the halo machinery.One of the two
RuntimeStatsExectaps is redundant. RSE#1 (below the Sort) and RSE#2 (above the ORRE) observe the same multiset, since the ORRE redistributes rows without dropping any. RSE#1 exists to feed local cut discovery in-process and never crosses the wire; RSE#2 ships. What RSE#2 uniquely provides is per-output-partition extremes for file routing, and those are a min/max fold rather than a sketch. Removing one full sketch ingest from the hot path is available.cut_partitionslearned a case nothing can reach yet. A file with a null count and no value range is all NULLs, and belongs to the partition the run occupies rather than to whatever an overlap check says about a range it does not have.The fast path could handle NULLs. Sorted input puts the run at one end, so the values occupy a contiguous span and the run stays contiguous with the selection, keeping the slice zero-copy where it currently copies. It would make plan-declared null placement load-bearing for slicing, which today's bail deliberately avoids, so it wants the existing path counters as its evidence.
sorted_on_keystill requires ascending. A descending input silently takes the predicate path. Correct but slower, and visible inslow_batches.What this does not do
No query in either benchmark suite exercises the widened keys. Both suites contain exactly two finite-RANGE window queries, both h2o
ORDER BY v2, a non-nullFloat64. Every unpartitioned h2o query on theInt64id3column uses a ROWS frame or an unbounded RANGE, so they are excluded by frame shape rather than key type. TPC-DS has no finite-RANGE frame at all.So Q8 proves no regression, not new capability. The
Int64, nullable, andTimestamp-with-interval paths are covered by unit tests. A bounded RANGE frame over a timestamp is the shape most likely to matter in production and needs a dataset we do not have.