feat(core): add OrderedRangeRepartitionExec — N sorted overlapping → K sorted disjoint

ballista/core/proto/ballista.proto

@@ -54,9 +54,26 @@ message BallistaPhysicalPlanNode {
     RuntimeStatsExecNode runtime_stats = 6;
     BufferExecNode buffer = 7;
     UnorderedRangeRepartitionExecNode unordered_range_repartition = 8;
+    OrderedRangeRepartitionExecNode ordered_range_repartition = 9;
   }
 }
 
+// Value-range router over N locally-sorted overlapping input partitions.
+// Redistributes them into K range-disjoint output partitions where each
+// output is fully sorted on the routing expression. Uses N × K scatter
+// channels feeding K per-output k-way merges (via DataFusion's SPM
+// internals). Discovery of cut boundaries is identical to the unordered
+// variant — walk the child subtree for a matching `RuntimeStatsExec`, snap
+// its T-Digest.
+message OrderedRangeRepartitionExecNode {
+  // Lexicographic ORDER BY. Routing keys off the first entry; the full
+  // ordering is preserved for downstream operators. Must match the
+  // input's declared `output_ordering` at plan time.
+  repeated datafusion.PhysicalSortExprNode order_by = 1;
+  // K — number of output partitions. Must be ≥ 2.
+  uint32 output_partitions = 2;
+}
+
 // Runtime-stats tap. Passes batches through unmodified while accumulating:
 //   * per-partition row counts (always)
 //   * a quantile sketch over the first ORDER BY expression's Float64 values

ballista/core/src/execution_plans/mod.rs

@@ -22,6 +22,7 @@ mod buffer;
 mod chaos_exec;
 mod distributed_explain_analyze;
 mod distributed_query;
+mod ordered_range_repartition;
 mod range_repartition_common;
 mod runtime_stats;
 mod shuffle_reader;
@@ -38,6 +39,7 @@ pub use chaos_exec::ChaosExec;
 use datafusion::common::exec_err;
 pub use distributed_explain_analyze::DistributedExplainAnalyzeExec;
 pub use distributed_query::{DistributedQueryExec, execute_physical_plan};
+pub use ordered_range_repartition::OrderedRangeRepartitionExec;
 pub use runtime_stats::{RuntimeStatsExec, sketch_from_proto, sketch_to_proto};
 pub use shuffle_reader::{CoalescePlan, PartitionGroup, ShuffleReaderExec};
 pub use shuffle_reader::{stats_for_partition, stats_for_partitions};