avantgardnerio opened a new pull request, #2169:
URL: https://github.com/apache/datafusion-ballista/pull/2169
## 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.
## Why — Stage 2 → Stage 3 of the parallel-window path
The parallel-window plan runs in three stages:
1. **Stage 1** — sample. `RuntimeStatsExec` (#2094) accumulates a T-Digest
as batches flow past. `BufferExec` (#2095) holds long enough for representative
samples to arrive.
2. **Stage 2** — sort each input partition locally. Cheap and embarrassingly
parallel.
3. **Stage 3** — one `BoundedWindowAggExec` per output partition, each fed a
*globally-sorted* view of its value range.
`UnorderedRangeRepartitionExec` covers the Stage-1→Stage-3 path for cases
where sortedness is destroyed anyway (aggregate windows). This PR covers the
Stage-2→Stage-3 path where downstream operators need the sort preserved.
## 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 #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
- [x] `cargo test -p ballista-core` — 175/175 pass (25 in the
range-repartition modules).
- [x] `cargo clippy --all-targets` clean across the workspace.
- [x] `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](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]
