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

[avantgardnerio]

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.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 h2o source branch

• RuntimeStatsExec::merged_quantile_sketch() returned Option<TDigest> when the h2o branch 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 h2o 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]

Closing — pushed to origin (apache/datafusion-ballista) by mistake instead of the avantgardnerio fork. Reopening as a fork PR.