backend-plan: serving-time cutover -- prefer plan over SketchStore reconstruction (Phase 4)

control_plane/src/sketch_algebra/cost_model.rs

@@ -399,6 +399,15 @@ impl CostModel for ForcedFamilyCostModel {
 /// `size_params` then fall back to the accuracy-driven default, which
 /// won't match anything registered either way, so the outcome
 /// (`find_candidates` finds nothing) is unchanged.
+///
+/// **Fallback status (design-backend-plan-wire-format.md §5):**
+/// `l4_lowering.rs` prefers reading planning's decision directly off an
+/// installed `BackendPlan`'s materializations (no reconstruction needed
+/// there — `Materialization.kind`/`.params` already ARE the pair
+/// `observed` needs). This type's caller
+/// (`observed_family_for_metric`, the `SketchStore`-metadata
+/// reconstruction) is the fallback for deploys with no `BackendPlan`
+/// installed yet, or for metrics a partial/stale plan doesn't cover.
 pub struct ObservedFamilyCostModel {
     inner: ControlPlaneCostModel,
     observed: Option<(SummaryKind, SummaryParams)>,

data_plane/src/main.rs

@@ -418,6 +418,18 @@ async fn main() -> Result<()> {
         None
     };
 
+    // BackendPlan wire format (design-backend-plan-wire-format.md):
+    // install an empty hot-reload handle so `GET/POST
+    // /api/v1/backend-plan` don't 503 before the control plane's first
+    // push lands — same "install empty, let the first push fill it in"
+    // pattern as `bootstrap_routing` below. Shared with both the query
+    // engine (serving-time cutover, Phase 4) and the HTTP server (the
+    // push target) so a POST is observable by the next query, same
+    // sharing contract as `hot_reload_config`.
+    let hot_reload_backend_plan = data_plane::storage_engines::types::HotReloadBackendPlan::new(
+        control_plane::backend_plan::BackendPlan::default(),
+    );
+
     // Setup query engine. ASAPQueryEngine shares the same
     // HotReloadStreamingConfig handle as the HTTP server, so a POST
     // to /api/v1/streaming-config is observable by the next query
@@ -434,7 +446,8 @@ async fn main() -> Result<()> {
         // drives the Phase 6 archive failover via
         // EngineError::CapabilityMiss when the ASAP tier is empty
         // / ghost / unknown.
-        .with_sketch_index(sketch_index.clone());
+        .with_sketch_index(sketch_index.clone())
+        .with_hot_reload_backend_plan(hot_reload_backend_plan.clone());
         if let Some(control_plane_endpoint) = args.control_plane_endpoint.as_ref() {
             info!(
                 "Capability-miss notifications enabled → {}",
@@ -712,18 +725,9 @@ async fn main() -> Result<()> {
     // `SchemaRegistry`. `POST /api/v1/streaming-config` drives
     // lifecycle transitions at the sid level via the shared
     // `SketchStore` (already passed in below).
-    // BackendPlan wire format (design-backend-plan-wire-format.md):
-    // install an empty hot-reload handle so `GET/POST
-    // /api/v1/backend-plan` don't 503 before the control plane's first
-    // push lands — same "install empty, let the first push fill it in"
-    // pattern as `bootstrap_routing` below.
-    let hot_reload_backend_plan = data_plane::storage_engines::types::HotReloadBackendPlan::new(
-        control_plane::backend_plan::BackendPlan::default(),
-    );
-
     let mut server = HttpServer::new(http_config, engine, sketch_index.clone())
         .with_hot_reload_config(hot_reload_config.clone())
-        .with_hot_reload_backend_plan(hot_reload_backend_plan)
+        .with_hot_reload_backend_plan(hot_reload_backend_plan.clone())
         .with_probe_cache(probe_cache.clone());
 
     // Per-metric storage-backend routing table (issue #46

data_plane/src/query_engines/asap_query_engine/engine.rs

@@ -56,6 +56,14 @@ pub struct ASAPQueryEngine {
     /// the rest of the routing matrix.
     archive_engine:
         Option<Arc<dyn crate::query_engines::routing::query_engine_routing::QueryEngine>>,
+    /// BackendPlan wire format (design-backend-plan-wire-format.md). When
+    /// `Some`, `l4_lowering.rs`'s serving-time family/params lookup
+    /// prefers reading the installed plan's materializations directly
+    /// over reconstructing from `SketchStore` metadata
+    /// (`ObservedFamilyCostModel`). `None` when not wired up (unit
+    /// tests, legacy callers), which falls back to `SketchStore`
+    /// reconstruction only.
+    hot_reload_backend_plan: Option<crate::storage_engines::types::HotReloadBackendPlan>,
 }
 
 impl ASAPQueryEngine {
@@ -86,9 +94,30 @@ impl ASAPQueryEngine {
             control_plane_client: None,
             sketch_index: None,
             archive_engine: None,
+            hot_reload_backend_plan: None,
         }
     }
 
+    /// Attach a `HotReloadBackendPlan` handle so serving-time family/params
+    /// lookups prefer the control plane's installed `BackendPlan` over
+    /// `SketchStore` reconstruction (see this struct's field doc).
+    /// Without this call, lookups fall back to `SketchStore`
+    /// reconstruction unconditionally.
+    pub fn with_hot_reload_backend_plan(
+        mut self,
+        handle: crate::storage_engines::types::HotReloadBackendPlan,
+    ) -> Self {
+        self.hot_reload_backend_plan = Some(handle);
+        self
+    }
+
+    /// Snapshot of the currently installed `BackendPlan`, if a hot-reload
+    /// handle is wired up. `None` otherwise — callers fall back to the
+    /// `SketchStore`-reconstruction path.
+    fn backend_plan_snapshot(&self) -> Option<Arc<control_plane::backend_plan::BackendPlan>> {
+        self.hot_reload_backend_plan.as_ref().map(|h| h.snapshot())
+    }
+
     /// Phase-5 hybrid-stitch builder — attach an archive engine the
     /// `QueryEngine` trait adapter will dispatch to when the ASAP-tier
     /// reducer reports a coverage narrower than the requested range.
@@ -423,9 +452,10 @@ impl ASAPQueryEngine {
             // and every other "can't safely serve this way" outcome —
             // none of these are answerable via the sketch tier anymore;
             // the caller fails over to archive.
+            let backend_plan_snap = self.backend_plan_snapshot();
             let live_served_result =
                 crate::query_engines::asap_query_engine::live_serve::try_serve_from_summary_executor(
-                    idx, query, start_ms, end_ms, false,
+                    idx, query, start_ms, end_ms, false, backend_plan_snap.as_deref(),
                 );
             let result = match live_served_result {
                 Some(result) => result,
@@ -1125,13 +1155,15 @@ impl crate::query_engines::routing::query_engine_routing::QueryEngine for ASAPQu
                 // `live_serve_hll_global_count_merges_across_sids` for the
                 // former), and every other "can't safely serve this way"
                 // outcome — the caller fails over to archive.
+                let backend_plan_snap = self.backend_plan_snapshot();
                 let live_served_result =
                     crate::query_engines::asap_query_engine::live_serve::try_serve_from_summary_executor(
                         idx,
                         query,
                         t0_ms,
                         now_ms,
                         effective_is_cumulative(candidate),
+                        backend_plan_snap.as_deref(),
                     );
                 let result = match live_served_result {
                     Some(r) => r,

data_plane/src/query_engines/asap_query_engine/l4_lowering.rs

@@ -192,6 +192,25 @@ fn observed_family_for_metric(index: &SketchStore, metric: &str) -> Option<(Summ
     None
 }
 
+/// Look up what family/params `plan` says is materialized for `metric` —
+/// the `BackendPlan`-sourced sibling of [`observed_family_for_metric`].
+/// Unlike that function, no reconstruction is needed:
+/// `Materialization.kind`/`.params` already ARE the pair this needs,
+/// straight off the wire the control plane pushed. Returns the first
+/// matching materialization found (mirrors
+/// `observed_family_for_metric`'s "first sketch-typed one found"
+/// semantics); `None` when the plan has no materialization for this
+/// metric.
+fn observed_family_for_metric_from_plan(
+    plan: &control_plane::backend_plan::BackendPlan,
+    metric: &str,
+) -> Option<(SummaryKind, SummaryParams)> {
+    plan.materializations.values().find_map(|m| {
+        matches!(&m.source, control_plane::intent_algebra::Source::TimeSeries { metric: mm } if mm == metric)
+            .then(|| (m.kind.clone(), m.params.clone()))
+    })
+}
+
 /// Lower a raw PromQL query string to the `L4Node` tree
 /// `asap_sketch::exec::execute`/`SummaryExecutor` needs — the actual
 /// serving cutover (`live_serve.rs`). Returns `Err` for any shape serving
@@ -202,6 +221,7 @@ pub fn lower_promql_to_l4node(
     index: &SketchStore,
     query: &str,
     accuracy: AccuracyTarget,
+    backend_plan: Option<&control_plane::backend_plan::BackendPlan>,
 ) -> Result<Rc<L4Node>, LoweringSkip> {
     // Reuse the SAME candidate analysis `engine.rs` already runs for the
     // legacy dispatch, rather than re-deriving rate detection via a
@@ -221,12 +241,22 @@ pub fn lower_promql_to_l4node(
         .map_err(|e| LoweringSkip::ParseFailed(e.to_string()))?;
 
     // Serving time must reproduce the REAL planning decision, not
-    // independently re-derive one -- see this module's docs. `observed`
-    // is `None` when this metric has nothing registered (or only an
-    // `ExactAgg` sid, which bypasses `CostModel` entirely), in which case
-    // `ObservedFamilyCostModel` transparently falls back to the same
-    // accuracy-driven behavior as before.
-    let observed = find_metric_in_query_expr(&qe).and_then(|metric| observed_family_for_metric(index, &metric));
+    // independently re-derive one -- see this module's docs. Prefer
+    // reading it straight off an installed `BackendPlan`'s
+    // materializations when one covers this metric --
+    // `Materialization.kind`/`.params` already ARE the
+    // `(SummaryKind, SummaryParams)` pair this needs, no
+    // `AggregationConfig` reconstruction required (design-backend-plan-wire-format.md
+    // §5). Otherwise fall back to the `SketchStore`-reconstruction path
+    // (`observed_family_for_metric`), which is `None` when this metric
+    // has nothing registered (or only an `ExactAgg` sid, which bypasses
+    // `CostModel` entirely) -- `ObservedFamilyCostModel` then falls back
+    // further to the accuracy-driven default.
+    let observed = find_metric_in_query_expr(&qe).and_then(|metric| {
+        backend_plan
+            .and_then(|plan| observed_family_for_metric_from_plan(plan, &metric))
+            .or_else(|| observed_family_for_metric(index, &metric))
+    });
     let cost_model = ObservedFamilyCostModel::new(accuracy, observed);
 
     let physical = bind_query_expr_with_cost_model(&qe, &cost_model)
@@ -264,7 +294,7 @@ mod tests {
     #[test]
     fn rate_query_is_skipped_before_binding() {
         let idx = empty_index();
-        let result = lower_promql_to_l4node(&idx, "rate(http_requests_total[5m])", accuracy());
+        let result = lower_promql_to_l4node(&idx, "rate(http_requests_total[5m])", accuracy(), None);
         assert!(
             matches!(result, Err(LoweringSkip::RateShape)),
             "expected RateShape, got {result:?}"
@@ -274,7 +304,7 @@ mod tests {
     #[test]
     fn irate_query_is_skipped_before_binding() {
         let idx = empty_index();
-        let result = lower_promql_to_l4node(&idx, "irate(http_requests_total[5m])", accuracy());
+        let result = lower_promql_to_l4node(&idx, "irate(http_requests_total[5m])", accuracy(), None);
         assert!(
             matches!(result, Err(LoweringSkip::RateShape)),
             "expected RateShape, got {result:?}"
@@ -284,7 +314,7 @@ mod tests {
     #[test]
     fn unparseable_query_is_skipped() {
         let idx = empty_index();
-        let result = lower_promql_to_l4node(&idx, "this is not promql (((", accuracy());
+        let result = lower_promql_to_l4node(&idx, "this is not promql (((", accuracy(), None);
         assert!(
             matches!(result, Err(LoweringSkip::ParseFailed(_))),
             "expected ParseFailed, got {result:?}"
@@ -304,7 +334,7 @@ mod tests {
         // expression stays one opaque `Logical` blob, which this module
         // surfaces as `NotRealized`.
         let idx = empty_index();
-        let result = lower_promql_to_l4node(&idx, "http_requests_total", accuracy());
+        let result = lower_promql_to_l4node(&idx, "http_requests_total", accuracy(), None);
         assert!(
             matches!(result, Err(LoweringSkip::NotRealized)),
             "expected NotRealized, got {result:?}"
@@ -326,7 +356,7 @@ mod tests {
         // before this module started consulting the `SketchStore`.
         let idx = empty_index();
         let node =
-            lower_promql_to_l4node(&idx, "count_over_time(http_requests_total[5m])", accuracy())
+            lower_promql_to_l4node(&idx, "count_over_time(http_requests_total[5m])", accuracy(), None)
                 .expect("Frequency intent must realize via bind_query_expr/ControlPlaneCostModel");
         assert!(
             !matches!(node.expr, SummaryExpr::Logical(_)),
@@ -347,10 +377,147 @@ mod tests {
             &idx,
             "topk(5, sum by (host) (rate(http_requests_total[5m])))",
             accuracy(),
+            None,
         );
         assert!(
             matches!(result, Err(LoweringSkip::NotRealized) | Err(LoweringSkip::RateShape)),
             "expected NotRealized or RateShape (both are valid skips for this shape), got {result:?}"
         );
     }
+
+    // ── BackendPlan-sourced family lookup (design-backend-plan-wire-format.md §5) ────
+
+    mod backend_plan_cutover {
+        use super::*;
+        use crate::storage_engines::sketch_db::index::{
+            AccuracyBound, Capability, SketchInstanceMetadata, SketchKindHandle,
+        };
+        use control_plane::backend_plan::{BackendPlan, Materialization, WindowSpec};
+        use control_plane::intent_algebra::{ColumnRef, Source, WindowKind};
+        use std::collections::HashMap;
+
+        fn register_kll(idx: &SketchStore, metric: &str) {
+            let cfg = SketchConfig::Kll { k: 200 };
+            idx.register(SketchInstanceMetadata {
+                sid: 1,
+                metric_name: metric.to_string(),
+                group_by_keys: Default::default(),
+                capability: Some(Capability::QuantileApprox(SketchKindHandle::Kll)),
+                agg_kind: AggKind::Sketch {
+                    kind: SketchKindHandle::Kll,
+                    config: cfg.clone(),
+                    spatial_filter_canonical: String::new(),
+                },
+                accuracy: Some(AccuracyBound::from_config(&cfg)),
+                first_seen_unix_ms: 0,
+                retired_at_ms: None,
+                expires_at_ms: None,
+                policy_fp: asap_types::PolicyFingerprint::UNSET,
+            });
+        }
+
+        fn plan_with_ddsketch_materialization(metric: &str) -> BackendPlan {
+            let fingerprint = asap_types::PolicyFingerprint(42);
+            let mut materializations = HashMap::new();
+            materializations.insert(
+                fingerprint,
+                Materialization {
+                    fingerprint,
+                    source: Source::TimeSeries {
+                        metric: metric.to_string(),
+                    },
+                    window: WindowSpec {
+                        kind: WindowKind::Tumbling,
+                        size_ms: 60_000,
+                        slide_ms: None,
+                    },
+                    group_by: Vec::new(),
+                    rollup: Vec::new(),
+                    kind: SummaryKind::DDSketch,
+                    params: SummaryParams::DDSketch { alpha: 0.01 },
+                    col: ColumnRef::SampleValue,
+                    retention: None,
+                },
+            );
+            BackendPlan {
+                plan_id: 1,
+                generated_at_unix_ms: 0,
+                materializations,
+                routing: Vec::new(),
+                monitors: Vec::new(),
+            }
+        }
+
+        /// Extract the bound `(SummaryKind, SummaryParams)` from the
+        /// `SummaryEstimate { summary_input: L4Node { expr: SummaryAgg {
+        /// summary, params, .. }, .. }, .. }` shape a bare
+        /// `quantile_over_time` query lowers to (confirmed by inspecting
+        /// the tree directly).
+        fn bound_family(node: &L4Node) -> (SummaryKind, SummaryParams) {
+            match &node.expr {
+                SummaryExpr::SummaryEstimate { summary_input, .. } => match &summary_input.expr {
+                    SummaryExpr::SummaryAgg { summary, params, .. } => {
+                        (summary.clone(), params.clone())
+                    }
+                    other => panic!("expected SummaryAgg, got {other:?}"),
+                },
+                other => panic!("expected SummaryEstimate, got {other:?}"),
+            }
+        }
+
+        #[test]
+        fn without_a_plan_sketchstore_reconstruction_wins() {
+            // Baseline: no `BackendPlan` -- `observed_family_for_metric`'s
+            // SketchStore reconstruction is the only source.
+            let idx = SketchStore::new();
+            register_kll(&idx, "m");
+            let node = lower_promql_to_l4node(&idx, "quantile_over_time(0.99, m[1m])", accuracy(), None)
+                .expect("should lower");
+            assert_eq!(bound_family(&node).0, SummaryKind::Kll);
+        }
+
+        #[test]
+        fn a_plan_materialization_wins_over_sketchstore_reconstruction() {
+            // `SketchStore` has Kll registered for `m` (what
+            // reconstruction alone would find), but the installed
+            // `BackendPlan` says DDSketch for
+            // the SAME metric. The plan must win -- serving time reads
+            // planning's real (plan-sourced) decision, not whatever
+            // `SketchStore` metadata happens to reconstruct to.
+            let idx = SketchStore::new();
+            register_kll(&idx, "m");
+            let plan = plan_with_ddsketch_materialization("m");
+            let node = lower_promql_to_l4node(
+                &idx,
+                "quantile_over_time(0.99, m[1m])",
+                accuracy(),
+                Some(&plan),
+            )
+            .expect("should lower");
+            assert_eq!(
+                bound_family(&node).0,
+                SummaryKind::DDSketch,
+                "BackendPlan's materialization must take priority over SketchStore reconstruction"
+            );
+        }
+
+        #[test]
+        fn plan_present_but_no_materialization_for_metric_falls_back_to_sketchstore() {
+            // The plan is installed but doesn't cover THIS metric --
+            // `observed_family_for_metric_from_plan` returns `None` for
+            // it, so the lookup must fall through to SketchStore
+            // reconstruction, not silently fail to observe anything.
+            let idx = SketchStore::new();
+            register_kll(&idx, "m");
+            let plan = plan_with_ddsketch_materialization("some_other_metric");
+            let node = lower_promql_to_l4node(
+                &idx,
+                "quantile_over_time(0.99, m[1m])",
+                accuracy(),
+                Some(&plan),
+            )
+            .expect("should lower");
+            assert_eq!(bound_family(&node).0, SummaryKind::Kll);
+        }
+    }
 }