Publish storage routing from the physical plan

control_plane/src/emit/stage_config.rs

@@ -715,15 +715,43 @@ pub fn emit_backend_storage_routing_for_tenant(
     tenant: &str,
     metric_plans: &[(String, &BackendStageConfig)],
 ) -> Result<JsonValue> {
-    let mut metrics_json: Vec<JsonValue> = Vec::with_capacity(metric_plans.len());
-    for (metric_name, backend_cfg) in metric_plans {
-        metrics_json.push(build_routing_entry(metric_name, backend_cfg));
-    }
-    Ok(json!({
+    let by_metric: Vec<(String, Vec<SketchAlgorithm>)> = metric_plans
+        .iter()
+        .map(|(metric_name, cfg)| (metric_name.clone(), routed_algorithms(cfg)))
+        .collect();
+    Ok(storage_routing_document(tenant, &by_metric))
+}
+
+/// Build the storage-routing document from the per-metric sketch algorithms a
+/// planning cycle decided to materialize. This is the shared entry point: the
+/// `BackendStageConfig` emitters above project onto it, and the physical
+/// compiler calls it directly from its own compiled aggregations, so both
+/// publication paths derive routing from one classifier.
+pub fn storage_routing_document(
+    tenant: &str,
+    metric_algorithms: &[(String, Vec<SketchAlgorithm>)],
+) -> JsonValue {
+    let metrics_json: Vec<JsonValue> = metric_algorithms
+        .iter()
+        .map(|(metric_name, algorithms)| build_routing_entry(metric_name, algorithms))
+        .collect();
+    json!({
         "tenant": tenant,
         "default_engine": "asap_query",
         "metrics": metrics_json,
-    }))
+    })
+}
+
+/// Sketch algorithms a `BackendStageConfig` materializes, in aggregation order.
+/// Non-sketch families contribute no routing capability.
+fn routed_algorithms(cfg: &BackendStageConfig) -> Vec<SketchAlgorithm> {
+    cfg.aggregations
+        .iter()
+        .filter_map(|a| match &a.family {
+            SummaryFamilyType::Sketch(kind, _) => Some(kind.algorithm().clone()),
+            _ => None,
+        })
+        .collect()
 }
 
 /// same as [`emit_backend_storage_routing`] but also
@@ -766,7 +794,10 @@ pub fn emit_backend_storage_routing_with_prometheus_for_tenant(
     let mut metrics_json: Vec<JsonValue> =
         Vec::with_capacity(metric_plans.len() + mode3_metrics.len());
     for (metric_name, backend_cfg) in metric_plans {
-        metrics_json.push(build_routing_entry(metric_name, backend_cfg));
+        metrics_json.push(build_routing_entry(
+            metric_name,
+            &routed_algorithms(backend_cfg),
+        ));
     }
     for metric_name in mode3_metrics {
         // Mode 3 — Prometheus owns the storage. Single target,
@@ -799,16 +830,7 @@ pub fn emit_backend_storage_routing_with_prometheus_for_tenant(
 /// ```
 /// where each `<target>` is either `{ "engine": <engine>, "applies_to_query_shape": [...] }`
 /// or `{ "engine": <engine> }` for the default slot.
-fn build_routing_entry(metric_name: &str, cfg: &BackendStageConfig) -> JsonValue {
-    let algorithms: Vec<&SketchAlgorithm> = cfg
-        .aggregations
-        .iter()
-        .filter_map(|a| match &a.family {
-            SummaryFamilyType::Sketch(kind, _) => Some(kind.algorithm()),
-            _ => None,
-        })
-        .collect();
-
+fn build_routing_entry(metric_name: &str, algorithms: &[SketchAlgorithm]) -> JsonValue {
     // Sketch-eligible shapes — the ASAP tier serves these natively
     // because we planned a sketch for them.
     let mut warm_shapes: Vec<&'static str> = Vec::new();

control_plane/src/main.rs

@@ -687,7 +687,11 @@ async fn compile_and_publish_physical_plan(
         }
     };
     if let Err(error) = backend
-        .post_catalog_plan_typed(&publication, None, &adaptation_evidence)
+        .post_catalog_plan_typed(
+            &publication,
+            Some(bundle.storage_routing.clone()),
+            &adaptation_evidence,
+        )
         .await
     {
         return (

control_plane/src/physical/compiler.rs

@@ -406,6 +406,11 @@ pub struct PhysicalPlan {
     pub precompute_plan: PrecomputePlan,
     pub transmission_plan: TransmissionPlan,
     pub query_plan: QueryPlan,
+    /// Backend storage-routing table for the metrics this plan materializes.
+    /// Published alongside the plan so the query-shape → engine split always
+    /// describes the generation that is actually installed; without it the
+    /// backend falls back to its default engine and archive-shape queries miss.
+    pub storage_routing: serde_json::Value,
     /// Lifecycle component only, not a complete physical-plan comparison.
     pub lifecycle_estimates: Vec<MaterializationLifecycleEstimate>,
     pub cost_comparison: Option<super::workload_cost::WorkloadCostComparison>,
@@ -1058,6 +1063,12 @@ impl PhysicalCompiler {
         )?;
         let mut lifecycle_estimates =
             BTreeMap::<asap_types::PolicyFingerprint, MaterializationLifecycleEstimate>::new();
+        // Per-metric sketch families this cycle materializes. The storage-routing
+        // classifier turns them into the warm/archive shape split, so routing is
+        // derived from the same decisions that produced the materializations
+        // rather than from a separately maintained table.
+        let mut routed_algorithms =
+            BTreeMap::<String, Vec<planner_types::post_asap::SketchAlgorithm>>::new();
 
         for (query_index, query) in request.queries.iter().enumerate() {
             let evidence = request.evidence.get(&query.query_id);
@@ -1233,6 +1244,12 @@ impl PhysicalCompiler {
                     aggregation_id.clone(),
                     environment.target,
                 );
+                if let SummaryFamilyType::Sketch(kind, _) = &aggregation.family {
+                    routed_algorithms
+                        .entry(aggregation.metric_name.clone())
+                        .or_default()
+                        .push(kind.algorithm().clone());
+                }
                 let precompute_materialization =
                     scoped_materialization(&aggregation, &selected.node)?;
                 let materialization = precompute_materialization.policy_fingerprint();
@@ -1865,13 +1882,18 @@ impl PhysicalCompiler {
                 })?;
         }
         query_plan.validate_against_catalog(&summary_catalog)?;
+        let storage_routing = crate::emit::stage_config::storage_routing_document(
+            crate::emit::stage_config::DEFAULT_TENANT,
+            &routed_algorithms.into_iter().collect::<Vec<_>>(),
+        );
         Ok(PhysicalPlan {
             envelope,
             summary_catalog,
             collector_plans,
             precompute_plan,
             transmission_plan,
             query_plan,
+            storage_routing,
             lifecycle_estimates: lifecycle_estimates.into_values().collect(),
             cost_comparison: None,
             logical_selection: request.logical_selection,
@@ -3569,6 +3591,43 @@ fn stable_workload_plan_id(
 pub(crate) mod tests {
     use super::*;
 
+    // Every metric the plan materializes must carry a routing entry, or the
+    // backend falls through to its default engine and archive-shape queries
+    // miss. Publication passes this document straight to the data plane.
+    #[test]
+    fn compiled_plan_routes_every_materialized_metric() {
+        let plan = quoted_snapshot(planning_snapshot(), false)
+            .compile()
+            .unwrap();
+        let routing = &plan.storage_routing;
+        assert_eq!(routing["default_engine"], "asap_query");
+        let routed: std::collections::BTreeSet<String> = routing["metrics"]
+            .as_array()
+            .expect("routing document carries a metrics array")
+            .iter()
+            .map(|entry| entry["name"].as_str().unwrap().to_string())
+            .collect();
+        assert!(
+            !routed.is_empty(),
+            "a plan with materializations emitted an empty routing table: {routing}"
+        );
+        // Archive-shape claims are what the default engine cannot serve; a
+        // routing entry without them would silently widen the warm tier.
+        for entry in routing["metrics"].as_array().unwrap() {
+            let engines: Vec<&str> = entry["targets"]
+                .as_array()
+                .unwrap()
+                .iter()
+                .map(|t| t["engine"].as_str().unwrap())
+                .collect();
+            assert!(
+                engines.contains(&"asap_query"),
+                "metric {} lost its warm target: {entry}",
+                entry["name"]
+            );
+        }
+    }
+
     // Complete deployment quotes must preserve one producer with two window readouts.
     #[test]
     fn complete_cost_selection_preserves_shared_sum_panes() {