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

ballista/core/proto/ballista.proto

@@ -53,6 +53,7 @@ message BallistaPhysicalPlanNode {
     ChaosExecNode chaos_exec = 5;
     RuntimeStatsExecNode runtime_stats = 6;
     BufferExecNode buffer = 7;
+    UnorderedRangeRepartitionExecNode unordered_range_repartition = 8;
   }
 }
 
@@ -91,6 +92,21 @@ enum BufferMode {
   // Add here as they land in Rust.
 }
 
+// Value-range router over 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 using cuts discovered at
+// runtime from a sibling RuntimeStatsExec's quantile sketch. The child
+// plan is plumbed by the framework as `inputs[0]`.
+message UnorderedRangeRepartitionExecNode {
+  // Lexicographic ORDER BY carried through from the wrapping window
+  // operator. Routing keys off the first entry today; the full ordering is
+  // preserved so downstream operators (SortExec, BWAG) that need it keep
+  // working.
+  repeated datafusion.PhysicalSortExprNode order_by = 1;
+  // K — number of output partitions. Must be ≥ 2.
+  uint32 output_partitions = 2;
+}
+
 message ChaosExecNode {
   double failure_probability = 1;
   string fault_type = 2;

ballista/core/src/execution_plans/mod.rs

@@ -22,11 +22,13 @@ mod buffer;
 mod chaos_exec;
 mod distributed_explain_analyze;
 mod distributed_query;
+mod range_repartition_common;
 mod runtime_stats;
 mod shuffle_reader;
 mod shuffle_writer;
 mod shuffle_writer_trait;
 pub mod sort_shuffle;
+mod unordered_range_repartition;
 mod unresolved_shuffle;
 
 use std::path::{Path, PathBuf};
@@ -44,6 +46,7 @@ pub use shuffle_writer::ShuffleWriterExec;
 pub use shuffle_writer::compute_global_output_partition_ids;
 pub use shuffle_writer_trait::ShuffleWriter;
 pub use sort_shuffle::SortShuffleWriterExec;
+pub use unordered_range_repartition::UnorderedRangeRepartitionExec;
 pub use unresolved_shuffle::UnresolvedShuffleExec;
 
 use crate::JobId;