feat(core): add OrderedRangeRepartitionExec — N sorted overlapping → K sorted disjoint

[avantgardnerio]

Summary

Adds OrderedRangeRepartitionExec — the sorted sibling of UnorderedRangeRepartitionExec (#2123). Takes N locally-sorted input partitions whose value ranges may overlap; produces K range-disjoint output partitions where each output is fully sorted on the routing expression. Concatenating the K outputs in order yields a globally-sorted stream.

Reuses range_repartition_common.rs verbatim (landed with #2123): same discovery mechanism — walk child subtree, find sibling RuntimeStatsExec, snapshot T-Digest, compute K − 1 quantile cuts. Same fallback semantics: any discovery failure → single-bucket routing where every row lands in output 0.

Execution model — N × K channels + K k-way merges

• N scatter tasks. Each reads one sorted input, calls split_batch_by_range on every batch, forwards sub-batches to K per-input senders (one per output partition).
• K output-partition streams. Each merges N sub-streams via DataFusion's StreamingMergeBuilder — the SortPreservingMerge internals, no custom heap machinery here.
• Backpressure. Bounded mpsc::channel(2) per input × output, symmetric to the Unordered variant. Total buffered at rest ≤ N × K × 2 batches.

Constructor requires input.output_ordering() to lead with the routing expression — otherwise the merger would produce garbled output. PlanProperties::eq_properties advertises each output partition as sorted on order_by, so downstream operators (BWAG, HaloDrop) can rely on that claim without inserting a redundant SortExec.

Scope

Just the operator + serde plumbing. Nothing inserts OrderedRangeRepartitionExec into a plan yet — no rewrite rule, no scheduler hook. Those land with the parallel-window detection rule alongside the Unordered sibling.

Included:

• ordered_range_repartition.rs — the operator + 8 unit tests.
• Proto: OrderedRangeRepartitionExecNode { repeated PhysicalSortExprNode order_by = 1; uint32 output_partitions = 2; } on BallistaPhysicalPlanNode.oneof (tag 9; tag 8 is Unordered).
• Codec encode/decode registration on BallistaPhysicalExtensionCodec.
• One serde roundtrip test.
• No new external deps.
• No changes to range_repartition_common.rs — reused verbatim.

Drift from source branch

• RuntimeStatsExec now rejects nullable routing expressions at construction (T-Digest has no NULL slot; tightened during feat(core): add UnorderedRangeRepartitionExec — value-range router over unordered inputs #2123's upstreaming). The original test suite declared v2 as nullable and typed the batch builder as Vec<Option<f64>> but never actually populated a None. Tightened schema + helper types to non-nullable f64; will loosen again when we swap T-Digest for KLL and NULL becomes a first-class routing key.

Tests (9 new)

8 operator unit tests covering: constructor validation (empty order_by, unsorted input, mismatched sort key), end-to-end row conservation across two overlapping inputs, per-output sortedness, range-disjointness of outputs, no-stats fallback, and the double-execute guard.

1 serde roundtrip test (single-key ORDER BY through the SortExec-wrap that satisfies the ordering claim).

Test plan

• cargo test -p ballista-core — 175/175 pass (25 in the range-repartition modules).
• cargo clippy --all-targets clean across the workspace.
• cargo fmt --all clean.

Next in the stack

The rewrite rule (ParallelWindowDetectRule) that inserts either operator above BoundedWindowAggExec based on whether the wrapping plan needs ordering preserved.

🤖 Generated with Claude Code

[avantgardnerio]

@phillipleblanc PTAL when you have a moment 🙌

[andygrove]

Thanks @avantgardnerio. This is purely additive and doesn't break any APIs, so LGTM. I did an AI review and nothing was flagged.

[andygrove]

why the underscore? just curious

[avantgardnerio]

It's just to signify that no one reads this. It's just a struct member purely so it can get held until the struct is dropped.