From 6312a14c15bcb94ba1d4d5c108a736e0a5a99aa6 Mon Sep 17 00:00:00 2001 From: zz_y Date: Wed, 29 Jul 2026 23:00:43 -0600 Subject: [PATCH] feat(backend-plan): serving-time cutover -- prefer plan over SketchStore reconstruction (Phase 4) Completes the BackendPlan wire-format cutover (control_plane/docs/design-backend-plan-wire-format.md). Serving time now reads planning's real family/params decision directly off an installed BackendPlan when one covers the query's metric, instead of always reconstructing it from SketchStore metadata. - l4_lowering.rs: new observed_family_for_metric_from_plan() reads Materialization.kind/.params directly (no AggregationConfig reconstruction needed -- they already are the (SummaryKind, SummaryParams) pair this needs). lower_promql_to_l4node() now takes an Option<&BackendPlan> and tries the plan first, falling back to the existing SketchStore-reconstruction path (observed_family_for_metric / ObservedFamilyCostModel) when no plan is installed or it doesn't cover the metric. Three new unit tests prove: (1) SketchStore reconstruction still works with no plan, (2) a plan materialization WINS over a disagreeing SketchStore registration for the same metric, (3) a plan that doesn't cover the metric falls through to SketchStore reconstruction rather than silently failing to observe anything. - l4_readout.rs / live_serve.rs: thread the same Option<&BackendPlan> parameter through unchanged otherwise. - engine.rs: ASAPQueryEngine gains a HotReloadBackendPlan handle (with_hot_reload_backend_plan) and a backend_plan_snapshot() helper; both live-serving call sites (range + instant query) pass the current snapshot through. - main.rs: the HotReloadBackendPlan handle is now shared between the query engine and the HTTP server (previously only the latter), so a POST /api/v1/backend-plan is observable by the next query, same sharing contract as hot_reload_config. - cost_model.rs: ObservedFamilyCostModel's doc updated to reflect its new fallback status. Co-Authored-By: Claude Sonnet 5 --- .../src/sketch_algebra/cost_model.rs | 9 + data_plane/src/main.rs | 26 ++- .../query_engines/asap_query_engine/engine.rs | 34 +++- .../asap_query_engine/l4_lowering.rs | 189 +++++++++++++++++- .../asap_query_engine/l4_readout.rs | 8 +- .../asap_query_engine/live_serve.rs | 19 +- 6 files changed, 255 insertions(+), 30 deletions(-) diff --git a/control_plane/src/sketch_algebra/cost_model.rs b/control_plane/src/sketch_algebra/cost_model.rs index 9fbd8955..26258f81 100644 --- a/control_plane/src/sketch_algebra/cost_model.rs +++ b/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)>, diff --git a/data_plane/src/main.rs b/data_plane/src/main.rs index 375dd277..18291a1f 100644 --- a/data_plane/src/main.rs +++ b/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 diff --git a/data_plane/src/query_engines/asap_query_engine/engine.rs b/data_plane/src/query_engines/asap_query_engine/engine.rs index c0a3cc5e..f33c215d 100644 --- a/data_plane/src/query_engines/asap_query_engine/engine.rs +++ b/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>, + /// 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, } 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> { + 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,6 +1155,7 @@ 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, @@ -1132,6 +1163,7 @@ impl crate::query_engines::routing::query_engine_routing::QueryEngine for ASAPQu t0_ms, now_ms, effective_is_cumulative(candidate), + backend_plan_snap.as_deref(), ); let result = match live_served_result { Some(r) => r, diff --git a/data_plane/src/query_engines/asap_query_engine/l4_lowering.rs b/data_plane/src/query_engines/asap_query_engine/l4_lowering.rs index 08371d1d..1898fa11 100644 --- a/data_plane/src/query_engines/asap_query_engine/l4_lowering.rs +++ b/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, 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); + } + } } diff --git a/data_plane/src/query_engines/asap_query_engine/l4_readout.rs b/data_plane/src/query_engines/asap_query_engine/l4_readout.rs index da4b26ca..15fb3b1c 100644 --- a/data_plane/src/query_engines/asap_query_engine/l4_readout.rs +++ b/data_plane/src/query_engines/asap_query_engine/l4_readout.rs @@ -69,8 +69,9 @@ pub fn execute_l4_readout( t1_ms: u64, is_cumulative: bool, accuracy: AccuracyTarget, + backend_plan: Option<&control_plane::backend_plan::BackendPlan>, ) -> Result { - let node = lower_promql_to_l4node(index, query, accuracy)?; + let node = lower_promql_to_l4node(index, query, accuracy, backend_plan)?; let ctx = QueryExecutionContext { index, @@ -253,6 +254,7 @@ mod tests { 2_000, true, accuracy(), + None, ) .expect("should execute"); assert_eq!(outcome.series.len(), 1); @@ -277,7 +279,7 @@ mod tests { register_hll(&idx, 1, "svc-a", &["a", "b", "c"]); register_hll(&idx, 2, "svc-b", &["d", "e", "f"]); let outcome = - execute_l4_readout(&idx, "count(unique_users)", 1_000, 2_000, true, accuracy()) + execute_l4_readout(&idx, "count(unique_users)", 1_000, 2_000, true, accuracy(), None) .expect("should execute"); assert_eq!( outcome.series.len(), @@ -322,7 +324,7 @@ mod tests { (1_000, 2_000), Box::new(crate::precompute_engine::operators::SumAccumulator::with_sum(42.0)), ); - let outcome = execute_l4_readout(&idx, "sum(bytes_total)", 1_000, 2_000, true, accuracy()) + let outcome = execute_l4_readout(&idx, "sum(bytes_total)", 1_000, 2_000, true, accuracy(), None) .expect("should execute"); // Window-end-only coverage: a single window (1_000, 2_000) is // keyed by its end (2_000) alone, so both bounds equal 2_000 -- diff --git a/data_plane/src/query_engines/asap_query_engine/live_serve.rs b/data_plane/src/query_engines/asap_query_engine/live_serve.rs index 1bbf0489..360358d6 100644 --- a/data_plane/src/query_engines/asap_query_engine/live_serve.rs +++ b/data_plane/src/query_engines/asap_query_engine/live_serve.rs @@ -73,13 +73,21 @@ pub fn try_serve_from_summary_executor( t0_ms: u64, t1_ms: u64, is_cumulative: bool, + backend_plan: Option<&control_plane::backend_plan::BackendPlan>, ) -> Option { if !summary_executor_live_enabled() { return None; } - let outcome = match execute_l4_readout(index, query, t0_ms, t1_ms, is_cumulative, LIVE_ACCURACY) - { + let outcome = match execute_l4_readout( + index, + query, + t0_ms, + t1_ms, + is_cumulative, + LIVE_ACCURACY, + backend_plan, + ) { Ok(outcome) => outcome, Err(skip) => { tracing::debug!( @@ -227,6 +235,7 @@ mod tests { 1_000, 2_000, true, + None, ); assert!(result.is_none(), "flag explicitly off must never serve"); } @@ -244,6 +253,7 @@ mod tests { 1_000, 2_000, true, + None, ); assert!(result.is_some(), "unset flag must default to serving"); } @@ -258,6 +268,7 @@ mod tests { 1_000, 2_000, true, + None, ); let result = result.expect("unambiguous single-series quantile must serve"); assert_eq!(result.series.len(), 1); @@ -278,7 +289,7 @@ mod tests { register_hll(&idx, 1, "svc-a", &["a", "b", "c"]); register_hll(&idx, 2, "svc-b", &["d", "e", "f"]); let result = - try_serve_from_summary_executor(&idx, "count(unique_users)", 1_000, 2_000, true); + try_serve_from_summary_executor(&idx, "count(unique_users)", 1_000, 2_000, true, None); let result = result.expect( "global-merge shape is no longer ambiguous -- it must be served, not declined", ); @@ -301,7 +312,7 @@ mod tests { let _guard = set_live_env("1"); let idx = SketchStore::new(); let result = - try_serve_from_summary_executor(&idx, "rate(http_requests_total[5m])", 0, 1000, true); + try_serve_from_summary_executor(&idx, "rate(http_requests_total[5m])", 0, 1000, true, None); assert!(result.is_none()); } }