feat(core): RangeShuffleReaderExec — ordering-preserving shuffle reader (purely additive)

[avantgardnerio]

Purely additive.

This is a new ShuffleReader for merging sorted partitions from previous stages. This allows for ordered range-re-partitioning between stages, when coupled with either a PerPartitionFilterExec its much faster range based cousin RangeFilterExec which will probably be the next PR in the series.

Extracted from #2223 (a working end-to-end parallel-window branch) as
the an atomic part of the epic to land parallel windows incrementally,
one reviewable piece at a time.

Sibling to ShuffleReaderExec: keeps each upstream source alive as its
own stream and feeds all N into a StreamingMerge keyed on the child's
declared output ordering.

Scope

• range_shuffle_reader.rs — operator impl + tests
• proto message + oneof slot 12
• serde encode/decode + roundtrip test
• three shuffle_reader helpers promoted to pub(crate) so RSR can
  reuse them: local_remote_read_split, fetch_partition_local,
  fetch_partition_remote

Test plan

• cargo clippy --all-targets --workspace clean
• cargo fmt --all -- --check clean
• cargo test --workspace — full suite green including new
  test_range_shuffle_reader_exec_roundtrip

Future work

Today the reader pulls whole upstream files and lets StreamingMerge
do the work. Once value-indexed shuffle files land end-to-end (writer
side in #2204), the reader will consult per-file ValueIndex offsets
and fetch only the byte ranges covering its target output partition —
dropping the "read the whole file to throw most of it away" cost that
dominates when K is large.

[avantgardnerio]

@phillipleblanc FYI

[andygrove]

Thanks Brent, this is a clean additive piece and I'm happy to see the epic broken up this way.

Checked it against the user personas doc and I don't see any red flags. Nothing constructs RangeShuffleReaderExec outside the tests and the serde round-trip, so no plan produced today changes shape or results, and none of the stage/task or AQE machinery is touched. No public API breaks either. Proto tag 12 is a fresh number with nothing renumbered or reused, so the wire format stays backward compatible for anything that doesn't emit the new node. The three shuffle_reader helpers going private to pub(crate) widens rather than narrows, which is fine.

One gap worth closing, either here or in the follow-up. DefaultExecutionEngine::create_query_stage_exec in ballista/executor/src/execution_engine.rs only downcasts to ShuffleReaderExec when it late-binds work_dir and client_pool, so RangeShuffleReaderExec never gets either. Nothing plants the operator yet so it can't bite today, but as soon as the planner side lands, the first task carrying one will fail with "RangeShuffleReaderExec work dir should have been set by executor". Extending that transform is only a few lines and it would make this piece actually runnable on its own. A TODO pointing at the follow-up works too if you'd rather keep the diff tight.

No benchmark numbers needed on this one since nothing in the diff can reach a running query. The follow-up that wires this into the planner will want them though, since swapping a coalescing reader for a k-way merge changes both plan shape and the memory profile of a stage.

Approving so this isn't blocked.

[avantgardnerio]

A TODO pointing at the follow-up works too if you'd rather keep the diff tight.

Thanks Andy! I went with the TODO for now. I'll land the planner rule soon and benchmark on my k8s cluster according to the benchmarking guide.