feat: execute Planner-selected exact per-series range-max indexes

control_plane/src/physical/compiler.rs

@@ -2139,7 +2139,7 @@ impl PhysicalCompiler {
             use std::hash::{Hash, Hasher};
             let mut hash = std::collections::hash_map::DefaultHasher::new();
             stable_workload_plan_id(&plan_materializations, &request.queries).hash(&mut hash);
-            "typed-local-residual-v1".hash(&mut hash);
+            "typed-local-residual-v2-range-max-index".hash(&mut hash);
             for query in &request.queries {
                 format!("{:?}", query.post_asap).hash(&mut hash);
             }

control_plane/src/physical/workload_cost.rs

@@ -207,13 +207,37 @@ pub fn manifest(
         // when no precomputed summary is installed. Deduplicate by source.
         for node in entry.nodes.values() {
             if let crate::query_plan::QueryPlanNode::Logical {
-                operator: crate::query_plan::logical::LogicalOperator::Scan { metric, .. },
+                operator:
+                    operator @ (crate::query_plan::logical::LogicalOperator::Scan { .. }
+                    | crate::query_plan::logical::LogicalOperator::ReadRangeMaxIndex {
+                        ..
+                    }),
                 ..
             } = node
             {
-                let metric = metric
-                    .as_ref()
-                    .ok_or_else(|| invalid("local raw scan requires named-source pricing"))?;
+                let metric = match operator {
+                    crate::query_plan::logical::LogicalOperator::Scan { metric, .. } => metric
+                        .as_ref()
+                        .ok_or_else(|| invalid("local raw scan requires named-source pricing"))?,
+                    crate::query_plan::logical::LogicalOperator::ReadRangeMaxIndex {
+                        metric,
+                        ..
+                    } => metric,
+                    _ => unreachable!(),
+                };
+                if matches!(
+                    operator,
+                    crate::query_plan::logical::LogicalOperator::ReadRangeMaxIndex { .. }
+                ) {
+                    for operation in ["build", "update", "residency", "retire"] {
+                        add(
+                            format!("range-max-index:{metric}:{operation}"),
+                            json!({"operation": operation, "metric": metric, "index": "exact_per_series_range_max_v1"}),
+                            "horizon",
+                            1.0,
+                        );
+                    }
+                }
                 let source = json!({"source": planner_types::pre_asap::Source::TimeSeries { metric: metric.clone() }, "location": "backend", "ingest": plan.precompute_plan.ingest});
                 add(format!("source:{}", source), source.clone(), "horizon", 1.0);
                 for operation in ["build", "update", "residency", "retire"] {
@@ -495,6 +519,52 @@ mod tests {
         snapshot
     }
 
+    #[test]
+    fn range_max_index_costs_share_state_across_filters_and_charge_retained_input() {
+        let mut snapshot = fixture();
+        let entries = snapshot.query_workload.repeating_queries.as_mut().unwrap();
+        entries[0].query = planner_types::workload::Query(
+            "max_over_time(service_retry_queue_depth{job=~\".+\"}[6h])".into(),
+        );
+        entries[0].requirements.accuracy = planner_types::workload::AccuracyRequirement::Explicit(
+            planner_types::types::AccuracyTarget::Exact,
+        );
+        let mut second = entries[0].clone();
+        second.query = planner_types::workload::Query(
+            "max_over_time(service_retry_queue_depth{job=\"order-service\"}[6h])".into(),
+        );
+        entries.push(second);
+        let (request, env) = snapshot.planning_request().unwrap();
+        let plan = PhysicalCompiler.compile(request.clone(), env).unwrap();
+        let costs = manifest(&plan, &request.queries).unwrap();
+        assert_eq!(
+            costs
+                .components
+                .keys()
+                .filter(|id| id.starts_with("range-max-index:"))
+                .count(),
+            4
+        );
+        assert_eq!(
+            costs
+                .components
+                .keys()
+                .filter(|id| id.starts_with("raw-state:"))
+                .count(),
+            4
+        );
+        for operation in ["build", "update", "residency", "retire"] {
+            assert_eq!(
+                costs.components[&format!("range-max-index:service_retry_queue_depth:{operation}")]
+                    .multiplicity,
+                1.0
+            );
+        }
+        assert_eq!(plan.query_plan.entries.values().flat_map(|entry| entry.nodes.values()).filter(|node|
+            matches!(node, crate::query_plan::QueryPlanNode::Logical { operator:
+                crate::query_plan::logical::LogicalOperator::ReadRangeMaxIndex { .. }, .. })).count(), 2);
+    }
+
     fn quoted() -> (
         Vec<PlanningRequest>,
         DeploymentEnvironment,

control_plane/src/query_plan.rs

@@ -429,6 +429,18 @@ where
         let id = QueryNodeId(self.next_id);
         self.next_id += 1;
         self.seen.insert(identity, id);
+        if let Some(original) = &self.logical_source {
+            if let Some(operator) = logical::selected_range_max_index(original, node)? {
+                self.nodes.insert(
+                    id,
+                    QueryPlanNode::Logical {
+                        operator,
+                        inputs: vec![],
+                    },
+                );
+                return Ok(id);
+            }
+        }
         let residual = match (&self.logical_source, &node.expr) {
             (Some(original), SummaryExpr::KeepPreAsap(expr)) => {
                 Some(logical::residual_nodes(original, expr)?)

control_plane/src/query_plan/logical.rs

@@ -12,6 +12,13 @@ use std::collections::BTreeMap;
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
 #[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
 pub enum LogicalOperator {
+    /// Planner-selected exact per-series MinMax state, queried as max over an
+    /// exact event-time interval. This is an installed index, not a raw scan.
+    ReadRangeMaxIndex {
+        metric: String,
+        matchers: Vec<LabelMatcher>,
+        range_ms: u64,
+    },
     Scan {
         metric: Option<String>,
         matchers: Vec<LabelMatcher>,
@@ -117,7 +124,7 @@ fn offset(value: &Option<Offset>) -> Result<i64, QueryPlanError> {
 impl LogicalOperator {
     pub fn validate(&self, inputs: usize) -> Result<(), QueryPlanError> {
         let expected = match self {
-            Self::Scan { .. } => 0,
+            Self::Scan { .. } | Self::ReadRangeMaxIndex { .. } => 0,
             Self::Binary { .. } | Self::HistogramQuantile => 2,
             _ => 1,
         };
@@ -133,6 +140,14 @@ impl LogicalOperator {
         ) {
             return Err(invalid("zero range"));
         }
+        if let Self::ReadRangeMaxIndex {
+            metric, range_ms, ..
+        } = self
+        {
+            if metric.is_empty() || *range_ms == 0 || *range_ms > i64::MAX as u64 {
+                return Err(invalid("invalid exact range-max index contract"));
+            }
+        }
         if let Self::Subquery {
             range_ms, step_ms, ..
         } = self
@@ -791,3 +806,113 @@ mod planner_workload_tests {
         }
     }
 }
+
+/// Preserve the exact original operator direction because MinMax family alone
+/// does not distinguish min from max. The full Planner-node witness is required.
+pub(super) fn selected_range_max_index(
+    original: &str,
+    node: &planner_types::post_asap::SummaryNode,
+) -> Result<Option<LogicalOperator>, QueryPlanError> {
+    use planner_types::post_asap::{ExactKind, SummaryExpr, SummaryFamilyType};
+    if !matches!(
+        &node.expr,
+        SummaryExpr::SummaryAgg {
+            family: SummaryFamilyType::ExactAggregate(ExactKind::MinMax, _),
+            reduction: planner_types::pre_asap::Reduction::PerEntity,
+            ..
+        }
+    ) {
+        return Ok(None);
+    }
+    let (root, nodes) = selected_residual_nodes(original, node)?;
+    let Some(QueryPlanNode::Logical {
+        operator:
+            LogicalOperator::Temporal {
+                operation: TemporalOperation::Max,
+            },
+        inputs,
+    }) = nodes.get(&root)
+    else {
+        return Ok(None);
+    };
+    if inputs.len() != 1 || nodes.len() != 2 {
+        return Ok(None);
+    }
+    let Some(QueryPlanNode::Logical {
+        operator:
+            LogicalOperator::Scan {
+                metric: Some(metric),
+                matchers,
+                range_ms: Some(range_ms),
+                offset_ms: 0,
+            },
+        ..
+    }) = nodes.get(&inputs[0])
+    else {
+        return Ok(None);
+    };
+    Ok(Some(LogicalOperator::ReadRangeMaxIndex {
+        metric: metric.clone(),
+        matchers: matchers.clone(),
+        range_ms: *range_ms,
+    }))
+}
+
+#[cfg(test)]
+mod range_max_index_tests {
+    use super::*;
+    #[test]
+    fn real_gauge_queries_have_planner_authorized_exact_indexes() {
+        for (query, metric, range_ms) in [
+            (
+                r#"max_over_time(service_cache_refresh_lag_seconds{job="user-service"}[12h])"#,
+                "service_cache_refresh_lag_seconds",
+                43_200_000,
+            ),
+            (
+                r#"max_over_time(service_retry_queue_depth{job=~".+"}[6h])"#,
+                "service_retry_queue_depth",
+                21_600_000,
+            ),
+            (
+                r#"max_over_time(service_retry_queue_depth{job="order-service"}[6h])"#,
+                "service_retry_queue_depth",
+                21_600_000,
+            ),
+        ] {
+            let original = crate::query_parser::parse_query_expr_canonical(
+                query,
+                planner_types::types::AccuracyTarget::Exact,
+            )
+            .unwrap();
+            let selected = crate::planner_selection::select_summary_default(&original).unwrap();
+            let index = selected_range_max_index(query, &selected).unwrap().unwrap();
+            assert!(
+                matches!(index, LogicalOperator::ReadRangeMaxIndex { metric: ref actual, range_ms: actual_range, .. } if actual == metric && actual_range == range_ms)
+            );
+            index.validate(0).unwrap();
+            assert!(index.validate(1).is_err());
+        }
+    }
+    #[test]
+    fn min_and_shifted_or_nested_windows_do_not_become_max_indexes() {
+        for query in [
+            "min_over_time(m[1m])",
+            "max_over_time(m[1m] offset 1m)",
+            "max_over_time((m + m)[1m:1s])",
+        ] {
+            let original = crate::query_parser::parse_query_expr_canonical(
+                query,
+                planner_types::types::AccuracyTarget::Exact,
+            )
+            .unwrap();
+            let selected = crate::planner_selection::select_summary_default(&original).unwrap();
+            assert!(
+                selected_range_max_index(query, &selected)
+                    .unwrap()
+                    .is_none(),
+                "{query}"
+            );
+        }
+    }
+}