From c9e6d997eebabea8c31509b07a9113f93bb2dfbb Mon Sep 17 00:00:00 2001 From: zz_y Date: Sun, 13 Sep 2026 11:10:48 -0600 Subject: [PATCH 1/2] fix: publish storage routing from the physical plan The physical-plan publication path passed `None` for storage routing, so a deployment planned through `compile-and-publish` installed no routing table. The backend then falls through to its default engine and archive-shape queries (`histogram_quantile`, `delta`, `deriv`, `absent`, `rate_post_hoc`, and `topk`/`count` without a matching sketch) miss. Only the legacy `POST /api/v1/plan` path emitted routing, via `emit_backend_storage_routing`. Derive routing from the compiled plan instead. `build_routing_entry` only ever needed the per-metric sketch algorithms, so it now takes those directly and `storage_routing_document` is the shared entry point. The `BackendStageConfig` emitters project onto it and the physical compiler accumulates algorithms while compiling aggregations, so both publication paths run one classifier over the same decisions. Co-Authored-By: Claude Opus 5 (1M context) --- control_plane/src/emit/stage_config.rs | 56 +++++++++++++++++-------- control_plane/src/main.rs | 6 ++- control_plane/src/physical/compiler.rs | 57 ++++++++++++++++++++++++++ 3 files changed, 101 insertions(+), 18 deletions(-) diff --git a/control_plane/src/emit/stage_config.rs b/control_plane/src/emit/stage_config.rs index a4ab229e..b46a5e62 100644 --- a/control_plane/src/emit/stage_config.rs +++ b/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 { - let mut metrics_json: Vec = 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)> = 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)], +) -> JsonValue { + let metrics_json: Vec = 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 { + 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 = 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 `` is either `{ "engine": , "applies_to_query_shape": [...] }` /// or `{ "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(); diff --git a/control_plane/src/main.rs b/control_plane/src/main.rs index 08790ef8..e108f2e2 100644 --- a/control_plane/src/main.rs +++ b/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 ( diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index 2ea40ab2..c9e365f9 100644 --- a/control_plane/src/physical/compiler.rs +++ b/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, pub cost_comparison: Option, @@ -1058,6 +1063,12 @@ impl PhysicalCompiler { )?; let mut lifecycle_estimates = BTreeMap::::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::>::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,6 +1882,10 @@ 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::>(), + ); Ok(PhysicalPlan { envelope, summary_catalog, @@ -1872,6 +1893,7 @@ impl PhysicalCompiler { 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,41 @@ 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 = 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() { From 17d087db00db849ff37f20a8741140dadac78bcf Mon Sep 17 00:00:00 2001 From: zz_y Date: Sun, 13 Sep 2026 15:03:30 -0600 Subject: [PATCH 2/2] style: rustfmt --- control_plane/src/physical/compiler.rs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index c9e365f9..64140372 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -3596,7 +3596,9 @@ pub(crate) mod tests { // 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 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 = routing["metrics"]