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

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

Adds `RangeFilterExec`: filter that applies a per-input-partition
half-open range predicate, widened by the operator's halo. Sibling of
`PerPartitionFilterExec` (which handles arbitrary predicates); RFE
specialises to `[lo, hi)` ranges over a `routing_expr` and gets a
fast path for sorted inputs (binary-search slice + Arc-clone
pass-through when a batch is entirely inside/outside the window).

- Bounds are late-bound via `resolve_bounds`; wire encode refuses
  before resolution. Constructors: `try_new_pending`, `try_new_resolved`.
- Halos + bounds use `ScalarValue` at the API + proto surface (generic
  over Arrow primitives); internal impl is Float64-only today and
  errors on non-Float64 until KLL widens support.
- No in-tree callers — this is purely additive.
- Includes serde roundtrip + fast/slow path unit tests.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* refactor(core): scope DataType import to tests, drop _touch_datatype

DataType was only referenced textually in the test module (via
`DataType::Float64` in schema builders); the non-test build treated
the top-level import as unused, papered over by a `#[allow(dead_code)]
fn _touch_datatype(_: DataType)`. Move the import into `mod tests`
and remove the workaround.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* docs(core): pub-export RangeBound + WidenedBound so RangeFilterExec signatures render as rustdoc links

Both aliases were reachable only through structural spelling — `Vec<(Option<ScalarValue>, Option<ScalarValue>)>` for the raw form, `Vec<(Option<f64>, Option<f64>)>` for the widened form — because `range_filter` is a private module and neither alias was `pub`-re-exported. Rustdoc rendered `try_new_resolved` / `raw_bounds` / `widened_bounds` signatures with bare unlinkable names as a result. Bumping both to `pub` and re-exporting from `execution_plans` restores the signature → alias-page links.

Addresses andygrove's review nits on #2262.

---------

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: avantgardnerio

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,