feat(core): add UnorderedRangeRepartitionExec — value-range router over unordered inputs#2123
Open
avantgardnerio wants to merge 5 commits into
Open
Conversation
…er unordered inputs 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. 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 sibling RuntimeStatsExec (apache#2094), snapshots its merged T-Digest, and computes K-1 quantile cuts. Every failure path (no matching stats, no sketch, empty sketch, mutex poisoning) falls back to a single-bucket routing that never crashes. Includes: - New `range_repartition_common.rs` with the shared building blocks (`discover_cuts`, `find_runtime_stats`, `preserves_distribution`, `split_batch_by_range`, `broadcast_error`). Reused by the pending OrderedRangeRepartitionExec follow-up. - `preserves_distribution` whitelists BufferExec (apache#2095) as a descend-through node so the discovery walker sees past a Dam. - Proto: `UnorderedRangeRepartitionExecNode` at BallistaPhysicalPlanNode tag 8. - Codec encode/decode + two serde roundtrip tests (single-key + multi-key). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
…Exec Preemptive polish before opening the PR for review — matches the pattern of reviewer asks Phillip made on apache#2094 and apache#2095: - Implement all 6 algebraic ExecutionPlan methods explicitly (required_input_ distribution, required_input_ordering, maintains_input_order, benefits_from_ input_partitioning, partition_statistics, cardinality_effect), each with a docstring justifying the value. maintains_input_order=false and partition_statistics=Unknown are the two that materially differ from the merged passthrough ops (RuntimeStatsExec, BufferExec). - Reject nullable routing expressions in try_new. The sibling RuntimeStatsExec already enforces this at plan-assembly time, but defense-in-depth catches the no-stats-pair path where discovery would silently fall back to single-bucket. TODO marker points at the KLL swap that lifts the restriction. - Add try_new_rejects_nullable_routing_key covering the new invariant. Includes two authorial touch-ups picked up along the way: a tone softening on the preserves_distribution docstring and a TODO on split_batch_by_range's row-by-row loop pointing at a vectorized Arrow rewrite. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
avantgardnerio
marked this pull request as ready for review
July 21, 2026 16:18
Contributor
Author
|
@phillipleblanc we're finally getting into the interesting stuff 😄 |
…xec module docs Mirrors DataFusion's RepartitionExec style. Shows P input partitions → the RuntimeStatsExec/BufferExec/URRE chain → K half-open value-range outputs, with the discovery walker's descent traced explicitly against the actual child operators rather than a side channel. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
phillipleblanc
left a comment
Contributor
There was a problem hiding this comment.
Thanks! I think the only blocking item from my side is dropping the JoinHandle for scatter_input_partition. Everything else looks good to me.
Addresses two review comments on PR apache#2123: 1. `scatter_input_partition` was spawned as a bare `tokio::spawn` whose `JoinHandle` was discarded. A panic inside the body silently dropped the sender, and downstream consumers saw a clean EOF while data was lost. Wrap it in `SpawnedTask` and add a `wait_for_task` helper (in `range_repartition_common.rs` — the ordered variant will share it) that awaits the scatter task and broadcasts any panic or DFError to every output channel. Hold the wait-tasks in `DispatchState._drop_helper` so dropping the exec aborts all P scatter tasks. 2. `PlanProperties::new` defaults to `Lazy` / `NonCooperative`; DRR eagerly drives its inputs from the moment `execute()` is called and yields at per-output channel sends, so set `EvaluationType::Eager` + `SchedulingType::Cooperative` — same combination DataFusion's `RepartitionExec` uses. New test `scatter_task_panic_surfaces_as_error` uses an inline `PanickingSourceExec` whose stream panics on first poll, and asserts at least one output partition surfaces the panic as `Err(..)` rather than a spurious clean EOF. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
…urceExec Follows 6b300df. Two changes: 1. Collapse two SpawnedTasks to one. DataFusion's model spawned an inner task for the scatter body and an outer task that awaited it — the outer task only existed to observe panics via `JoinError::is_panic()`. A single task using `AssertUnwindSafe(fut).catch_unwind()` does the same job in one spawn. Replaces `wait_for_task` with `guarded_scatter(fut, senders)` and adds `panic_payload_message` for `&str` / `String` payload downcasts so the original panic message survives. 2. Move `PanickingSourceExec` into `range_repartition_common::test_util`. The ordered variant will exercise the same `guarded_scatter` panic path; sharing the source keeps the two operator test files from re-declaring it. Also renamed the synthetic message from a joke to `PanickingSourceExec: synthetic test panic` (surfaced via a `SYNTHETIC_PANIC_MESSAGE` const so tests can assert-round-trip through the payload downcast). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
phillipleblanc
approved these changes
Jul 22, 2026
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
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.
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: partitionpowns[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 pendingOrderedRangeRepartitionExecfollow-up will also consume.Why — closing the loop on #2094 + #2095
Sketch-based routing needs three pieces:
RuntimeStatsExec(feat(core): RuntimeStatsExec — passthrough tap for row counts + quantile sketches #2094) — the tap that accumulates the T-Digest as batches flow past.BufferExec(feat(core): BufferExec — generic flow-control op with a memory-pool Dam mode #2095) — the dam that holds batches back long enough for the tap to see representative data.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 computesK − 1quantile cuts at1/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_distributionwhitelist).BufferExecis 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
UnorderedRangeRepartitionExecinto a plan yet — no rewrite rule, no scheduler hook. Those land with the parallel-window detection rule alongside theOrderedRangeRepartitionExecsibling.Included:
range_repartition_common.rs—pub(super)helpers; importsBufferExec+RuntimeStatsExecfrom main.unordered_range_repartition.rs— the operator itself + 12 unit tests.UnorderedRangeRepartitionExecNode { repeated PhysicalSortExprNode order_by = 1; uint32 output_partitions = 2; }onBallistaPhysicalPlanNode.oneof(tag 8; tags 6 and 7 taken byRuntimeStatsExecandBufferExec).BallistaPhysicalExtensionCodec.Drift from source branch
RuntimeStatsExec::merged_quantile_sketch()returnedOption<TDigest>when this code was authored; on main it returnsResult<Option<TDigest>>(mutex-poisoning was made explicit during the feat(core): RuntimeStatsExec — passthrough tap for row counts + quantile sketches #2094 upstreaming). Adapteddiscover_cutsto distinguishOk(None)(sketch not allocated) fromErr(_)(snapshot failed) in the warn message; both still degrade to single-bucket fallback.RuntimeStatsExecnow rejects nullable routing expressions at construction (T-Digest has no NULL slot). The original test suite declaredv2as nullable and typed the batch builder asVec<Option<f64>>but never actually populated aNone. Tightened the schema + helper types to non-nullablef64; 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
cargo test -p ballista-core— 158/158 pass (14 new).cargo clippy --all-targetsclean across the workspace.cargo fmt --allclean.Next in the stack
OrderedRangeRepartitionExec— same discovery mechanism, adds per-output k-way merge to preserve sort order. Reuses this PR'srange_repartition_common.rsverbatim.🤖 Generated with Claude Code