diff --git a/.cursor/rules.md b/.cursor/rules.md
new file mode 100644
index 0000000000..4ee2be4ad3
--- /dev/null
+++ b/.cursor/rules.md
@@ -0,0 +1,5 @@
+When making a code review never link to any issue or a pull request
+in any GitHub repository!
+
+Do not review AI agents' config files like CLAUDE.md, AGENTS.md or rules.md!
+
diff --git a/.gemini/rules.md b/.gemini/rules.md
new file mode 100644
index 0000000000..4ee2be4ad3
--- /dev/null
+++ b/.gemini/rules.md
@@ -0,0 +1,5 @@
+When making a code review never link to any issue or a pull request
+in any GitHub repository!
+
+Do not review AI agents' config files like CLAUDE.md, AGENTS.md or rules.md!
+
diff --git a/AGENTS.md b/AGENTS.md
new file mode 100644
index 0000000000..4ee2be4ad3
--- /dev/null
+++ b/AGENTS.md
@@ -0,0 +1,5 @@
+When making a code review never link to any issue or a pull request
+in any GitHub repository!
+
+Do not review AI agents' config files like CLAUDE.md, AGENTS.md or rules.md!
+
diff --git a/CLAUDE.md b/CLAUDE.md
new file mode 100644
index 0000000000..4ee2be4ad3
--- /dev/null
+++ b/CLAUDE.md
@@ -0,0 +1,5 @@
+When making a code review never link to any issue or a pull request
+in any GitHub repository!
+
+Do not review AI agents' config files like CLAUDE.md, AGENTS.md or rules.md!
+
diff --git a/ballista/core/src/execution_plans/mod.rs b/ballista/core/src/execution_plans/mod.rs
index 0b9e89104f..659108e52b 100644
--- a/ballista/core/src/execution_plans/mod.rs
+++ b/ballista/core/src/execution_plans/mod.rs
@@ -41,11 +41,13 @@ 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 per_partition_filter::PerPartitionFilterExec;
+pub use per_partition_filter::{PerPartitionFilterExec, range_partition_predicates};
 pub use runtime_stats::{
     MergedRuntimeStats, RuntimeStatsExec, TaskRuntimeStats,
-    collect_reports as collect_runtime_stats_reports, log_merged_runtime_stats,
-    merge_reports as merge_runtime_stats_reports, sketch_from_proto, sketch_to_proto,
+    collect_reports as collect_runtime_stats_reports, compute_overlapping_locations,
+    find_range_repartition_routing_expr, log_merged_runtime_stats,
+    merge_reports as merge_runtime_stats_reports, overlap_remap_partitions,
+    plan_contains_range_repartition, sketch_from_proto, sketch_to_proto,
 };
 pub use shuffle_reader::{CoalescePlan, PartitionGroup, ShuffleReaderExec};
 pub use shuffle_reader::{stats_for_partition, stats_for_partitions};
diff --git a/ballista/core/src/execution_plans/per_partition_filter.rs b/ballista/core/src/execution_plans/per_partition_filter.rs
index 8583e616cc..e824a33760 100644
--- a/ballista/core/src/execution_plans/per_partition_filter.rs
+++ b/ballista/core/src/execution_plans/per_partition_filter.rs
@@ -18,10 +18,11 @@
 //! Filter with a distinct predicate per input partition.
 //!
 //! `FilterExec` in DataFusion carries a single predicate applied to every
-//! partition. That's wrong for the URRE-consuming reader in the adaptive
-//! range shuffle: each downstream partition `k` needs a range predicate
-//! `cuts[k-1] <= key < cuts[k]` unique to that partition, so straddling
-//! sub-parts from the producer are trimmed to just partition `k`'s slice.
+//! partition. That's wrong for the range-repartition-consuming reader in
+//! the adaptive range shuffle: each downstream partition `k` needs a range
+//! predicate `cuts[k-1] <= key < cuts[k]` unique to that partition, so
+//! straddling sub-parts from the producer are trimmed to just partition
+//! `k`'s slice.
 //!
 //! One-task-per-downstream-partition + plain `FilterExec` would work but
 //! defeats vcore packing (`K` tasks instead of `K / vcores`). This operator
@@ -270,6 +271,71 @@ impl RecordBatchStream for PerPartitionFilterStream {
     }
 }
 
+/// Build the `K = cuts.len() + 1` half-open range predicates a
+/// `PerPartitionFilterExec` needs to reproduce the range repartition's
+/// write-side routing on the read side.
+///
+/// Partition `i` receives the predicate
+///
+/// ```text
+///   i = 0        →  routing_expr < cuts[0]
+///   0 < i < K-1  →  cuts[i-1] <= routing_expr AND routing_expr < cuts[i]
+///   i = K-1      →  routing_expr >= cuts[K-2]
+///   K = 1        →  lit(true)   // empty cuts, single-bucket range repartition
+/// ```
+///
+/// Consistent with the private `range_repartition_common::split_batch_by_range`
+/// helper, which uses the same half-open convention on the write side.
+/// Callers pass the range repartition's routing expression verbatim
+/// (`CAST(order_by[0] AS Float64)` today).
+///
+/// Non-null routing expressions only. Both `UnorderedRangeRepartitionExec`
+/// and `OrderedRangeRepartitionExec` refuse nullable routing exprs at
+/// `try_new`, so any expression that reaches this helper via
+/// `RangeRepartitionRouting` is guaranteed non-null — no `IS NULL` branch
+/// needed.
+pub fn range_partition_predicates(
+    routing_expr: Arc<dyn PhysicalExpr>,
+    cuts: &[f64],
+) -> Vec<Arc<dyn PhysicalExpr>> {
+    use datafusion::logical_expr::Operator;
+    use datafusion::physical_expr::expressions::{BinaryExpr, Literal};
+    use datafusion::scalar::ScalarValue;
+
+    let k = cuts.len() + 1;
+    let lit = |v: f64| -> Arc<dyn PhysicalExpr> {
+        Arc::new(Literal::new(ScalarValue::Float64(Some(v))))
+    };
+    let ge = |lo: f64| -> Arc<dyn PhysicalExpr> {
+        Arc::new(BinaryExpr::new(
+            routing_expr.clone(),
+            Operator::GtEq,
+            lit(lo),
+        ))
+    };
+    let lt = |hi: f64| -> Arc<dyn PhysicalExpr> {
+        Arc::new(BinaryExpr::new(routing_expr.clone(), Operator::Lt, lit(hi)))
+    };
+    (0..k)
+        .map(|i| {
+            let lo = i.checked_sub(1).and_then(|j| cuts.get(j).copied());
+            let hi = cuts.get(i).copied();
+            match (lo, hi) {
+                (None, None) => {
+                    // K == 1: single bucket covers everything.
+                    Arc::new(Literal::new(ScalarValue::Boolean(Some(true))))
+                        as Arc<dyn PhysicalExpr>
+                }
+                (None, Some(hi)) => lt(hi),
+                (Some(lo), None) => ge(lo),
+                (Some(lo), Some(hi)) => {
+                    Arc::new(BinaryExpr::new(ge(lo), Operator::And, lt(hi)))
+                }
+            }
+        })
+        .collect()
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -400,6 +466,83 @@ mod tests {
         );
     }
 
+    /// The K=4 range predicates cover every value under the half-open
+    /// convention, and each row lands in exactly one predicate. Random
+    /// probe values are routed through the predicates and expected to
+    /// match the same partition assignment as the range repartition's
+    /// write-side `split_batch_by_range` would produce.
+    #[test]
+    fn range_partition_predicates_partition_every_value_exactly_once() {
+        use datafusion::arrow::array::Float64Array;
+        use datafusion::arrow::datatypes::Field;
+        use datafusion::physical_expr::expressions::Column;
+
+        let cuts = vec![10.0, 20.0, 30.0];
+        let k = cuts.len() + 1;
+        let routing: Arc<dyn PhysicalExpr> = Arc::new(Column::new("v", 0));
+        let preds = range_partition_predicates(routing, &cuts);
+        assert_eq!(preds.len(), k);
+
+        let schema =
+            Arc::new(Schema::new(vec![Field::new("v", DataType::Float64, false)]));
+        let values: Vec<f64> = vec![
+            -5.0, 0.0, 9.999, 10.0, 15.0, 19.999, 20.0, 25.0, 30.0, 100.0,
+        ];
+        let arr = Float64Array::from_iter_values(values.iter().copied());
+        let batch = RecordBatch::try_new(schema.clone(), vec![Arc::new(arr)]).unwrap();
+
+        // For each row, find the unique partition whose predicate accepts it.
+        for (row, &v) in values.iter().enumerate() {
+            let mut hits = 0;
+            for pred in &preds {
+                let mask = pred
+                    .evaluate(&batch)
+                    .and_then(|v| v.into_array(batch.num_rows()))
+                    .unwrap();
+                let mask = as_boolean_array(&mask).unwrap();
+                if mask.value(row) {
+                    hits += 1;
+                }
+            }
+            assert_eq!(
+                hits, 1,
+                "value {v} matched {hits} predicates, expected exactly 1"
+            );
+        }
+
+        // Expected assignment mirrors split_batch_by_range: `partition_point`
+        // returns the count of cuts `<= key`, which is the partition index
+        // under the half-open convention.
+        let expected: Vec<usize> = values
+            .iter()
+            .map(|v| cuts.partition_point(|&c| c <= *v))
+            .collect();
+        for (row, want) in expected.iter().enumerate() {
+            let mask = preds[*want]
+                .evaluate(&batch)
+                .and_then(|v| v.into_array(batch.num_rows()))
+                .unwrap();
+            let mask = as_boolean_array(&mask).unwrap();
+            assert!(
+                mask.value(row),
+                "value {} should have landed in partition {}",
+                values[row],
+                want
+            );
+        }
+    }
+
+    /// Degenerate K=1 (empty cuts) yields a single lit(true) predicate.
+    #[test]
+    fn range_partition_predicates_single_bucket_when_cuts_empty() {
+        use datafusion::physical_expr::expressions::Column;
+
+        let routing: Arc<dyn PhysicalExpr> = Arc::new(Column::new("v", 0));
+        let preds = range_partition_predicates(routing, &[]);
+        assert_eq!(preds.len(), 1);
+        assert_eq!(preds[0].to_string(), "true");
+    }
+
     /// `with_new_children` swaps the input while preserving the predicate
     /// vector. Wrapping the original source in a `RepartitionExec` that
     /// keeps the partition count (RoundRobin(3)) gives a valid child; the
diff --git a/ballista/core/src/execution_plans/runtime_stats.rs b/ballista/core/src/execution_plans/runtime_stats.rs
index d654f93956..5683b76b74 100644
--- a/ballista/core/src/execution_plans/runtime_stats.rs
+++ b/ballista/core/src/execution_plans/runtime_stats.rs
@@ -791,8 +791,9 @@ fn merge_group(
 /// `producer_task_id` that emitted it. The scheduler stores these on
 /// `RunningStage.runtime_stats_reports` so downstream stages can address
 /// individual producer files as `(producer_task_id, partition_id)` pairs —
-/// the partition_id inside a report is producer-local (0..K URRE sub-parts),
-/// so the pair is what uniquely identifies a shuffle file across producers.
+/// the partition_id inside a report is producer-local (0..K range-repartition
+/// sub-parts), so the pair is what uniquely identifies a shuffle file across
+/// producers.
 #[derive(Debug, Clone)]
 pub struct TaskRuntimeStats {
     /// Producer task's task_id at the time it emitted the report. Matches
@@ -803,6 +804,218 @@ pub struct TaskRuntimeStats {
     pub report: crate::serde::protobuf::RuntimeStatsReport,
 }
 
+/// One producer file's coordinates: the pair that uniquely identifies a
+/// range-repartition-stage output file across all producers.
+/// `producer_task_id` matches the `file_id` field on emitted
+/// `ShuffleWritePartition` records; `sub_part_id` matches the `partition_id`
+/// inside each producer's `RuntimeStatsReport`. The pair together names a
+/// specific file on disk (`work_dir/job/stage/sub_part/task_id.data` in the
+/// passthrough writer's path convention).
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct SubPartLocation {
+    pub producer_task_id: usize,
+    pub sub_part_id: u32,
+}
+
+/// True iff `plan`'s subtree contains an `UnorderedRangeRepartitionExec` or
+/// `OrderedRangeRepartitionExec` — a signal that the stage's output files
+/// are range-K-space-routed, so downstream consumers need overlap-based
+/// routing rather than the passthrough-by-partition-id default.
+///
+/// Walks the tree top-down. Doesn't descend into leaves — the range
+/// repartition would sit above any DataSourceExec at the bottom of a
+/// Ballista stage.
+pub fn plan_contains_range_repartition(plan: &dyn ExecutionPlan) -> bool {
+    if plan
+        .downcast_ref::<super::UnorderedRangeRepartitionExec>()
+        .is_some()
+        || plan
+            .downcast_ref::<super::OrderedRangeRepartitionExec>()
+            .is_some()
+    {
+        return true;
+    }
+    plan.children()
+        .iter()
+        .any(|c| plan_contains_range_repartition(c.as_ref()))
+}
+
+/// Walk `plan` for the first `UnorderedRangeRepartitionExec` or
+/// `OrderedRangeRepartitionExec` and return its routing expression
+/// (`order_by[0].expr`). Consumers that need the routing expression to
+/// build downstream range filters use this instead of
+/// [`plan_contains_range_repartition`], which is a boolean-only sibling.
+pub fn find_range_repartition_routing_expr(
+    plan: &dyn ExecutionPlan,
+) -> Option<Arc<dyn datafusion::physical_expr::PhysicalExpr>> {
+    if let Some(rre) = plan.downcast_ref::<super::UnorderedRangeRepartitionExec>() {
+        return Some(rre.order_by()[0].expr.clone());
+    }
+    if let Some(rre) = plan.downcast_ref::<super::OrderedRangeRepartitionExec>() {
+        return Some(rre.order_by()[0].expr.clone());
+    }
+    for c in plan.children() {
+        if let Some(expr) = find_range_repartition_routing_expr(c.as_ref()) {
+            return Some(expr);
+        }
+    }
+    None
+}
+
+/// Rebuild a stage's `Vec<Vec<PartitionLocation>>` under range-repartition
+/// overlap semantics. Consumes `default` (the passthrough-by-`partition_id`
+/// mapping) so its per-file metadata (executor, stats, file_id,
+/// is_sort_shuffle) can be pulled onto the remapped output.
+///
+/// # Returns
+///
+/// A new `Vec<Vec<PartitionLocation>>` where the outer index is the
+/// downstream stage-N+1 input partition k, and the inner list is the
+/// specific stage-N producer files whose sketched range overlaps k's
+/// assigned global range. Content-wise identical to what
+/// [`compute_overlapping_locations`] returns, but each `SubPartLocation`
+/// is expanded into a full `PartitionLocation` (executor, stats, etc.
+/// carried over from `default`).
+///
+/// # Errors
+///
+/// - Sketch decode failure inside `compute_overlapping_locations`.
+/// - `default` missing an entry the overlap step asks for
+///   (`(producer_task_id, sub_part_id)` combination not found), which
+///   would indicate a bookkeeping inconsistency between stage-output
+///   accumulation and sketch reporting.
+pub fn overlap_remap_partitions(
+    default: Vec<Vec<crate::serde::scheduler::PartitionLocation>>,
+    reports: &[TaskRuntimeStats],
+    global_cuts: &[f64],
+) -> Result<Vec<Vec<crate::serde::scheduler::PartitionLocation>>> {
+    use std::collections::HashMap;
+    let assignments = compute_overlapping_locations(reports, global_cuts)?;
+
+    // Index the default's flat entries by `(producer_task_id, sub_part_id)`.
+    // Producer task_id lives on `PartitionLocation.file_id`; sub_part_id
+    // lives on `PartitionLocation.partition_id.partition_id`.
+    let mut lookup: HashMap<(usize, u32), crate::serde::scheduler::PartitionLocation> =
+        HashMap::new();
+    for bucket in &default {
+        for loc in bucket {
+            let Some(file_id) = loc.file_id else {
+                return internal_err!(
+                    "range-repartition overlap remap: PartitionLocation missing \
+                     file_id (range-repartition stages emit passthrough files \
+                     with file_id=task_id) — partition_id={}",
+                    loc.partition_id.partition_id
+                );
+            };
+            let sub_part_id = loc.partition_id.partition_id as u32;
+            lookup.insert((file_id as usize, sub_part_id), loc.clone());
+        }
+    }
+
+    let mut remapped = Vec::with_capacity(assignments.len());
+    for bucket in assignments {
+        let mut inner = Vec::with_capacity(bucket.len());
+        for spl in bucket {
+            let Some(loc) = lookup.get(&(spl.producer_task_id, spl.sub_part_id)) else {
+                let key = format!(
+                    "(producer_task_id={}, sub_part_id={})",
+                    spl.producer_task_id, spl.sub_part_id
+                );
+                return internal_err!(
+                    "range-repartition overlap remap: no default PartitionLocation \
+                     for {} — stage-output vs. runtime-stats-report inconsistency",
+                    key
+                );
+            };
+            inner.push(loc.clone());
+        }
+        remapped.push(inner);
+    }
+    Ok(remapped)
+}
+
+/// Compute, for each stage-N+1 input partition, which stage-N producer
+/// files it needs to fetch — based on sketch-range overlap with the
+/// globally-assigned cut range.
+///
+/// # Arguments
+///
+/// - `reports` — per-task sketches from every stage-N producer task,
+///   keyed by producer `task_id` and (within each report) by
+///   range-repartition sub-part `partition_id`.
+/// - `global_cuts` — `K - 1` scheduler-computed cut points that divide
+///   the value range into K globally-consistent bins, where K = the
+///   stage-N+1 partition count.
+///
+/// # Returns
+///
+/// `Vec<Vec<SubPartLocation>>` where:
+/// - **Outer index** `k` = downstream stage-N+1 input partition, in
+///   `0..K` (`K = global_cuts.len() + 1`).
+/// - **Inner Vec** = the stage-N producer files (each identified by
+///   `SubPartLocation { producer_task_id, sub_part_id }`) whose sketched
+///   value range overlaps k's assigned global half-open range. Every
+///   file listed here is a source the downstream task-k reader must
+///   fetch; the injected `FilterExec` on the downstream side drops rows
+///   from over-inclusive sub-parts that fell outside k's range.
+///
+/// Downstream partition ranges follow the design-doc's half-open
+/// convention:
+/// - `k = 0`         → `(-∞, cuts[0])`
+/// - `0 < k < K - 1` → `[cuts[k-1], cuts[k])`
+/// - `k = K - 1`     → `[cuts[K-2], +∞)`
+///
+/// A sketch's range is `[sketch.min(), sketch.max()]` (both inclusive).
+/// `[min, max]` overlaps `[lower, upper)` iff `max >= lower AND min < upper`.
+///
+/// Reports whose sketch is `None` (row-count-only mode) or whose sketch
+/// has zero samples contribute no locations — nothing to route.
+///
+/// # Errors
+///
+/// Sketch decode errors abort the whole computation; the caller sees the
+/// error rather than silent misrouting. Wire corruption in a sketch
+/// means the stage's routing decisions are suspect.
+pub fn compute_overlapping_locations(
+    reports: &[TaskRuntimeStats],
+    global_cuts: &[f64],
+) -> Result<Vec<Vec<SubPartLocation>>> {
+    let k = global_cuts.len() + 1;
+    let mut out: Vec<Vec<SubPartLocation>> = vec![Vec::new(); k];
+    for stats in reports {
+        for entry in &stats.report.partitions {
+            let Some(sketch_proto) = entry.sketch.as_ref() else {
+                continue;
+            };
+            let sketch = sketch_from_proto(sketch_proto)?;
+            if sketch.count() == 0.0 {
+                continue;
+            }
+            let smin = sketch.min();
+            let smax = sketch.max();
+            for (partition_k, bucket) in out.iter_mut().enumerate() {
+                let lower = if partition_k == 0 {
+                    f64::NEG_INFINITY
+                } else {
+                    global_cuts[partition_k - 1]
+                };
+                let upper = if partition_k == k - 1 {
+                    f64::INFINITY
+                } else {
+                    global_cuts[partition_k]
+                };
+                if smax >= lower && smin < upper {
+                    bucket.push(SubPartLocation {
+                        producer_task_id: stats.producer_task_id,
+                        sub_part_id: entry.partition_id,
+                    });
+                }
+            }
+        }
+    }
+    Ok(out)
+}
+
 /// Merge `reports` and log each group's merged view at `debug!`
 /// (`RUST_LOG` promotes when needed). Any merge error is logged at
 /// `warn!` — the scheduler doesn't want telemetry loss to tank a query
@@ -1393,3 +1606,313 @@ mod merge_tests {
         assert!(merge_reports(&[]).unwrap().is_empty());
     }
 }
+
+#[cfg(test)]
+mod plan_walker_tests {
+    //! `plan_contains_range_repartition` and
+    //! `find_range_repartition_routing_expr` — recursive walks that recognize
+    //! both `UnorderedRangeRepartitionExec` and `OrderedRangeRepartitionExec`
+    //! at any depth.
+
+    use super::*;
+    use crate::execution_plans::{
+        OrderedRangeRepartitionExec, UnorderedRangeRepartitionExec,
+    };
+    use datafusion::arrow::compute::SortOptions;
+    use datafusion::arrow::datatypes::{DataType, Field, Schema};
+    use datafusion::datasource::memory::MemorySourceConfig;
+    use datafusion::datasource::source::DataSourceExec;
+    use datafusion::physical_expr::{LexOrdering, PhysicalSortExpr};
+    use datafusion::physical_plan::expressions::col;
+    use datafusion::physical_plan::sorts::sort::SortExec;
+
+    fn v_schema() -> Arc<Schema> {
+        Arc::new(Schema::new(vec![Field::new("v", DataType::Float64, false)]))
+    }
+
+    fn v_source() -> Arc<dyn ExecutionPlan> {
+        let schema = v_schema();
+        let memory =
+            Arc::new(MemorySourceConfig::try_new(&[vec![]], schema, None).unwrap());
+        Arc::new(DataSourceExec::new(memory))
+    }
+
+    fn sort_expr_v() -> PhysicalSortExpr {
+        let schema = v_schema();
+        PhysicalSortExpr {
+            expr: col("v", schema.as_ref()).unwrap(),
+            options: SortOptions {
+                descending: false,
+                nulls_first: false,
+            },
+        }
+    }
+
+    fn urre_over_source(k: usize) -> Arc<dyn ExecutionPlan> {
+        Arc::new(
+            UnorderedRangeRepartitionExec::try_new(v_source(), vec![sort_expr_v()], k)
+                .unwrap(),
+        )
+    }
+
+    fn orre_over_source(k: usize) -> Arc<dyn ExecutionPlan> {
+        // ORRE demands sorted input.
+        let sort = Arc::new(SortExec::new(
+            LexOrdering::new(vec![sort_expr_v()]).unwrap(),
+            v_source(),
+        ));
+        Arc::new(
+            OrderedRangeRepartitionExec::try_new(sort, vec![sort_expr_v()], k).unwrap(),
+        )
+    }
+
+    #[test]
+    fn plan_contains_range_repartition_bare_source_returns_false() {
+        assert!(!plan_contains_range_repartition(v_source().as_ref()));
+    }
+
+    #[test]
+    fn plan_contains_range_repartition_urre_at_root_returns_true() {
+        assert!(plan_contains_range_repartition(
+            urre_over_source(4).as_ref()
+        ));
+    }
+
+    #[test]
+    fn plan_contains_range_repartition_orre_at_root_returns_true() {
+        assert!(plan_contains_range_repartition(
+            orre_over_source(4).as_ref()
+        ));
+    }
+
+    #[test]
+    fn plan_contains_range_repartition_urre_nested_returns_true() {
+        // Canonical shape: RuntimeStatsExec above URRE, plus one more
+        // stats-wrapper layer to make sure the walk descends more than
+        // once.
+        let urre = urre_over_source(4);
+        let inner_stats =
+            Arc::new(RuntimeStatsExec::try_new(urre, Some(vec![sort_expr_v()])).unwrap());
+        let outer: Arc<dyn ExecutionPlan> = Arc::new(
+            RuntimeStatsExec::try_new(inner_stats, Some(vec![sort_expr_v()])).unwrap(),
+        );
+        assert!(plan_contains_range_repartition(outer.as_ref()));
+    }
+
+    #[test]
+    fn plan_contains_range_repartition_orre_nested_returns_true() {
+        let orre = orre_over_source(4);
+        let inner_stats =
+            Arc::new(RuntimeStatsExec::try_new(orre, Some(vec![sort_expr_v()])).unwrap());
+        let outer: Arc<dyn ExecutionPlan> = Arc::new(
+            RuntimeStatsExec::try_new(inner_stats, Some(vec![sort_expr_v()])).unwrap(),
+        );
+        assert!(plan_contains_range_repartition(outer.as_ref()));
+    }
+
+    #[test]
+    fn find_range_repartition_routing_expr_bare_source_returns_none() {
+        assert!(find_range_repartition_routing_expr(v_source().as_ref()).is_none());
+    }
+
+    #[test]
+    fn find_range_repartition_routing_expr_urre_returns_first_order_by_expr() {
+        let urre = urre_over_source(4);
+        let expr = find_range_repartition_routing_expr(urre.as_ref())
+            .expect("URRE must expose its routing expression");
+        // The expression is the `v` column — verify by string equality since
+        // PhysicalExpr doesn't implement PartialEq.
+        assert_eq!(format!("{expr}"), format!("{}", sort_expr_v().expr));
+    }
+
+    #[test]
+    fn find_range_repartition_routing_expr_orre_returns_first_order_by_expr() {
+        let orre = orre_over_source(4);
+        let expr = find_range_repartition_routing_expr(orre.as_ref())
+            .expect("ORRE must expose its routing expression");
+        assert_eq!(format!("{expr}"), format!("{}", sort_expr_v().expr));
+    }
+
+    #[test]
+    fn find_range_repartition_routing_expr_descends_through_stats_wrapper() {
+        // Canonical shape: RuntimeStatsExec above URRE — the walker must
+        // recurse through the stats wrapper to find the URRE beneath.
+        let urre = urre_over_source(4);
+        let stats: Arc<dyn ExecutionPlan> =
+            Arc::new(RuntimeStatsExec::try_new(urre, Some(vec![sort_expr_v()])).unwrap());
+        let expr = find_range_repartition_routing_expr(stats.as_ref())
+            .expect("walker must descend past RuntimeStatsExec");
+        assert_eq!(format!("{expr}"), format!("{}", sort_expr_v().expr));
+    }
+}
+
+#[cfg(test)]
+mod overlap_remap_tests {
+    //! `overlap_remap_partitions` — takes a passthrough
+    //! `Vec<Vec<PartitionLocation>>` and rewrites it under overlap semantics
+    //! computed from merged sketches. Straddling sub-parts appear in every
+    //! downstream partition whose range they touch; bookkeeping mismatches
+    //! surface as errors rather than silent misroutes.
+
+    use super::*;
+    use crate::serde::protobuf::{RuntimeStatsPartitionEntry, RuntimeStatsReport};
+    use crate::serde::scheduler::{
+        ExecutorMetadata, ExecutorOperatingSystemSpecification, ExecutorSpecification,
+        PartitionId, PartitionLocation, PartitionStats,
+    };
+
+    /// Build a `PartitionLocation` for a producer file identified by
+    /// `(producer_task_id, sub_part_id)`. `file_id = producer_task_id`
+    /// matches the passthrough writer's convention.
+    fn location(sub_part_id: usize, producer_task_id: usize) -> PartitionLocation {
+        PartitionLocation {
+            map_partition_id: 0,
+            partition_id: PartitionId {
+                job_id: "test-job".into(),
+                stage_id: 0,
+                partition_id: sub_part_id,
+            },
+            executor_meta: ExecutorMetadata {
+                id: format!("exec-{producer_task_id}"),
+                host: "".to_string(),
+                port: 0,
+                grpc_port: 0,
+                specification: ExecutorSpecification::default().with_vcores(0),
+                os_info: ExecutorOperatingSystemSpecification::default(),
+            },
+            partition_stats: PartitionStats::default(),
+            file_id: Some(producer_task_id as u64),
+            is_sort_shuffle: false,
+        }
+    }
+
+    /// Build a report whose sub-parts each carry a T-Digest sketch over
+    /// `values`. Slot `sub_part_id` gets a sketch of `values[sub_part_id]`.
+    fn sketch_report(
+        producer_task_id: usize,
+        values_per_sub_part: Vec<Vec<f64>>,
+    ) -> TaskRuntimeStats {
+        let partitions = values_per_sub_part
+            .into_iter()
+            .enumerate()
+            .map(|(sub_part_id, samples)| {
+                let sketch = if samples.is_empty() {
+                    None
+                } else {
+                    let digest = TDigest::new(100).merge_unsorted_f64(samples.clone());
+                    Some(sketch_to_proto(&digest).unwrap())
+                };
+                RuntimeStatsPartitionEntry {
+                    partition_id: sub_part_id as u32,
+                    row_count: samples.len() as u64,
+                    sketch,
+                }
+            })
+            .collect();
+        TaskRuntimeStats {
+            producer_task_id,
+            report: RuntimeStatsReport {
+                order_by: vec![],
+                partitions,
+            },
+        }
+    }
+
+    /// Two producers with disjoint value ranges + one downstream cut →
+    /// each downstream partition gets exactly one producer's files.
+    #[test]
+    fn overlap_remap_disjoint_producers_route_to_single_partition() {
+        // Producer 100 covers [0, 10); producer 200 covers [20, 30).
+        let reports = vec![
+            sketch_report(100, vec![vec![0.0, 5.0, 9.0]]),
+            sketch_report(200, vec![vec![20.0, 25.0, 29.0]]),
+        ];
+        // Cut at 15 → partition 0 = (-∞, 15), partition 1 = [15, +∞).
+        let cuts = vec![15.0];
+        // Default passthrough map: both producers wrote to sub_part_id=0.
+        let default = vec![vec![location(0, 100), location(0, 200)]];
+
+        let remapped = overlap_remap_partitions(default, &reports, &cuts).unwrap();
+        assert_eq!(remapped.len(), 2, "K = cuts.len() + 1");
+        // Partition 0: only producer 100.
+        assert_eq!(remapped[0].len(), 1);
+        assert_eq!(remapped[0][0].file_id, Some(100));
+        // Partition 1: only producer 200.
+        assert_eq!(remapped[1].len(), 1);
+        assert_eq!(remapped[1][0].file_id, Some(200));
+    }
+
+    /// A straddling sub-part — one whose sketched [min, max] spans the cut
+    /// — appears in BOTH downstream partitions' lists. This is the case
+    /// PerPartitionFilterExec exists to clean up.
+    #[test]
+    fn overlap_remap_straddling_producer_appears_in_both_partitions() {
+        // Producer 300 covers [5, 25) — straddles the cut at 15.
+        let reports = vec![sketch_report(300, vec![vec![5.0, 15.0, 25.0]])];
+        let cuts = vec![15.0];
+        let default = vec![vec![location(0, 300)]];
+
+        let remapped = overlap_remap_partitions(default, &reports, &cuts).unwrap();
+        assert_eq!(remapped.len(), 2);
+        assert_eq!(remapped[0].len(), 1, "straddler in partition 0");
+        assert_eq!(remapped[0][0].file_id, Some(300));
+        assert_eq!(remapped[1].len(), 1, "straddler in partition 1");
+        assert_eq!(remapped[1][0].file_id, Some(300));
+    }
+
+    /// A producer file without `file_id` means the writer that produced it
+    /// wasn't the passthrough writer — remap can't identify the file, so
+    /// error rather than silently misroute.
+    #[test]
+    fn overlap_remap_missing_file_id_errors() {
+        let reports = vec![sketch_report(100, vec![vec![1.0, 2.0, 3.0]])];
+        let cuts = vec![10.0];
+        // Loc has file_id=None — invalid for URRE/ORRE stages.
+        let mut bad = location(0, 100);
+        bad.file_id = None;
+        let default = vec![vec![bad]];
+
+        let err = overlap_remap_partitions(default, &reports, &cuts)
+            .expect_err("missing file_id must surface as an error");
+        assert!(
+            err.to_string().contains("missing file_id"),
+            "unexpected error: {err}"
+        );
+    }
+
+    /// The overlap step names a (producer_task_id, sub_part_id) pair that
+    /// isn't in the default passthrough map — bookkeeping mismatch between
+    /// stage-output accumulation and sketch reporting.
+    #[test]
+    fn overlap_remap_missing_default_location_errors() {
+        // Report from producer 100 sub_part_id=0.
+        let reports = vec![sketch_report(100, vec![vec![1.0, 2.0, 3.0]])];
+        let cuts = vec![10.0];
+        // But default only has producer 200 — no match.
+        let default = vec![vec![location(0, 200)]];
+
+        let err = overlap_remap_partitions(default, &reports, &cuts)
+            .expect_err("missing default location must surface as an error");
+        let msg = err.to_string();
+        assert!(
+            msg.contains("no default PartitionLocation")
+                && msg.contains("producer_task_id=100"),
+            "unexpected error: {msg}"
+        );
+    }
+
+    /// Empty-sketch entries contribute no locations — every downstream
+    /// partition ends up empty. Not an error; the caller sees this as
+    /// "no data flowed through this producer."
+    #[test]
+    fn overlap_remap_empty_sketches_produce_empty_partitions() {
+        let reports = vec![sketch_report(100, vec![vec![]])];
+        let cuts = vec![10.0];
+        let default = vec![vec![location(0, 100)]];
+
+        let remapped = overlap_remap_partitions(default, &reports, &cuts).unwrap();
+        assert_eq!(remapped.len(), 2);
+        assert!(remapped[0].is_empty());
+        assert!(remapped[1].is_empty());
+    }
+}
diff --git a/ballista/core/src/execution_plans/shuffle_writer.rs b/ballista/core/src/execution_plans/shuffle_writer.rs
index 0742dbf213..847df4756f 100644
--- a/ballista/core/src/execution_plans/shuffle_writer.rs
+++ b/ballista/core/src/execution_plans/shuffle_writer.rs
@@ -73,9 +73,15 @@ enum GlobalPartitionMap {
     /// The plan collapses to a single output partition (e.g.
     /// `SortPreservingMergeExec`). Every local index → global partition 0.
     Collapsed,
-    /// The plan re-establishes a hash-partition K-space (e.g.
-    /// `RepartitionExec::Hash(_, K)`). Local index == global (0..K).
-    HashSpace,
+    /// The plan re-establishes a fresh K-space of output partitions — every
+    /// operator that fans rows into K distinct outputs and assigns each output
+    /// a fresh index. `RepartitionExec::Hash(_, K)` and
+    /// `RepartitionExec::RoundRobinBatch(K)` qualify;
+    /// `UnorderedRangeRepartitionExec` (range-routed) qualifies too. Local
+    /// index == global (0..K) regardless of the routing algorithm — the K-space
+    /// is per-stage and per-task-slot, `file_id` disambiguates the same
+    /// `partition_id` across producers.
+    KSpace,
     /// Nothing between the writer and the leaves rewrites partitioning —
     /// local index `i` is `slice[i]` globally. Empty when no slice was
     /// stamped, in which case local is used as-is (identity).
@@ -86,7 +92,7 @@ impl GlobalPartitionMap {
     fn resolve(&self, local: usize) -> u64 {
         match self {
             GlobalPartitionMap::Collapsed => 0,
-            GlobalPartitionMap::HashSpace => local as u64,
+            GlobalPartitionMap::KSpace => local as u64,
             GlobalPartitionMap::PassThrough(slice) => {
                 slice.get(local).copied().unwrap_or(local) as u64
             }
@@ -98,7 +104,14 @@ impl GlobalPartitionMap {
 /// determines the output partitioning shape:
 ///
 /// - `SortPreservingMergeExec` → `Collapsed` (fan-in to 1).
-/// - `RepartitionExec` producing a hash / round-robin K-space → `HashSpace`.
+/// - `RepartitionExec(Hash|RoundRobin)` → `KSpace` (fresh K-space by hash /
+///   round-robin routing).
+/// - `UnorderedRangeRepartitionExec` / `OrderedRangeRepartitionExec` →
+///   `KSpace` (fresh K-space by range routing, unordered or order-preserving).
+///   Same shape as hash-repartition from the writer's perspective: K distinct
+///   outputs, local index == global partition, `file_id` disambiguates across
+///   producers. Content-range info (which K-slot holds which value range)
+///   travels through the separate `RuntimeStatsExec` sketch-report channel.
 /// - Otherwise recurse into the sole child (Filter/Sort/Projection/… are
 ///   partitioning-preserving passthroughs).
 /// - If we hit a leaf or a fan-in without recognising it, treat it as
@@ -113,7 +126,7 @@ fn walk_child_partition_mapping(
     if let Some(repart) = plan.downcast_ref::<RepartitionExec>() {
         match repart.partitioning() {
             Partitioning::Hash(_, _) | Partitioning::RoundRobinBatch(_) => {
-                return GlobalPartitionMap::HashSpace;
+                return GlobalPartitionMap::KSpace;
             }
             Partitioning::UnknownPartitioning(_) => {
                 // RepartitionExec still exchanges rows and freshly numbers
@@ -128,6 +141,15 @@ fn walk_child_partition_mapping(
             }
         }
     }
+    if plan
+        .downcast_ref::<crate::execution_plans::UnorderedRangeRepartitionExec>()
+        .is_some()
+        || plan
+            .downcast_ref::<crate::execution_plans::OrderedRangeRepartitionExec>()
+            .is_some()
+    {
+        return GlobalPartitionMap::KSpace;
+    }
     let children = plan.children();
     if children.len() == 1 {
         return walk_child_partition_mapping(children[0], global_output_partition_ids);
@@ -144,7 +166,7 @@ fn walk_child_partition_mapping(
 ///
 /// Two cases:
 ///
-/// - `SortShuffleWriter(Hash(K))` — HashSpace by construction; the K-space
+/// - `SortShuffleWriter(Hash(K))` — KSpace by construction; the K-space
 ///   `[0..K-1]` is intrinsic. Input ids are irrelevant.
 /// - `ShuffleWriter` — always passthrough; the child's plan shape decides:
 ///   - `SortPreservingMergeExec` in the child chain → `[0]` (collapse).
@@ -171,7 +193,7 @@ pub fn compute_global_output_partition_ids(
         };
         return match walk_child_partition_mapping(child, global_input_partition_ids) {
             GlobalPartitionMap::Collapsed => vec![0],
-            GlobalPartitionMap::HashSpace => {
+            GlobalPartitionMap::KSpace => {
                 let k = child.properties().output_partitioning().partition_count();
                 (0..k).collect()
             }
@@ -494,7 +516,7 @@ impl ShuffleWriterExec {
     /// operator's output_partitioning). `summary.partition_id` is the
     /// **global** output partition id downstream will address, computed via
     /// `walk_child_partition_mapping` over the child plan — either
-    /// `global_output_partition_ids[local]`, `local` (hash K-space), or `0`
+    /// `global_output_partition_ids[local]`, `local` (K-space), or `0`
     /// (collapsed / SPM).
     pub fn execute_shuffle_write(
         self,
diff --git a/ballista/scheduler/src/state/aqe/adapter.rs b/ballista/scheduler/src/state/aqe/adapter.rs
index 2040982341..964d7bcc09 100644
--- a/ballista/scheduler/src/state/aqe/adapter.rs
+++ b/ballista/scheduler/src/state/aqe/adapter.rs
@@ -19,7 +19,9 @@ use crate::planner::create_shuffle_writer_with_config;
 use crate::state::aqe::execution_plan::{AdaptiveDatafusionExec, ExchangeExec};
 use crate::state::aqe::planner::AdaptiveStageInfo;
 use ballista_core::JobId;
-use ballista_core::execution_plans::ShuffleReaderExec;
+use ballista_core::execution_plans::{
+    PerPartitionFilterExec, ShuffleReaderExec, range_partition_predicates,
+};
 use datafusion::common::exec_err;
 use datafusion::config::ConfigOptions;
 use datafusion::error::DataFusionError;
@@ -28,6 +30,7 @@ use datafusion::{
     common::tree_node::{Transformed, TreeNode},
     physical_plan::ExecutionPlan,
 };
+use log::info;
 use std::sync::Arc;
 
 #[derive(Debug, Clone, Default)]
@@ -105,7 +108,33 @@ impl BallistaAdapter {
                 )?,
             };
 
-            Ok(Transformed::yes(Arc::new(reader)))
+            let reader: Arc<dyn ExecutionPlan> = Arc::new(reader);
+            // If the upstream stage was range-repartition-routed and cuts
+            // were recovered at final-success, wrap the reader in a
+            // `PerPartitionFilterExec` so downstream partition k receives
+            // only rows whose routing value falls in the k-th cut range.
+            // Straddling sub-parts of the producer would otherwise feed
+            // multiple partitions and `FinalPartitioned` would split partial
+            // sums.
+            //
+            // Predicates are indexed on the global K-space here; the per-task
+            // specialization in `restrict_plan_to_partitions` slices them in
+            // lockstep with the reader's `partition` field so each task sees
+            // predicates and partitions matched positionally.
+            if let Some(routing) = exchange.range_repartition_routing() {
+                let predicates =
+                    range_partition_predicates(routing.routing_expr, &routing.cuts);
+                info!(
+                    "range-repartition: injecting PerPartitionFilterExec above \
+                     ShuffleReader for stage {} — {} predicates over {} cuts",
+                    stage_id,
+                    predicates.len(),
+                    routing.cuts.len(),
+                );
+                let filtered = PerPartitionFilterExec::try_new(reader, predicates)?;
+                return Ok(Transformed::yes(Arc::new(filtered)));
+            }
+            Ok(Transformed::yes(reader))
         } else {
             Ok(Transformed::no(plan))
         }
diff --git a/ballista/scheduler/src/state/aqe/execution_plan/exchange.rs b/ballista/scheduler/src/state/aqe/execution_plan/exchange.rs
index 63cc181502..ee224fea1d 100644
--- a/ballista/scheduler/src/state/aqe/execution_plan/exchange.rs
+++ b/ballista/scheduler/src/state/aqe/execution_plan/exchange.rs
@@ -19,6 +19,7 @@ use ballista_core::execution_plans::{
     CoalescePlan, stats_for_partition, stats_for_partitions,
 };
 use ballista_core::serde::scheduler::PartitionLocation;
+use datafusion::physical_expr::PhysicalExpr;
 use datafusion::physical_plan::Statistics;
 use datafusion::{
     error::{DataFusionError, Result},
@@ -32,6 +33,26 @@ use parking_lot::Mutex;
 use std::ops::Deref;
 use std::sync::{Arc, atomic::AtomicI64};
 
+/// Range-partition boundaries recovered from an
+/// `UnorderedRangeRepartitionExec` / `OrderedRangeRepartitionExec` upstream
+/// of this exchange. Written after the range-repartition-producing stage
+/// completes and its runtime-stats sketches are merged; read at
+/// task-specialization time to build per-downstream-partition range filters
+/// (see `PerPartitionFilterExec`).
+///
+/// `cuts` are `K - 1` monotone `f64` boundaries expressed in the value space
+/// of `routing_expr`; downstream partition `k` owns `[cuts[k-1], cuts[k])`
+/// with virtual `-∞`/`+∞` sentinels on the ends (matching the range
+/// repartition's write-side convention). `routing_expr` is the same
+/// expression the range repartition routes on — a `CAST(order_by[0] AS
+/// Float64)` today — so the filter is symmetric with the writer's placement
+/// decision.
+#[derive(Clone, Debug)]
+pub struct RangeRepartitionRouting {
+    pub cuts: Vec<f64>,
+    pub routing_expr: Arc<dyn PhysicalExpr>,
+}
+
 /// Execution plan representing an exchange/shuffle boundary used by the
 /// scheduler during adaptive query execution (AQE).
 ///
@@ -74,6 +95,19 @@ pub struct ExchangeExec {
     /// transform-rebuilt parent chains. Same pattern as `shuffle_partitions`.
     coalesce: Arc<Mutex<Option<Arc<CoalescePlan>>>>,
 
+    /// Range-partition boundaries recovered at runtime from an upstream
+    /// range-repartition op (URRE or ORRE). Written by
+    /// `AdaptiveExecutionGraph::maybe_range_repartition_overlap_remap` when
+    /// the range-repartition-producing stage completes and its per-sub-part
+    /// quantile sketches have been merged. Read at task-specialization time
+    /// to build `PerPartitionFilterExec` predicates for downstream stage
+    /// `N+1`.
+    ///
+    /// `None` on any exchange that isn't downstream of a range repartition
+    /// (the common case — hash/range/round-robin exchanges never populate
+    /// this).
+    range_repartition_routing: Arc<Mutex<Option<RangeRepartitionRouting>>>,
+
     /// this disables stage from running even it would be suitable to run.
     ///
     /// the main reason for this property this is to allow rules to override
@@ -103,6 +137,7 @@ impl ExchangeExec {
             Arc::new(AtomicI64::new(-1)),
             Arc::new(Mutex::new(None)),
             Arc::new(Mutex::new(None)),
+            Arc::new(Mutex::new(None)),
             false,
             false,
         )
@@ -120,6 +155,7 @@ impl ExchangeExec {
             Arc::new(AtomicI64::new(-1)),
             Arc::new(Mutex::new(None)),
             Arc::new(Mutex::new(None)),
+            Arc::new(Mutex::new(None)),
             true,
             false,
         )
@@ -133,6 +169,7 @@ impl ExchangeExec {
             self.stage_id.clone(),
             self.shuffle_partitions.clone(),
             self.coalesce.clone(),
+            self.range_repartition_routing.clone(),
             true,
             self.inactive_stage,
         )
@@ -149,6 +186,7 @@ impl ExchangeExec {
         stage_id: Arc<AtomicI64>,
         stage_partitions: Arc<Mutex<Option<Vec<Vec<PartitionLocation>>>>>,
         coalesce: Arc<Mutex<Option<Arc<CoalescePlan>>>>,
+        range_repartition_routing: Arc<Mutex<Option<RangeRepartitionRouting>>>,
         broadcast: bool,
         inactive_stage: bool,
     ) -> Self {
@@ -173,6 +211,7 @@ impl ExchangeExec {
             shuffle_partitions: stage_partitions,
             partitioning,
             coalesce,
+            range_repartition_routing,
             inactive_stage,
             broadcast,
         }
@@ -257,6 +296,24 @@ impl ExchangeExec {
     pub fn coalesce(&self) -> Option<Arc<CoalescePlan>> {
         self.coalesce.lock().clone()
     }
+
+    /// Publishes range-repartition-recovered range boundaries on this
+    /// exchange. Called from
+    /// `AdaptiveExecutionGraph::maybe_range_repartition_overlap_remap` once
+    /// the upstream range-repartition stage completes and its per-sub-part
+    /// quantile sketches have been merged into `K - 1` monotone cuts.
+    /// Idempotent overwrite matches the `set_coalesce` pattern.
+    pub fn resolve_range_repartition_routing(&self, routing: RangeRepartitionRouting) {
+        self.range_repartition_routing.lock().replace(routing);
+    }
+
+    /// Returns the range-repartition routing info if
+    /// `resolve_range_repartition_routing` has fired. Consumers use
+    /// `Some(_)` as the signal that this exchange is downstream of a range
+    /// repartition and its tasks need per-partition range filters.
+    pub fn range_repartition_routing(&self) -> Option<RangeRepartitionRouting> {
+        self.range_repartition_routing.lock().clone()
+    }
 }
 
 impl DisplayAs for ExchangeExec {
@@ -288,6 +345,9 @@ impl DisplayAs for ExchangeExec {
                         cp.upstream_partition_count,
                     )?;
                 }
+                if let Some(r) = self.range_repartition_routing.lock().as_ref() {
+                    write!(f, ", range_repartition_cuts={}", r.cuts.len())?;
+                }
                 if self.broadcast {
                     write!(f, ", broadcast=true",)?
                 }
@@ -354,6 +414,10 @@ impl ExecutionPlan for ExchangeExec {
                 // doesn't lose the rule's decision.
                 self.shuffle_partitions.clone(),
                 self.coalesce.clone(),
+                // Same rationale for range-repartition routing: transform
+                // passes that rebuild the parent must not drop range
+                // boundaries the scheduler already parked on this exchange.
+                self.range_repartition_routing.clone(),
                 self.broadcast,
                 self.inactive_stage,
             );
@@ -421,3 +485,94 @@ impl ExecutionPlan for ExchangeExec {
         }
     }
 }
+
+#[cfg(test)]
+mod range_repartition_routing_tests {
+    //! `RangeRepartitionRouting` parking on `ExchangeExec`. The AQE hook
+    //! writes here at range-repartition-stage completion; task
+    //! specialization reads it back at
+    //! `BallistaAdapter::transform_children` time to wrap the
+    //! ShuffleReader in a `PerPartitionFilterExec`. Neither side is
+    //! exercised end-to-end without the URRE-inserting rule (a follow-up
+    //! PR), so tests here cover the slot itself: roundtrip through
+    //! `resolve_range_repartition_routing` → `range_repartition_routing()`,
+    //! and preservation across `with_new_children`.
+    use super::*;
+    use datafusion::arrow::datatypes::{DataType, Field, Schema};
+    use datafusion::datasource::memory::MemorySourceConfig;
+    use datafusion::datasource::source::DataSourceExec;
+    use datafusion::physical_plan::expressions::col;
+    use datafusion::physical_plan::{ExecutionPlan, Partitioning};
+
+    fn v_source() -> Arc<dyn ExecutionPlan> {
+        let schema =
+            Arc::new(Schema::new(vec![Field::new("v", DataType::Float64, false)]));
+        let memory =
+            Arc::new(MemorySourceConfig::try_new(&[vec![]], schema, None).unwrap());
+        Arc::new(DataSourceExec::new(memory))
+    }
+
+    fn v_routing_expr() -> Arc<dyn PhysicalExpr> {
+        let schema = v_source().schema();
+        col("v", schema.as_ref()).unwrap()
+    }
+
+    fn sample_routing() -> RangeRepartitionRouting {
+        RangeRepartitionRouting {
+            cuts: vec![10.0, 20.0, 30.0],
+            routing_expr: v_routing_expr(),
+        }
+    }
+
+    #[test]
+    fn range_repartition_routing_unresolved_returns_none() {
+        let exchange = ExchangeExec::new(v_source(), None, 42);
+        assert!(exchange.range_repartition_routing().is_none());
+    }
+
+    #[test]
+    fn resolve_range_repartition_routing_roundtrips() {
+        let exchange = ExchangeExec::new(v_source(), None, 42);
+        exchange.resolve_range_repartition_routing(sample_routing());
+        let recovered = exchange
+            .range_repartition_routing()
+            .expect("routing must be Some after resolve");
+        assert_eq!(recovered.cuts, vec![10.0, 20.0, 30.0]);
+    }
+
+    #[test]
+    fn resolve_range_repartition_routing_overwrites_prior_value() {
+        // Idempotent-overwrite semantics match `set_coalesce`.
+        let exchange = ExchangeExec::new(v_source(), None, 42);
+        exchange.resolve_range_repartition_routing(sample_routing());
+        exchange.resolve_range_repartition_routing(RangeRepartitionRouting {
+            cuts: vec![100.0],
+            routing_expr: v_routing_expr(),
+        });
+        let recovered = exchange.range_repartition_routing().unwrap();
+        assert_eq!(recovered.cuts, vec![100.0], "second resolve wins");
+    }
+
+    /// `with_new_children` must carry the routing slot through: transform
+    /// passes that rebuild the parent chain would otherwise silently drop
+    /// range boundaries the scheduler already parked here.
+    #[test]
+    fn with_new_children_preserves_range_repartition_routing() {
+        let partitioning = Some(Partitioning::UnknownPartitioning(4));
+        let exchange = Arc::new(ExchangeExec::new(v_source(), partitioning, 42));
+        exchange.resolve_range_repartition_routing(sample_routing());
+
+        // Rebuild with a fresh (equivalent-schema) child.
+        let rebuilt = exchange
+            .clone()
+            .with_new_children(vec![v_source()])
+            .unwrap();
+        let rebuilt_exchange = rebuilt
+            .downcast_ref::<ExchangeExec>()
+            .expect("with_new_children must return an ExchangeExec");
+        let recovered = rebuilt_exchange
+            .range_repartition_routing()
+            .expect("routing must survive with_new_children");
+        assert_eq!(recovered.cuts, vec![10.0, 20.0, 30.0]);
+    }
+}
diff --git a/ballista/scheduler/src/state/aqe/mod.rs b/ballista/scheduler/src/state/aqe/mod.rs
index 1c92580e93..e9b1bfe487 100644
--- a/ballista/scheduler/src/state/aqe/mod.rs
+++ b/ballista/scheduler/src/state/aqe/mod.rs
@@ -18,6 +18,7 @@
 use crate::display::print_stage_metrics;
 use crate::scheduler_server::event::QueryStageSchedulerEvent;
 use crate::scheduler_server::timestamp_millis;
+use crate::state::aqe::execution_plan::RangeRepartitionRouting;
 use crate::state::aqe::planner::AdaptivePlanner;
 use crate::state::execution_graph::{
     ExecutionGraph, ExecutionGraphBox, ExecutionStage, ResolvedStage, RunningTaskInfo,
@@ -283,6 +284,104 @@ impl AdaptiveExecutionGraph {
         Ok(events)
     }
 
+    /// If the completed stage's plan contains a range repartition (URRE or
+    /// ORRE), remap the default `partition_id`-passthrough location list to
+    /// the overlap-based mapping computed from merged sketches, and return
+    /// the recovered `RangeRepartitionRouting` (cuts + routing expression)
+    /// so the caller can park it on the stage's downstream exchange for use
+    /// at task-specialization time.
+    ///
+    /// Returns `(default, None)` when no range repartition is present or
+    /// when there are no non-empty sketches to merge — the ordinary
+    /// passthrough behavior.
+    fn maybe_range_repartition_overlap_remap(
+        &self,
+        stage_id: usize,
+        default: Vec<Vec<PartitionLocation>>,
+    ) -> ballista_core::error::Result<(
+        Vec<Vec<PartitionLocation>>,
+        Option<RangeRepartitionRouting>,
+    )> {
+        let Some(ExecutionStage::Running(running_stage)) = self.stages.get(&stage_id)
+        else {
+            return Ok((default, None));
+        };
+        if !ballista_core::execution_plans::plan_contains_range_repartition(
+            running_stage.plan.as_ref(),
+        ) {
+            return Ok((default, None));
+        }
+        if running_stage.runtime_stats_reports.is_empty() {
+            info!(
+                "range-repartition stage {} completed with no runtime-stats \
+                 reports; falling back to passthrough partition mapping",
+                stage_id,
+            );
+            return Ok((default, None));
+        }
+        // Merge sketches to get global cuts, then remap.
+        let raw: Vec<_> = running_stage
+            .runtime_stats_reports
+            .iter()
+            .map(|t| t.report.clone())
+            .collect();
+        let merged = ballista_core::execution_plans::merge_runtime_stats_reports(&raw)
+            .map_err(|e| {
+                BallistaError::General(format!(
+                    "range-repartition overlap remap: merge_reports failed for \
+                     stage {stage_id}: {e}"
+                ))
+            })?;
+        // A range-repartition stage produces one report group (its single
+        // routing expression). Pick the first non-empty cuts and remap; if
+        // there are no cuts (all sketches empty), pass the default through.
+        let Some(cuts) = merged.iter().map(|m| &m.cuts).find(|c| !c.is_empty()) else {
+            info!(
+                "range-repartition stage {} produced no non-empty sketches; \
+                 falling back to passthrough partition mapping",
+                stage_id,
+            );
+            return Ok((default, None));
+        };
+        let cuts = cuts.clone();
+        let remapped = ballista_core::execution_plans::overlap_remap_partitions(
+            default,
+            &running_stage.runtime_stats_reports,
+            &cuts,
+        )
+        .map_err(|e| {
+            BallistaError::General(format!(
+                "range-repartition overlap remap failed for stage {stage_id}: {e}"
+            ))
+        })?;
+        info!(
+            "range-repartition stage {} partition mapping remapped by \
+             sketch-overlap: {} downstream partitions, total {} producer files",
+            stage_id,
+            remapped.len(),
+            remapped.iter().map(|v| v.len()).sum::<usize>(),
+        );
+        // The routing expression on the range repartition is the same one
+        // downstream filters need to reference. Missing it here is a
+        // plan-shape bug (range repartition detected by
+        // `plan_contains_range_repartition` but the walker can't find it) —
+        // surface as the ordinary passthrough case so the query still
+        // completes, at the cost of straddling correctness.
+        let routing =
+            ballista_core::execution_plans::find_range_repartition_routing_expr(
+                running_stage.plan.as_ref(),
+            )
+            .map(|routing_expr| RangeRepartitionRouting { cuts, routing_expr });
+        if routing.is_none() {
+            info!(
+                "range-repartition stage {} passed \
+                 plan_contains_range_repartition but no routing expression was \
+                 located; downstream filters will not be injected",
+                stage_id,
+            );
+        }
+        Ok((remapped, routing))
+    }
     /// Return a Vec of stages to cancel
     fn update_stage_progress(
         &mut self,
@@ -294,7 +393,25 @@ impl AdaptiveExecutionGraph {
             .update_exchange_locations(stage_id, locations)?;
 
         if is_completed {
-            let locations = self.planner.finalise_stage(stage_id)?;
+            // Range-repartition-terminated stages need overlap-based remap of
+            // the per-downstream-partition location list. For every other
+            // stage, pass the accumulated stage output through unchanged (the
+            // default `partition_id`-passthrough mapping).
+            let default = self.planner.take_stage_output_partitions(stage_id)?;
+            let (locations, range_repartition_routing) =
+                self.maybe_range_repartition_overlap_remap(stage_id, default)?;
+            // Park recovered range boundaries on the stage's boundary
+            // ExchangeExec BEFORE `resolve_stage_partitions` runs, because
+            // that call removes the stage from `runnable_stage_cache` —
+            // after which the exchange is no longer reachable by stage_id.
+            // Downstream stage-N+1 task specialization needs the routing to
+            // wrap the ShuffleReaderExec in a `PerPartitionFilterExec`.
+            if let Some(routing) = range_repartition_routing {
+                self.planner
+                    .set_range_repartition_routing(stage_id, routing)?;
+            }
+            self.planner
+                .resolve_stage_partitions(stage_id, locations.clone())?;
 
             let (runnable, stages_to_cancel) = self.planner.actionable_stages()?;
 
diff --git a/ballista/scheduler/src/state/aqe/optimizer_rule/distributed_exchange.rs b/ballista/scheduler/src/state/aqe/optimizer_rule/distributed_exchange.rs
index a07312d14e..5d961610ed 100644
--- a/ballista/scheduler/src/state/aqe/optimizer_rule/distributed_exchange.rs
+++ b/ballista/scheduler/src/state/aqe/optimizer_rule/distributed_exchange.rs
@@ -16,6 +16,9 @@
 // under the License.
 
 use crate::state::aqe::execution_plan::{AdaptiveDatafusionExec, ExchangeExec};
+use ballista_core::execution_plans::{
+    OrderedRangeRepartitionExec, RuntimeStatsExec, UnorderedRangeRepartitionExec,
+};
 use datafusion::common::tree_node::{Transformed, TreeNode};
 use datafusion::physical_optimizer::PhysicalOptimizerRule;
 use datafusion::physical_plan::coalesce_partitions::CoalescePartitionsExec;
@@ -90,11 +93,68 @@ impl DistributedExchangeRule {
                 );
                 return Ok(Transformed::yes(Arc::new(exchange_exec)));
             }
+        } else if execution_plan.downcast_ref::<ExchangeExec>().is_none() {
+            // Fourth classifier arm: any node whose immediate child is a
+            // range-repartition chain top (a `RuntimeStatsExec` whose input
+            // is `UnorderedRangeRepartitionExec` or
+            // `OrderedRangeRepartitionExec`) needs a passthrough
+            // `ExchangeExec(None)` inserted between them, so the range
+            // repartition and its `stats_report` end up on the upstream side
+            // of the stage boundary — that's the stage the executor's report
+            // walker runs on, so the sketches ship to the scheduler at
+            // stage-N completion.
+            //
+            // Idempotent: after insertion the child slot holds an
+            // `ExchangeExec`, which is not a range-repartition chain top;
+            // the guard on this arm
+            // (`execution_plan.downcast_ref::<ExchangeExec>().is_none()`)
+            // stops us from wrapping again on subsequent AQE replans, since
+            // the visited node would be the wrapping `ExchangeExec` itself.
+            let children = execution_plan.children();
+            if children.iter().any(|c| is_range_repartition_chain_top(c)) {
+                let new_children: Vec<Arc<dyn ExecutionPlan>> = children
+                    .into_iter()
+                    .map(|c| {
+                        if is_range_repartition_chain_top(c) {
+                            let id = self
+                                .plan_id_generator
+                                .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+                            Arc::new(ExchangeExec::new((*c).clone(), None, id))
+                                as Arc<dyn ExecutionPlan>
+                        } else {
+                            (*c).clone()
+                        }
+                    })
+                    .collect();
+                return Ok(Transformed::yes(
+                    execution_plan.with_new_children(new_children)?,
+                ));
+            }
         }
         Ok(Transformed::no(execution_plan))
     }
 }
 
+/// True iff `plan` is a `RuntimeStatsExec` sitting directly above an
+/// `UnorderedRangeRepartitionExec` or `OrderedRangeRepartitionExec` — the
+/// canonical "stats_report" shape a range-repartition-inserting rule emits
+/// at the top of its splice.
+fn is_range_repartition_chain_top(plan: &Arc<dyn ExecutionPlan>) -> bool {
+    if plan.downcast_ref::<RuntimeStatsExec>().is_none() {
+        return false;
+    }
+    let children = plan.children();
+    let [child] = children.as_slice() else {
+        return false;
+    };
+    child
+        .downcast_ref::<UnorderedRangeRepartitionExec>()
+        .is_some()
+        || child
+            .downcast_ref::<OrderedRangeRepartitionExec>()
+            .is_some()
+}
+
 impl PhysicalOptimizerRule for DistributedExchangeRule {
     fn optimize(
         &self,
diff --git a/ballista/scheduler/src/state/aqe/planner.rs b/ballista/scheduler/src/state/aqe/planner.rs
index 67d303761a..84acd925a4 100644
--- a/ballista/scheduler/src/state/aqe/planner.rs
+++ b/ballista/scheduler/src/state/aqe/planner.rs
@@ -16,7 +16,9 @@
 // under the License.
 use crate::physical_optimizer::filter_pushdown::FilterPushdown;
 use crate::state::aqe::adapter::BallistaAdapter;
-use crate::state::aqe::execution_plan::{AdaptiveDatafusionExec, ExchangeExec};
+use crate::state::aqe::execution_plan::{
+    AdaptiveDatafusionExec, ExchangeExec, RangeRepartitionRouting,
+};
 use crate::state::aqe::optimizer_rule::chaos_exec::ChaosCreatingRule;
 use crate::state::aqe::optimizer_rule::{
     CoalescePartitionsRule, DelayJoinSelectionRule, DistributedExchangeRule,
@@ -207,6 +209,30 @@ impl AdaptivePlanner {
     ///
     /// # Returns
     /// A `Result` indicating success or failure.
+    /// Attaches range-repartition-recovered range boundaries to the
+    /// boundary `ExchangeExec` for `stage_id`. Called right after the
+    /// completed range-repartition stage's partition mapping is resolved;
+    /// the routing carries the cuts + routing expression that downstream
+    /// task specialization needs to build per-partition range filters.
+    ///
+    /// No-op if the stage's boundary root is an `AdaptiveDatafusionExec`
+    /// (query root) — a range repartition writes at a shuffle boundary, so
+    /// the parking slot only exists on the `ExchangeExec` variant.
+    pub(super) fn set_range_repartition_routing(
+        &mut self,
+        stage_id: usize,
+        routing: RangeRepartitionRouting,
+    ) -> common::Result<()> {
+        if let Some(exchange) = self
+            .runnable_stage_cache
+            .get(&stage_id)
+            .and_then(|stage| stage.downcast_ref::<ExchangeExec>())
+        {
+            exchange.resolve_range_repartition_routing(routing);
+        }
+        Ok(())
+    }
+
     pub(super) fn finalise_stage_internal(
         &mut self,
         stage_id: usize,
@@ -261,9 +287,12 @@ impl AdaptivePlanner {
         }
     }
 
-    /// Once all tasks has been completed marks stage as resolved
-    /// and returns partition allocations
-    pub fn finalise_stage(
+    /// Once all tasks have completed, pop the accumulated stage output as a
+    /// K-shaped `Vec<Vec<PartitionLocation>>` (or the broadcast-shape variant)
+    /// *without* parking it on the ExchangeExec. Caller can post-process
+    /// (e.g. range-repartition overlap remap) before calling
+    /// [`resolve_stage_partitions`](Self::resolve_stage_partitions).
+    pub fn take_stage_output_partitions(
         &mut self,
         stage_id: usize,
     ) -> common::Result<Vec<Vec<PartitionLocation>>> {
@@ -293,10 +322,20 @@ impl AdaptivePlanner {
                 ))?
                 .partition_locations(output_partition_count)
         };
-        self.finalise_stage_internal(stage_id, stage_output.clone())?;
         Ok(stage_output)
     }
 
+    /// Park the given partition list on the stage's ExchangeExec and trigger
+    /// a replan. Pairs with
+    /// [`take_stage_output_partitions`](Self::take_stage_output_partitions).
+    pub fn resolve_stage_partitions(
+        &mut self,
+        stage_id: usize,
+        partitions: Vec<Vec<PartitionLocation>>,
+    ) -> common::Result<()> {
+        self.finalise_stage_internal(stage_id, partitions)
+    }
+
     /// Replans the stages by applying physical optimizations.
     ///
     /// # Returns
diff --git a/ballista/scheduler/src/state/task_builder.rs b/ballista/scheduler/src/state/task_builder.rs
index 3c4c208db1..2982090356 100644
--- a/ballista/scheduler/src/state/task_builder.rs
+++ b/ballista/scheduler/src/state/task_builder.rs
@@ -36,7 +36,7 @@
 //! flows from parent to descendants via function arguments, so sibling
 //! subtrees never share state and there's no traversal-order dependency.
 
-use ballista_core::execution_plans::ShuffleReaderExec;
+use ballista_core::execution_plans::{PerPartitionFilterExec, ShuffleReaderExec};
 use datafusion::datasource::memory::MemorySourceConfig;
 use datafusion::datasource::physical_plan::{
     FileGroup, FileScanConfig, FileScanConfigBuilder,
@@ -83,6 +83,28 @@ fn restrict(
         return Ok(rewritten);
     }
 
+    // PerPartitionFilterExec: its `predicates` vec is positionally aligned
+    // with the child's output partitions (predicates[k] filters
+    // input.execute(k)). When we restrict the child from K partitions to
+    // `partitions.len()`, the predicate vec must be sliced by the same
+    // indices in the same order — otherwise the operator's
+    // `predicates.len() == child.partition_count()` invariant breaks and
+    // even if it didn't, task-local partition `j` would be filtered
+    // through the wrong global predicate. Under `under_collect`, neither
+    // child nor predicates get sliced and the generic path below applies.
+    if !under_collect && let Some(ppf) = plan.downcast_ref::<PerPartitionFilterExec>() {
+        let child = plan.children()[0].clone();
+        let new_child = restrict(child, partitions, false)?;
+        let new_predicates: Vec<_> = partitions
+            .iter()
+            .map(|&p| ppf.predicates()[p].clone())
+            .collect();
+        return Ok(Arc::new(PerPartitionFilterExec::try_new(
+            new_child,
+            new_predicates,
+        )?));
+    }
+
     // UnionExec: parent partition `p` maps to exactly one child's local
     // partition (`p` minus the sum of preceding children's counts). Split
     // `partitions` into disjoint per-child sub-slices and recurse; a child
@@ -624,4 +646,73 @@ mod tests {
             "streaming right side is pinned to this task's partition"
         );
     }
+
+    /// A `PerPartitionFilterExec` restricted to a subset of partitions must
+    /// slice its `predicates` vector by the same indices, in the same order,
+    /// as its child. Otherwise the operator's construction invariant
+    /// (`predicates.len() == child.partition_count()`) breaks and
+    /// task-local partition `j` would filter through a global predicate
+    /// that no longer matches.
+    #[test]
+    fn per_partition_filter_predicates_are_sliced_with_partitions() {
+        use ballista_core::execution_plans::PerPartitionFilterExec;
+        use datafusion::logical_expr::Operator;
+        use datafusion::physical_expr::expressions::{BinaryExpr, Column, Literal};
+        use datafusion::physical_expr::{Partitioning, PhysicalExpr};
+        use datafusion::scalar::ScalarValue;
+
+        // 4 upstream partitions, each with its own bespoke predicate so we
+        // can assert the slice ordering survives.
+        let schema = Arc::new(Schema::new(vec![Field::new("v", DataType::Int64, false)]));
+        let partitions_locs: Vec<Vec<PartitionLocation>> =
+            (0..4).map(|i| vec![create_partition(i)]).collect();
+        let reader = ShuffleReaderExec::try_new(
+            1,
+            partitions_locs,
+            schema.clone(),
+            Partitioning::UnknownPartitioning(4),
+        )
+        .unwrap();
+        let make_pred = |lo: i64| -> Arc<dyn PhysicalExpr> {
+            Arc::new(BinaryExpr::new(
+                Arc::new(Column::new("v", 0)),
+                Operator::GtEq,
+                Arc::new(Literal::new(ScalarValue::Int64(Some(lo)))),
+            ))
+        };
+        let predicates: Vec<Arc<dyn PhysicalExpr>> =
+            (0..4).map(|i| make_pred(i as i64 * 100)).collect();
+        let plan: Arc<dyn ExecutionPlan> = Arc::new(
+            PerPartitionFilterExec::try_new(
+                Arc::new(reader) as Arc<dyn ExecutionPlan>,
+                predicates.clone(),
+            )
+            .unwrap(),
+        );
+
+        let restricted = restrict_plan_to_partitions(plan, &[1, 3]).unwrap();
+        let ppf = restricted
+            .downcast_ref::<PerPartitionFilterExec>()
+            .expect("top must remain PerPartitionFilterExec");
+        assert_eq!(ppf.predicates().len(), 2);
+        assert_eq!(
+            ppf.predicates()[0].to_string(),
+            predicates[1].to_string(),
+            "local partition 0 must carry the global-partition-1 predicate"
+        );
+        assert_eq!(
+            ppf.predicates()[1].to_string(),
+            predicates[3].to_string(),
+            "local partition 1 must carry the global-partition-3 predicate"
+        );
+
+        // Reader below must have been restricted in the same order.
+        let child = ppf.children()[0].clone();
+        let reader = child
+            .downcast_ref::<ShuffleReaderExec>()
+            .expect("child must be a ShuffleReaderExec");
+        assert_eq!(reader.partition.len(), 2);
+        assert_eq!(reader.partition[0][0].partition_id.partition_id, 1);
+        assert_eq!(reader.partition[1][0].partition_id.partition_id, 3);
+    }
 }