avantgardnerio opened a new pull request, #2123:
URL: https://github.com/apache/datafusion-ballista/pull/2123
## Summary
Adds `UnorderedRangeRepartitionExec`: a physical operator that reads P input
partitions with no sort assumption, evaluates the first ORDER BY expression per
row, and routes each row to one of K output partitions under the half-open
convention: partition `p` owns `[cut[p-1], cut[p])` with virtual `-∞`/`+∞`
sentinels on the ends.
Also adds `range_repartition_common.rs` — the shared building blocks
(`discover_cuts`, `find_runtime_stats`, `preserves_distribution`,
`split_batch_by_range`, `broadcast_error`) that the pending
`OrderedRangeRepartitionExec` follow-up will also consume.
## Why — closing the loop on #2094 + #2095
Sketch-based routing needs three pieces:
1. **`RuntimeStatsExec`** (#2094) — the tap that accumulates the T-Digest as
batches flow past.
2. **`BufferExec`** (#2095) — the dam that holds batches back long enough
for the tap to see representative data.
3. **This PR** — the router that reads the sketch and does the actual
repartitioning.
Boundaries are discovered *at runtime*, not baked in at plan time. On the
first batch to arrive from any input partition, the operator walks its own
child subtree for a matching sibling `RuntimeStatsExec`, snapshots its merged
T-Digest, and computes `K − 1` quantile cuts at `1/K, 2/K, …, (K−1)/K`. All
batches (including the one that triggered the snapshot) then route through
those cuts.
The walker only descends through distribution-preserving operators
(`preserves_distribution` whitelist). `BufferExec` is on the whitelist so a Dam
between the tap and this router doesn't hide the sketch. `FilterExec`, unions,
etc. are *not* on it — they'd stale the distribution.
Every discovery failure path (no matching stats, no sketch, empty sketch,
mutex poisoning) falls back to a single-bucket routing that lands every row in
output partition 0. Runtime routing must never crash — degraded-but-alive beats
the alternative.
## Scope
Just the operator, its shared common module, proto/serde plumbing, and
tests. Nothing inserts `UnorderedRangeRepartitionExec` into a plan yet — no
rewrite rule, no scheduler hook. Those land with the parallel-window detection
rule alongside the `OrderedRangeRepartitionExec` sibling.
Included:
- `range_repartition_common.rs` — `pub(super)` helpers; imports `BufferExec`
+ `RuntimeStatsExec` from main.
- `unordered_range_repartition.rs` — the operator itself + 12 unit tests.
- Proto: `UnorderedRangeRepartitionExecNode { repeated PhysicalSortExprNode
order_by = 1; uint32 output_partitions = 2; }` on
`BallistaPhysicalPlanNode.oneof` (tag 8; tags 6 and 7 taken by
`RuntimeStatsExec` and `BufferExec`).
- Codec encode/decode registration on `BallistaPhysicalExtensionCodec`.
- No new external deps.
## Drift from source branch
- `RuntimeStatsExec::merged_quantile_sketch()` returned `Option<TDigest>`
when this code was authored; on main it returns `Result<Option<TDigest>>`
(mutex-poisoning was made explicit during the #2094 upstreaming). Adapted
`discover_cuts` to distinguish `Ok(None)` (sketch not allocated) from `Err(_)`
(snapshot failed) in the warn message; both still degrade to single-bucket
fallback.
- `RuntimeStatsExec` now rejects nullable routing expressions at
construction (T-Digest has no NULL slot). The original test suite declared `v2`
as nullable and typed the batch builder as `Vec<Option<f64>>` but never
actually populated a `None`. Tightened the 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 (14 new)
12 operator unit tests covering constructor validation, output-partitioning
shape, discovery walker's descend/refuse policy, end-to-end row conservation,
range-disjointness of outputs, multi-input-partition handling, and every
discovery-failure fallback path.
2 serde roundtrip tests (single-key + multi-key ORDER BY) — the wire format
must not silently drop non-primary ORDER BY entries (they're the substrate for
the ROWS-frame follow-up).
## Test plan
- [x] `cargo test -p ballista-core` — 158/158 pass (14 new).
- [x] `cargo clippy --all-targets` clean across the workspace.
- [x] `cargo fmt --all` clean.
## Next in the stack
`OrderedRangeRepartitionExec` — same discovery mechanism, adds per-output
k-way merge to preserve sort order. Reuses this PR's
`range_repartition_common.rs` verbatim.
🤖 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]
