Skip to content

feat(core): add UnorderedRangeRepartitionExec — value-range router over unordered inputs#2123

Open
avantgardnerio wants to merge 5 commits into
apache:mainfrom
avantgardnerio:brent/unordered-range-repartition-exec
Open

feat(core): add UnorderedRangeRepartitionExec — value-range router over unordered inputs#2123
avantgardnerio wants to merge 5 commits into
apache:mainfrom
avantgardnerio:brent/unordered-range-repartition-exec

Conversation

@avantgardnerio

Copy link
Copy Markdown
Contributor

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 (feat(core): RuntimeStatsExec — passthrough tap for row counts + quantile sketches #2094) — the tap that accumulates the T-Digest as batches flow past.
  2. 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.
  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.rspub(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 feat(core): RuntimeStatsExec — passthrough tap for row counts + quantile sketches #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

  • cargo test -p ballista-core — 158/158 pass (14 new).
  • cargo clippy --all-targets clean across the workspace.
  • 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

avantgardnerio and others added 2 commits July 21, 2026 08:47
…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
avantgardnerio marked this pull request as ready for review July 21, 2026 16:18
@avantgardnerio

Copy link
Copy Markdown
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 phillipleblanc left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks! I think the only blocking item from my side is dropping the JoinHandle for scatter_input_partition. Everything else looks good to me.

Comment thread ballista/core/src/execution_plans/unordered_range_repartition.rs Outdated
Comment thread ballista/core/src/execution_plans/unordered_range_repartition.rs Outdated
Comment thread ballista/core/src/execution_plans/range_repartition_common.rs
avantgardnerio and others added 2 commits July 22, 2026 04:53
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>
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants