feat(asapquery): execute planned PromQL with conformant fallback

control_plane/src/backend_plan/from_stage_config.rs

@@ -77,7 +77,36 @@ pub fn from_stage_config(
         );
     }
 
-    let mut routing = Vec::with_capacity(cfg.readouts.len());
+    // Exact aggregates are already finalized by their accumulator and do not
+    // have a SketchQuery readout node. Their warm route therefore comes from
+    // the physical aggregation itself; approximate routes remain readout-
+    // driven below.
+    let mut routing = cfg
+        .aggregations
+        .iter()
+        .filter_map(|agg| {
+            let planner_types::post_asap::SummaryFamilyType::ExactAggregate(kind, _) = &agg.family
+            else {
+                return None;
+            };
+            let fingerprint = *fingerprint_by_agg_id.get(agg.aggregation_id.as_str())?;
+            let agg_type = match kind {
+                planner_types::post_asap::ExactKind::Sum
+                | planner_types::post_asap::ExactKind::Count => asap_types::AggregationType::Sum,
+                planner_types::post_asap::ExactKind::MinMax => asap_types::AggregationType::MinMax,
+                planner_types::post_asap::ExactKind::Increase
+                | planner_types::post_asap::ExactKind::Rate => {
+                    asap_types::AggregationType::Increase
+                }
+            };
+            Some(RoutingEntry {
+                satisfies: Capability::ExactAgg(agg_type),
+                materialization: fingerprint,
+                storage_backend: StorageBackend::SketchStore,
+            })
+        })
+        .collect::<Vec<_>>();
+    routing.reserve(cfg.readouts.len());
     for readout in &cfg.readouts {
         let Some(&fingerprint) = fingerprint_by_agg_id.get(readout.aggregation_id.as_str()) else {
             // Orphan readout (no matching aggregation in this cycle's
@@ -94,11 +123,14 @@ pub fn from_stage_config(
             continue;
         };
         let satisfies = capability_for_readout(agg, &readout.op)?;
-        routing.push(RoutingEntry {
+        let route = RoutingEntry {
             satisfies,
             materialization: fingerprint,
             storage_backend: StorageBackend::SketchStore,
-        });
+        };
+        if !routing.contains(&route) {
+            routing.push(route);
+        }
     }
 
     let plan_monitors = monitors

control_plane/src/physical/compiler.rs

@@ -1713,13 +1713,14 @@ impl PhysicalCompiler {
             for (ordinal, selected) in selected.into_iter().enumerate() {
                 let metric = selected.metric.clone();
                 let aggregation_id = format!("{}:{ordinal}:{}", query.query_id, metric);
+                // Rate is a readout over the same reset-aware counter state
+                // as Increase. Keep that semantic distinction in QueryPlan,
+                // while the physical store binds both to Increase state.
+                let physical_family = physical_materialization_family(&selected.family);
                 let aggregation = BackendAggregation {
                     aggregation_id: aggregation_id.clone(),
                     metric_name: metric.clone(),
-                    family: SummaryFamilyType::Sketch(
-                        selected.kind.clone(),
-                        planner_types::post_asap::GroupingStrategy::PerSubpopulationInstance,
-                    ),
+                    family: physical_family,
                     window_secs: query.window_secs,
                     spatial_filter: String::new(),
                     grouping: query.group_by.clone(),
@@ -1756,16 +1757,18 @@ impl PhysicalCompiler {
                     }
                 }
                 aggregations.push(aggregation);
-                readouts.push(BackendReadout {
-                    aggregation_id,
-                    op: selected.readout.clone(),
-                });
+                if let Some(readout) = selected.readout.clone() {
+                    readouts.push(BackendReadout {
+                        aggregation_id,
+                        op: readout,
+                    });
+                }
                 collector_materializations.push(CollectorMaterialization {
                     query_id: query.query_id.clone(),
                     materialization,
                     metric: metric.clone(),
-                    algorithm: format!("{:?}", selected.kind.algorithm()).to_ascii_lowercase(),
-                    parameters: sketch_params_json(&selected.params),
+                    algorithm: selected.algorithm,
+                    parameters: selected.parameters,
                     group_by: query.group_by.clone(),
                     window_secs: query.window_secs,
                     abstract_window_framework: planner_selection.window_framework.clone(),
@@ -1909,7 +1912,7 @@ impl PhysicalCompiler {
                                 fingerprint.0
                             ))
                         })?;
-                    if &materialization.family != node_family
+                    if materialization.family != physical_materialization_family(node_family)
                         || materialization.window.size_ms != query.window_secs.saturating_mul(1_000)
                         || materialization.group_by != query.group_by
                     {
@@ -2264,12 +2267,13 @@ fn select_lifecycle(
     })
 }
 
-struct SelectedSketch {
+struct SelectedMaterialization {
     node_identity: usize,
     metric: String,
-    kind: planner_types::post_asap::SketchKind,
-    params: planner_types::post_asap::SketchParams,
-    readout: SketchQuery,
+    family: SummaryFamilyType,
+    readout: Option<SketchQuery>,
+    algorithm: String,
+    parameters: Value,
 }
 
 /// Collect every executable materialization leaf in the selected post-ASAP
@@ -2280,11 +2284,11 @@ struct SelectedSketch {
 /// fallback node and no unused warm state is provisioned.
 fn collect_selected_materializations(
     node: &Rc<SummaryNode>,
-) -> Result<Vec<SelectedSketch>, String> {
+) -> Result<Vec<SelectedMaterialization>, String> {
     fn walk(
         node: &Rc<SummaryNode>,
         readout: Option<&SketchQuery>,
-        selected: &mut Vec<SelectedSketch>,
+        selected: &mut Vec<SelectedMaterialization>,
     ) -> Result<(), String> {
         match &node.expr {
             SummaryExpr::SummaryEstimate {
@@ -2304,15 +2308,35 @@ fn collect_selected_materializations(
                     let metric = summary_agg_metric(node).ok_or_else(|| {
                         "SummaryAgg has no unique time-series source in post-ASAP IR".to_string()
                     })?;
-                    selected.push(SelectedSketch {
+                    selected.push(SelectedMaterialization {
                         node_identity: Rc::as_ptr(node) as usize,
                         metric,
-                        kind: kind.clone(),
-                        params: kind.params().clone(),
-                        readout: readout.clone(),
+                        family: SummaryFamilyType::Sketch(
+                            kind.clone(),
+                            planner_types::post_asap::GroupingStrategy::PerSubpopulationInstance,
+                        ),
+                        readout: Some(readout.clone()),
+                        algorithm: format!("{:?}", kind.algorithm()).to_ascii_lowercase(),
+                        parameters: sketch_params_json(kind.params()),
                     });
                 }
             }
+            SummaryExpr::SummaryAgg {
+                family: SummaryFamilyType::ExactAggregate(kind, params),
+                ..
+            } => {
+                let metric = summary_agg_metric(node).ok_or_else(|| {
+                    "SummaryAgg has no unique time-series source in post-ASAP IR".to_string()
+                })?;
+                selected.push(SelectedMaterialization {
+                    node_identity: Rc::as_ptr(node) as usize,
+                    metric,
+                    family: SummaryFamilyType::ExactAggregate(kind.clone(), params.clone()),
+                    readout: None,
+                    algorithm: format!("{kind:?}").to_ascii_lowercase(),
+                    parameters: Value::Object(Default::default()),
+                });
+            }
             SummaryExpr::KeepPreAsap(_)
             | SummaryExpr::SummaryAgg { .. }
             | SummaryExpr::SummaryJoin { .. }
@@ -2327,6 +2351,18 @@ fn collect_selected_materializations(
     Ok(selected)
 }
 
+fn physical_materialization_family(family: &SummaryFamilyType) -> SummaryFamilyType {
+    match family {
+        SummaryFamilyType::ExactAggregate(planner_types::post_asap::ExactKind::Rate, _) => {
+            SummaryFamilyType::ExactAggregate(
+                planner_types::post_asap::ExactKind::Increase,
+                planner_types::post_asap::ExactParams::Increase,
+            )
+        }
+        _ => family.clone(),
+    }
+}
+
 fn sketch_params_json(params: &planner_types::post_asap::SketchParams) -> Value {
     use planner_types::post_asap::SketchParams as P;
     match params {
@@ -2688,6 +2724,53 @@ mod tests {
         assert_eq!(bundle.envelope.planner_revision, PLANNER_REVISION);
     }
 
+    #[test]
+    fn backend_local_compiler_materializes_declared_exact_promql_cases() {
+        for (query_id, promql, expected_readout) in [
+            (
+                "q-rate",
+                "rate(m[1m])",
+                crate::query_plan::ExactReadout::Rate,
+            ),
+            (
+                "q-increase",
+                "increase(m[1m])",
+                crate::query_plan::ExactReadout::Increase,
+            ),
+            (
+                "q-sum",
+                "sum_over_time(m[1m])",
+                crate::query_plan::ExactReadout::Sum,
+            ),
+        ] {
+            let mut deployment = environment(10_000);
+            deployment.target = PhysicalDeploymentTarget::BackendLocalRemoteWrite;
+            deployment.collector_ids.clear();
+            let plan = PhysicalCompiler
+                .compile(request(query_id, promql), deployment)
+                .unwrap_or_else(|error| panic!("{promql} must compile: {error}"));
+            assert_eq!(plan.backend_plan.materializations.len(), 1, "{promql}");
+            assert_eq!(plan.query_plan.entries.len(), 1, "{promql}");
+            assert!(plan.collector_plans.is_empty(), "{promql}");
+            let entry = plan.query_plan.entries.values().next().unwrap();
+            assert!(matches!(
+                entry.nodes.get(&entry.root),
+                Some(crate::query_plan::QueryPlanNode::ExactReadout { readout, .. })
+                    if *readout == expected_readout
+            ));
+            if expected_readout == crate::query_plan::ExactReadout::Rate {
+                let materialization = plan.backend_plan.materializations.values().next().unwrap();
+                assert_eq!(
+                    materialization.family,
+                    SummaryFamilyType::ExactAggregate(
+                        planner_types::post_asap::ExactKind::Increase,
+                        planner_types::post_asap::ExactParams::Increase,
+                    )
+                );
+            }
+        }
+    }
+
     #[test]
     fn checked_in_backend_local_snapshot_is_canonical_and_compilable() {
         let source = include_str!("../../../docs/examples/asapquery-planning-snapshot.json");

control_plane/src/query_plan.rs

@@ -241,6 +241,10 @@ pub enum QueryPlanNode {
         input: QueryNodeId,
         query: QueryReadout,
     },
+    ExactReadout {
+        input: QueryNodeId,
+        readout: ExactReadout,
+    },
     SummaryMerge {
         inputs: Vec<QueryNodeId>,
     },
@@ -253,12 +257,22 @@ impl QueryPlanNode {
     pub fn inputs(&self) -> &[QueryNodeId] {
         match self {
             Self::ReadMaterialization { .. } | Self::ExactFallback { .. } => &[],
-            Self::SummaryEstimate { input, .. } => std::slice::from_ref(input),
+            Self::SummaryEstimate { input, .. } | Self::ExactReadout { input, .. } => {
+                std::slice::from_ref(input)
+            }
             Self::SummaryMerge { inputs } => inputs,
         }
     }
 }
 
+#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(rename_all = "snake_case")]
+pub enum ExactReadout {
+    Sum,
+    Increase,
+    Rate,
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
 #[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
 pub enum QueryReadout {
@@ -325,19 +339,24 @@ where
                 reduction,
                 child,
                 ..
-            } => {
-                if !matches!(
-                    family,
-                    SummaryFamilyType::ExactAggregate(..) | SummaryFamilyType::Sketch(..)
-                ) {
-                    return Err(QueryPlanError::UnsupportedNode(format!(
-                        "summary family {family:?}"
-                    )));
+            } => match family {
+                SummaryFamilyType::ExactAggregate(..) | SummaryFamilyType::Sketch(..) => {
+                    let mut binding = (self.bind)(node, family)?;
+                    binding.output_grouping = physical_grouping(reduction, child)?;
+                    if let Some(readout) = exact_readout(family) {
+                        let input = QueryNodeId(self.next_id);
+                        self.next_id += 1;
+                        self.nodes
+                            .insert(input, QueryPlanNode::ReadMaterialization { binding });
+                        QueryPlanNode::ExactReadout { input, readout }
+                    } else {
+                        QueryPlanNode::ReadMaterialization { binding }
+                    }
                 }
-                let mut binding = (self.bind)(node, family)?;
-                binding.output_grouping = physical_grouping(reduction, child)?;
-                QueryPlanNode::ReadMaterialization { binding }
-            }
+                other => QueryPlanNode::ExactFallback {
+                    reason: format!("summary family {other:?} is not executable by the warm tier"),
+                },
+            },
             SummaryExpr::SummaryEstimate {
                 summary_input,
                 query,
@@ -374,6 +393,16 @@ where
     }
 }
 
+fn exact_readout(family: &SummaryFamilyType) -> Option<ExactReadout> {
+    use planner_types::post_asap::ExactKind;
+    match family {
+        SummaryFamilyType::ExactAggregate(ExactKind::Sum, _) => Some(ExactReadout::Sum),
+        SummaryFamilyType::ExactAggregate(ExactKind::Increase, _) => Some(ExactReadout::Increase),
+        SummaryFamilyType::ExactAggregate(ExactKind::Rate, _) => Some(ExactReadout::Rate),
+        _ => None,
+    }
+}
+
 fn physical_grouping(
     reduction: &Reduction,
     child: &SummaryNode,

data_plane/src/drivers/query/adapters/prometheus_http.rs

@@ -158,7 +158,11 @@ impl PrometheusHttpAdapter {
                 .as_secs_f64()
         };
 
-        Ok(ParsedQueryRequest { query, time })
+        Ok(ParsedQueryRequest {
+            query,
+            time,
+            timeout: params.get("timeout").cloned(),
+        })
     }
 
     /// Helper to parse range query parameters
@@ -206,6 +210,7 @@ impl PrometheusHttpAdapter {
             start,
             end,
             step,
+            timeout: params.get("timeout").cloned(),
         })
     }
 }

data_plane/src/drivers/query/adapters/traits.rs

@@ -16,6 +16,8 @@ use crate::query_engines::QueryResult;
 pub struct ParsedQueryRequest {
     pub query: String,
     pub time: f64,
+    /// Prometheus timeout syntax, preserved verbatim for exact fallback.
+    pub timeout: Option<String>,
 }
 
 /// Parsed range query request with validated parameters
@@ -25,6 +27,8 @@ pub struct ParsedRangeQueryRequest {
     pub start: f64, // epoch seconds
     pub end: f64,   // epoch seconds
     pub step: f64,  // seconds, must be multiple of tumbling window
+    /// Prometheus timeout syntax, preserved verbatim for exact fallback.
+    pub timeout: Option<String>,
 }
 
 /// Result of query execution (before formatting for protocol)