feat(core): RangeFilterExec — sorted-input per-partition range filter (purely additive)

ballista/core/proto/ballista.proto

@@ -58,6 +58,7 @@ message BallistaPhysicalPlanNode {
     PerPartitionFilterExecNode per_partition_filter = 10;
     PartitionedBoundedWindowAggExecNode partitioned_bounded_window_agg = 11;
     RangeShuffleReaderExecNode range_shuffle_reader = 12;
+    RangeFilterExecNode range_filter = 13;
   }
 }
 
@@ -135,6 +136,28 @@ message PerPartitionFilterExecNode {
   repeated datafusion.PhysicalExprNode predicates = 1;
 }
 
+// Filter inputs with a per-input-partition half-open range predicate
+// widened by `halo_lo` / `halo_hi`. `raw_bounds[k]` is the cut range for
+// input partition `k` before halo widening; RFE widens internally at
+// resolve time. Zero halo recovers the exact range-repartition trim used
+// above `ShuffleReaderExec`; non-zero halo widens each partition's read
+// range to include a boundary "context" band (bounded RANGE-frame windows).
+// The child plan is plumbed by the framework as `inputs[0]` during decode.
+// Serialization requires bounds to be resolved.
+message RangeFilterExecNode {
+  datafusion.PhysicalExprNode routing_expr = 1;
+  datafusion_common.ScalarValue halo_lo = 2;
+  datafusion_common.ScalarValue halo_hi = 3;
+  repeated RangeBound raw_bounds = 4;
+}
+
+// Half-open `[lo, hi)` cut range for one input partition. Either side may be
+// unset to signal ±∞.
+message RangeBound {
+  datafusion_common.ScalarValue lo = 1;
+  datafusion_common.ScalarValue hi = 2;
+}
+
 // Wrapper for `BoundedWindowAggExec` that overrides
 // `required_input_distribution` to `Unspecified` — see the module doc on
 // `execution_plans::partitioned_bounded_window_agg` for what makes that safe.

ballista/core/src/execution_plans/mod.rs

@@ -26,6 +26,7 @@ mod ordered_range_repartition;
 mod partitioned_bounded_window_agg;
 mod per_partition_filter;
 pub mod plan_algebra;
+mod range_filter;
 mod range_repartition_common;
 mod range_shuffle_reader;
 mod runtime_stats;
@@ -47,6 +48,7 @@ pub use ordered_range_repartition::OrderedRangeRepartitionExec;
 pub use partitioned_bounded_window_agg::PartitionedBoundedWindowAggExec;
 pub use per_partition_filter::{PerPartitionFilterExec, range_partition_predicates};
 pub use plan_algebra::{preserves_distribution, preserves_partitioning};
+pub use range_filter::{RangeBound, RangeFilterExec, WidenedBound};
 pub use range_shuffle_reader::RangeShuffleReaderExec;
 pub use runtime_stats::{
     MergedRuntimeStats, RuntimeStatsExec, TaskRuntimeStats,