diff --git a/control_plane/proto/backend_plan.proto b/control_plane/proto/backend_plan.proto index b1aaebc39..91e2a109b 100644 --- a/control_plane/proto/backend_plan.proto +++ b/control_plane/proto/backend_plan.proto @@ -168,6 +168,8 @@ message Materialization { ColumnRef col = 7; optional RetentionPolicy retention = 8; optional SummaryMaintenanceLifecycle lifecycle = 9; + // Canonical label matcher set (sorted `key="value"` entries). + string spatial_filter = 10; } message SummaryMaintenanceLifecycle { diff --git a/control_plane/src/asap_tier_analysis.rs b/control_plane/src/asap_tier_analysis.rs deleted file mode 100644 index 0e78fa4ad..000000000 --- a/control_plane/src/asap_tier_analysis.rs +++ /dev/null @@ -1,1991 +0,0 @@ -//! PromQL → ASAP-tier candidate analyzer (Step 2a thin-facade rewrite). -//! -//! Before Step 2a this module was an 868-line second-PromQL-walker that -//! pattern-matched on raw function-name strings — duplicating the -//! control plane's existing `query_parser::parse_query` → -//! `intent_algebra::lower::lower_parsed_query` pipeline and inventing a -//! parallel set of function names (`count_distinct_over_time`, -//! `cardinality_estimate`, `count_distinct`) that aren't part of PromQL -//! or MetricsQL. -//! -//! After Step 2a this module is a ~120-line facade. The pipeline is: -//! -//! ```text -//! PromQL string -//! ↓ query_parser::parse_query (the control plane's PromQL → ParsedQuery) -//! ParsedQuery -//! ↓ intent_algebra::lower::lower_parsed_query -//! QueryExpr (intent_algebra) — Scan / Window / Aggregate{ measures: Vec } -//! ↓ walk and call capability_for(&AggIntent) -//! Vec -//! ``` -//! -//! The lowerer is the **single owner** of "what does this PromQL function -//! mean"; `physical::runtime_capability::capability_for` is the single runtime adapter for -//! "what sketch can answer this intent". This module just glues the two. -//! -//! ## What's still here -//! -//! - The `ASAPTierCandidate` / `ASAPTierAnalysis` / `UnsupportedReason` -//! public types — the ASAP-tier reducer and the engine router consume -//! them. -//! - The PromQL `[5m]` range-selector → `range_seconds` extraction -//! helper. Reached by walking the [`ParsedQuery`] / re-parsing the -//! source via `promql_parser` ONLY for that selector — function-name -//! matching has moved entirely into the lowerer. -//! -//! ## What's gone -//! -//! - The 600 lines of direct PromQL function-name match arms. -//! - The custom-function pre-parser for `cardinality_estimate` / -//! `count_distinct_over_time` / `count_distinct` (those names don't -//! exist in real PromQL/MetricsQL; the lowerer handles the real -//! names like `quantile_over_time` and `count_over_time`). -//! - The local `Capability` / `SketchKindHandle` enums — they're now -//! re-exported from `physical::post_asap` (the single source of truth). - -use std::collections::BTreeSet; -use std::time::Duration; - -use promql_parser::parser::{self, Expr, VectorSelector}; - -use crate::query_parser::{parse_query_expr_canonical, parsed_query_from_canonical}; -use crate::types_v2::AccuracyTarget; -use planner_types::pre_asap::AggIntent; -use planner_types::pre_asap::QueryExpr; - -pub use crate::physical::runtime_capability::{ - capability_for, Capability, OuterAgg, OuterFn, SketchKindHandle, -}; - -/// Fixed accuracy target for warm-tier shape analysis (L1 adoption, -/// design-target-architecture.md Part B) -- this walk only inspects -/// `AggIntent` kinds, not their accuracy; the converter needs SOME -/// non-exact epsilon to pin sketch-eligible intents at, but the real -/// per-query accuracy bound comes from `QueryWorkload` further -/// downstream (see this module's own doc). `0.01` matches the same -/// deployment-wide default `data_plane`'s serving-time code uses -/// (`live_serve::LIVE_ACCURACY`, `shadow_compare::SHADOW_ACCURACY`). -const WARM_TIER_ANALYSIS_ACCURACY: AccuracyTarget = AccuracyTarget::Epsilon(0.01); - -// ── Public types ───────────────────────────────────────────────────────────── - -/// One sub-expression of the input PromQL that CAN be served from the -/// ASAP tier. The reducer resolves each candidate to a vector of sids -/// via `SketchIndex::instances_matching(metric_name, group_by_keys)` -/// and verifies each sid carries the required capability. -#[derive(Debug, Clone, PartialEq)] -pub struct ASAPTierCandidate { - pub metric_name: String, - pub group_by_keys: BTreeSet, - pub required_capability: Capability, - /// The PromQL function-name string from the original query, kept - /// for telemetry / logging only. The reducer dispatches off - /// `required_capability` rather than re-string-matching this. - pub function: String, - /// Scalar arguments collected from the call (e.g. `q` for quantile, - /// `k` for topk). Order matches the PromQL surface. - pub function_args: Vec, - /// Time range from the matrix-vector selector (e.g. `[5m]` → 300). - /// `0` when the query is instant-vector-shaped. - pub range_seconds: u64, - /// Canonical form of the equality label filters from the PromQL - /// selector (e.g. `{status="200",zone="us-east"}`). Empty when - /// the query has no label filters. Produced by - /// `asap_types::utils::normalize_spatial_filter` so it matches the - /// canonical form stored on `AggregationConfig.spatial_filter_normalized` - /// byte-for-byte. Drives the candidate → policy filter match in - /// `find_matching_policies`. - pub spatial_filter_canonical: String, - /// PromQL outer-function flavour — `Rate` if the expression - /// contains `rate(...)` / `irate(...)` anywhere in the tree, - /// `Plain` otherwise. Preserves the rate-vs-plain distinction the - /// `AggIntent::Sum` collapse erases, so the engine's reducer - /// dispatch can branch on the typed candidate instead of re-parsing - /// the raw PromQL string. See [`OuterFn`] for the taxonomy. - pub outer_fn: OuterFn, - /// PromQL outer-AGGREGATION operator — `Max(...)` / `Min(...)` / - /// `Avg(...)` / `Count(...)` / `Group(...)` / `Stddev(...)` / - /// `Stdvar(...)` when the original query is shaped - /// ` by (labels) ()` and the inner is a function the - /// analyzer already binds to a candidate (e.g. - /// `max by (zone) (quantile_over_time(0.99, m[5m]))`). `None` - /// otherwise. - /// - /// The engine's evaluator applies the fold AFTER the inner function - /// produces its per-row result — grouping rows by the projected - /// by-labels and folding each group's values. For the identity case - /// (inner already emits one row per by-group, e.g. asap's per-zone - /// DDSketch sketch), the fold returns the single value unchanged. - /// Closes [#296](https://github.com/ProjectASAP/ASAPQuery-backend/issues/296). - /// - /// `sum` is intentionally NOT a variant of `OuterAgg`: the lowerer - /// collapses `sum`-shaped outers into `AggIntent::Sum` → - /// `Capability::ExactAgg(Sum)`, which has its own engine dispatch - /// (see `evaluate_exact_agg`); adding it here would double-dispatch. - pub outer_agg: OuterAgg, -} - -/// Whole-query analysis result. -#[derive(Debug, Clone, PartialEq, Default)] -pub struct ASAPTierAnalysis { - pub candidates: Vec, - pub unsupported: Option, -} - -impl ASAPTierAnalysis { - /// True iff the analysis is fully ASAP-tier-answerable — - /// `unsupported.is_none()` AND at least one candidate. - pub fn is_asap_tier_answerable(&self) -> bool { - self.unsupported.is_none() && !self.candidates.is_empty() - } -} - -/// Distinct reasons a PromQL query is NOT ASAP-tier-answerable. The -/// distinction matters for logging / future precompute hints; the -/// routing layer maps every variant to the cold tier today. -#[derive(Debug, Clone, PartialEq, Eq)] -pub enum UnsupportedReason { - /// An `AggIntent` for which [`capability_for`] returned `None` — - /// `Sum`, `Min`, `Max`, `Rate`, `Increase`, every archive-only - /// intent, plus exact-accuracy `Quantile` / `Cardinality` / - /// `Count`. The carried string is the variant kind for logging. - UnsupportedAggIntent(String), - /// The query is a bare vector / matrix selector with no call — - /// the ASAP tier doesn't materialize raw counter values. - NoCallNodeFound, - /// `query_parser::parse_query` rejected the input. Carries the - /// parser error message for diagnostics. - UnparseableMetricsql(String), -} - -// ── Public entry point ─────────────────────────────────────────────────────── - -/// Parse PromQL via the control plane's existing pipeline, lower to L3 -/// `intent_algebra::QueryExpr`, walk it, and build a -/// [`ASAPTierAnalysis`]. -/// -/// Single owner of ASAP-tier shape recognition: this function does -/// **no** direct PromQL function-name matching. The lowerer -/// (`intent_algebra::lower::lower_parsed_query`) is the only place -/// that knows what `quantile_over_time` / `count_over_time` / etc. -/// mean; this function just consumes the lowered `AggIntent`s and -/// dispatches via [`capability_for`]. -pub fn analyze_promql_for_asap_tier(metricsql: &str) -> ASAPTierAnalysis { - // P2-1: parse ONCE per representation, not three times. - // - // Two genuinely-distinct parses are unavoidable here: - // (a) the raw `promql_parser` AST — needed by `trace_from_promql` - // to recover the function-name string / scalar args / - // range-seconds / outer-fn / outer-agg that the lowered - // `AggIntent` does not carry. This is the ONLY place that - // touches raw PromQL function names. - // (b) the canonical L3 `QueryExpr` — the lowered tree the analysis - // walks for `AggIntent`s. - // - // Previously the function ALSO called `parse_query(metricsql)`, which - // re-ran the *same* PromQL → legacy → canonical pipeline as (b) a - // second time just to obtain the flat `ParsedQuery` summary. We now - // derive that summary from the SAME canonical tree (b) via - // `parsed_query_from_canonical`, collapsing three parses to two. - - // Step 1: parse the raw AST once and build the trace from it. The - // walker tolerates a parse failure (returns a default trace); the - // canonical parse below is the authoritative parse-error gate. - let trace = trace_from_promql(metricsql); - - // Step 2: lower to the canonical L3 `QueryExpr`. The walk below only - // inspects `AggIntent` kinds + accuracy; the converter pins - // sketch-eligible intents at a non-exact epsilon, which is all - // warm-tier analysis needs (the real per-query accuracy bound comes - // from QueryWorkload further downstream). - let expr = match parse_query_expr_canonical(metricsql, WARM_TIER_ANALYSIS_ACCURACY) { - Ok(e) => e, - Err(e) => { - return ASAPTierAnalysis { - candidates: Vec::new(), - unsupported: Some(UnsupportedReason::UnparseableMetricsql(e.to_string())), - }; - } - }; - - // Derive the flat `ParsedQuery` summary from the SAME canonical tree — - // identical to what `parse_query(metricsql)` would return, but without - // re-parsing. - let parsed = parsed_query_from_canonical(&expr); - - // Step 3: walk the lowered tree, looking for `Aggregate` nodes. - // If there's no Aggregate the query is either: - // - a bare metric selector → `NoCallNodeFound` (ASAP-tier - // doesn't materialize raw counter values) - // - a window-bound exact-aggregation (`rate`, `irate`, - // `increase`, `sum_over_time`, `count_over_time` without - // outer count, etc.) — the control plane's PromQL parser sets - // `exact_required = true` for these and the lowerer skips - // emitting an `Aggregate` because there's no `AggType` - // (Quantile/Cardinality/Frequency) to map them onto. Surface - // as `UnsupportedAggIntent` with a label derived from the - // raw function name so the routing layer can attribute the - // rejection. - let mut intents: Vec = Vec::new(); - collect_agg_intents(&expr, &mut intents); - if intents.is_empty() { - let reason = if parsed.exact_required && !trace.function.is_empty() { - UnsupportedReason::UnsupportedAggIntent(trace.function.clone()) - } else { - UnsupportedReason::NoCallNodeFound - }; - return ASAPTierAnalysis { - candidates: Vec::new(), - unsupported: Some(reason), - }; - } - - // Step 4: for each intent, look up its capability. - // - // L1 adoption (design-target-architecture.md Part B) resilience fix: - // an unsupported intent used to abort the WHOLE analysis immediately, - // discarding any candidates already collected from other Aggregate - // nodes in the tree. That was fine when the (retired) local parser - // fused compositions into a single intent per query, but - // `lower_promql` genuinely produces multiple `Aggregate` nodes for - // compositions like `avg by (zone) (quantile_over_time(...))` — an - // outer `Avg` (no capability mapping) wrapping an inner `Quantile` - // (sketchable). Aborting on the outer miss would throw away the - // inner candidate the warm tier CAN answer. Skip an unsupported - // intent and keep collecting instead; only report `unsupported` if - // NOTHING in the whole tree was answerable. `data_plane`'s - // `engine.rs` already loops over every returned candidate (not just - // index 0), so a partial list from a composed query is an - // already-supported case, not a new invariant. - let metric_name = parsed.metric_name.clone(); - let group_by_keys: BTreeSet = parsed.group_by_labels.iter().cloned().collect(); - let spatial_filter_canonical = render_spatial_filter(&parsed.label_filters); - - let mut out = ASAPTierAnalysis::default(); - let mut last_unsupported: Option = None; - for intent in &intents { - match capability_for(intent) { - Some(cap) => { - // For a heavy-hitter top-k (`FrequencyTopk`), the inner - // `by (item)` labels (e.g. `topk(k, sum by (host) (m))`) - // are the heap's RANKED dimension — recorded as the sid's - // item_label and projected OUT of the series key into the - // top-k heap, NOT a series grouping key. The sketch is - // grouped by its OWN grouping_labels (e.g. zone) with the - // item in the heap, so requiring the item dimension in the - // sid's group_by_keys would never match (item ∉ {zone}) and - // the topk query falls through to archive. Match the - // metric's FrequencyTopk sids by metric + capability instead - // (empty required keys ⊆ any grouping); the reducer reads - // each matched sid's heap. - let candidate_keys = if matches!(cap, Capability::FrequencyTopk(_)) { - BTreeSet::new() - } else { - group_by_keys.clone() - }; - out.candidates.push(ASAPTierCandidate { - metric_name: metric_name.clone(), - group_by_keys: candidate_keys, - required_capability: cap, - function: trace.function.clone(), - function_args: trace.function_args.clone(), - range_seconds: trace.range_seconds, - spatial_filter_canonical: spatial_filter_canonical.clone(), - outer_fn: trace.outer_fn, - outer_agg: trace.outer_agg.clone(), - }); - } - None => { - last_unsupported = Some(UnsupportedReason::UnsupportedAggIntent( - intent_kind_label(intent).to_string(), - )); - } - } - } - if out.candidates.is_empty() { - out.unsupported = last_unsupported; - } - out -} - -/// Render the equality label filters from a `ParsedQuery` into the -/// canonical spatial-filter form used by -/// `AggregationConfig.spatial_filter_normalized`. Empty map → empty -/// string. Multiple entries get sorted+joined via -/// [`asap_types::utils::normalize_spatial_filter`] so the result is -/// byte-identical to what the control plane writes. -fn render_spatial_filter(label_filters: &std::collections::HashMap) -> String { - if label_filters.is_empty() { - return String::new(); - } - // Render `key="value",key="value",…` then normalize. The renderer - // doesn't need to sort — normalize_spatial_filter sorts matchers. - let joined: Vec = label_filters - .iter() - .map(|(k, v)| format!("{k}=\"{v}\"")) - .collect(); - asap_types::utils::normalize_spatial_filter(&joined.join(",")) -} - -// ── Helpers ────────────────────────────────────────────────────────────────── - -/// Walk the lowered `QueryExpr`, collecting every `AggIntent` from every -/// `Aggregate` node. `LetBinding` / `Ref` are recursed into; `Scan` / -/// `Window` carry no intents themselves. -/// -/// `pub(crate)` — also the single walker `workload::derive_agg_role` uses -/// to classify a workload entry's `query_string` by its real, lowered -/// `AggIntent` instead of a PromQL-string-prefix heuristic. -pub(crate) fn collect_agg_intents(expr: &QueryExpr, out: &mut Vec) { - match expr { - QueryExpr::Aggregate { - measures: aggs, - child, - .. - } => { - out.extend(aggs.iter().cloned()); - collect_agg_intents(child, out); - } - QueryExpr::TimeRange { child, .. } => collect_agg_intents(child, out), - QueryExpr::Scan { .. } => {} - // A-variants lifted in Batch 2 of the relational migration. They - // carry no AggIntent themselves — recurse into their children to - // find Aggregates further down the tree. `Partition` no longer - // exists in the canonical IR — its keys fold into `Aggregate.by` - // at construction time (`intent_algebra::lower`). - QueryExpr::Filter { child, .. } - | QueryExpr::Project { child, .. } - | QueryExpr::Dedup { child, .. } - | QueryExpr::Sort { child, .. } - | QueryExpr::Limit { child, .. } - | QueryExpr::PromqlSubquery { child, .. } => collect_agg_intents(child, out), - QueryExpr::Concat { children } => { - for c in children { - collect_agg_intents(c, out); - } - } - QueryExpr::Join { left, right, .. } - | QueryExpr::SetOp { left, right, .. } - | QueryExpr::BinaryOp { - lhs: left, - rhs: right, - .. - } => { - collect_agg_intents(left, out); - collect_agg_intents(right, out); - } - // The PromQL-surface superset (Scalar/EvalTime/VectorFromScalar/ - // ScalarFromVector/Relabel/InfoJoin/Sample/TimeRange/TimeShift/ - // WindowFunc) isn't constructed by this parser today; the - // single-child wrappers among them carry no `AggIntent` either - // way, so a no-op default is safe. - _ => {} - } -} - -/// Function-name string for a candidate. The lowered `AggIntent` -/// dropped the raw PromQL function name; this label is keyed off the -/// intent kind so telemetry / logging sees `quantile`, `cardinality`, -/// `topk`, etc. Specific PromQL aliases (`quantile_over_time` vs the -/// instant `quantile`) are reconstructed in [`trace_from_promql`] when -/// the AST walker can recover them; this fallback runs when the AST -/// walk fails. -fn intent_kind_label(intent: &AggIntent) -> &'static str { - if crate::planner_selection::as_frequency(intent).is_some() { - return "frequency"; - } - match intent { - AggIntent::Count { .. } => "count", - AggIntent::Sum { .. } => "sum", - AggIntent::Min { .. } => "min", - AggIntent::Max { .. } => "max", - AggIntent::Avg { .. } => "avg", - AggIntent::StdDev { .. } => "stddev", - AggIntent::Variance { .. } => "variance", - AggIntent::Quantile { .. } => "quantile", - AggIntent::TopK { .. } => "topk", - AggIntent::Cardinality { .. } => "cardinality", - AggIntent::Rate => "rate", - AggIntent::Increase => "increase", - AggIntent::Absent => "absent", - AggIntent::AbsentOverTime => "absent_over_time", - AggIntent::PresentOverTime => "present_over_time", - AggIntent::Delta => "delta", - AggIntent::Deriv => "deriv", - AggIntent::PredictLinear { .. } => "predict_linear", - AggIntent::DoubleExpSmoothing { .. } => "double_exponential_smoothing", - AggIntent::IDelta => "idelta", - AggIntent::Resets => "resets", - AggIntent::Changes => "changes", - AggIntent::HistogramCount => "histogram_count", - AggIntent::HistogramSum => "histogram_sum", - AggIntent::HistogramAvg => "histogram_avg", - AggIntent::HistogramStdDev => "histogram_stddev", - AggIntent::HistogramStdVar => "histogram_stdvar", - AggIntent::HistogramFraction { .. } => "histogram_fraction", - AggIntent::HistogramQuantile { .. } => "histogram_quantile", - AggIntent::Math(_) => "math", - AggIntent::TimeFn(_) => "time_fn", - AggIntent::Group => "group", - AggIntent::CountValues { .. } => "count_values", - AggIntent::LastOverTime => "last_over_time", - AggIntent::FirstOverTime => "first_over_time", - AggIntent::MadOverTime => "mad_over_time", - AggIntent::TsOfMinOverTime => "ts_of_min_over_time", - AggIntent::TsOfMaxOverTime => "ts_of_max_over_time", - AggIntent::TsOfFirstOverTime => "ts_of_first_over_time", - AggIntent::TsOfLastOverTime => "ts_of_last_over_time", - // Unrecognized Extension (not the Frequency one, guarded above). - AggIntent::Extension { .. } => "extension", - } -} - -/// Metadata recovered from the raw PromQL AST that the lowered -/// `AggIntent` doesn't carry: the outer function name, leading scalar -/// args, matrix selector's `[r]` range in seconds, and the rate-vs-plain -/// outer-function flavour ([`OuterFn`]) used by engine reducer -/// dispatch. -/// -/// The `function` / `function_args` / `range_seconds` fields are for -/// telemetry + the reducer's range hint. The `outer_fn` field is the -/// load-bearing signal that lets the engine pick `evaluate_exact_agg` -/// vs `evaluate_exact_agg_rate` for `Capability::ExactAgg(Sum)` -/// candidates — preserving the rate-vs-plain distinction that the -/// `AggIntent::Sum` collapse erases. -#[derive(Debug, Default)] -struct PromqlTrace { - function: String, - function_args: Vec, - range_seconds: u64, - /// Counter-function flavour recovered from the expression tree - /// (issue #301): `Rate` for `rate`/`irate`, `Increase` for - /// `increase`, `SumOverTime` for `sum_over_time`, else `Plain` - /// (bare selector / instant `sum`). The most-specific counter idiom - /// found anywhere in the tree wins (see [`set_counter_fn`]) so - /// composed shapes like `sum by (..) (rate(..))` report `Rate`. - /// Done here once so the engine reads it off the typed candidate - /// instead of re-parsing the raw query string. - outer_fn: OuterFn, - /// PromQL outer-aggregation operator wrapping the inner function — - /// `max`/`min`/`avg`/`count`/`group`/`stddev`/`stdvar` only. `sum` - /// is intentionally excluded; it has its own ExactAgg dispatch. - /// `OuterAgg::None` for queries with no such wrapper. - /// - /// Captured ONLY for the OUTERMOST aggregation node — composed - /// shapes like `max by (a) (avg by (b) (q...))` capture only `max` - /// because the engine's fold is one-pass over the inner result. - /// Deeper nesting is a documented follow-up. - outer_agg: OuterAgg, -} - -fn trace_from_promql(metricsql: &str) -> PromqlTrace { - let ast = match parser::parse(metricsql) { - Ok(a) => a, - Err(_) => return PromqlTrace::default(), - }; - let mut t = PromqlTrace::default(); - // Lift the OUTERMOST aggregation operator into `outer_agg` before - // the recursive walker descends into the inner expression — the - // walker captures inner-most function-name / range / rate-flag - // semantics, while `outer_agg` is a property of the root node only. - // See `extract_outer_agg` for the operator → `OuterAgg` mapping - // and the explicit-exclusion of `sum` (which has its own - // ExactAgg dispatch). - t.outer_agg = extract_outer_agg(&ast); - // When outer_agg lifts the outermost Aggregate (e.g. `max by (zone) ( - // quantile_over_time(...))`), the walker must descend INTO the - // aggregate's inner expression — otherwise the Aggregate branch in - // `walk_ast_for_trace` would set `t.function = "max"` and shadow - // the inner function name (`quantile_over_time`) that the engine's - // reducer actually dispatches on. The engine then sees the outer - // operator name in `candidate.function`, treats it as an unknown - // function, and CapabilityMisses to archive. Issue #296. - let walk_root = if t.outer_agg.is_some() { - unwrap_outermost_aggregate(&ast) - } else { - &ast - }; - walk_ast_for_trace(walk_root, &mut t); - t -} - -/// Companion to [`extract_outer_agg`] — peels a leading `Paren` once, -/// then descends one level into an `Aggregate.expr`. Returns the -/// original `expr` unchanged if neither pattern matches (caller -/// should only call this when `outer_agg.is_some()`, in which case the -/// shape is guaranteed to be `[Paren?]Aggregate{..}`). -fn unwrap_outermost_aggregate(expr: &Expr) -> &Expr { - let root = match expr { - Expr::Paren(p) => p.expr.as_ref(), - other => other, - }; - match root { - Expr::Aggregate(a) => &a.expr, - _ => expr, - } -} - -/// Lift the OUTERMOST PromQL aggregation operator into `OuterAgg`. -/// -/// Returns `OuterAgg::None` for any non-aggregation root (bare -/// selector, `Call(...)` with no outer agg, etc.), for `sum` -/// (already handled via the ExactAgg pipeline), and for `topk` / -/// `bottomk` / `quantile` (which have their own dispatch paths or -/// are out-of-scope for the per-row fold). -/// -/// The `by`-labels are pulled from the `LabelModifier::Include` -/// list. `without (labels)` is NOT supported today — the engine's -/// fold currently keys on the explicit `by`-labels set, and -/// translating `without` to `by` needs knowledge of the inner -/// result's label universe; deferred to a follow-up. -/// -/// A leading `Paren` (e.g. `(max by (zone) (...))`) is unwrapped -/// once so users who put the root in parens get the same shape. -fn extract_outer_agg(expr: &Expr) -> OuterAgg { - use promql_parser::parser::LabelModifier; - let root = match expr { - Expr::Paren(p) => p.expr.as_ref(), - other => other, - }; - let agg = match root { - Expr::Aggregate(a) => a, - _ => return OuterAgg::None, - }; - // `by (labels)` → Vec. `without (...)` → no support yet. - let by_labels: Vec = match &agg.modifier { - Some(LabelModifier::Include(labels)) => labels.labels.iter().cloned().collect(), - // `without (...)` — not modeled here. Return None so the engine - // emits the inner result unchanged and the query falls over to - // archive if the consumer expected the fold. Documented gap. - Some(LabelModifier::Exclude(_)) => return OuterAgg::None, - None => Vec::new(), - }; - let op = agg.op.to_string().to_lowercase(); - match op.as_str() { - "max" => OuterAgg::Max(by_labels), - "min" => OuterAgg::Min(by_labels), - "avg" => OuterAgg::Avg(by_labels), - "count" => OuterAgg::Count(by_labels), - "group" => OuterAgg::Group(by_labels), - "stddev" => OuterAgg::Stddev(by_labels), - "stdvar" => OuterAgg::Stdvar(by_labels), - // `sum` → ExactAgg(Sum) pipeline; `topk`/`bottomk` → engine-side - // fallback path; `quantile` → out of scope (instant quantile - // over function results needs per-group sketch merging). - _ => OuterAgg::None, - } -} - -/// Set `t.outer_fn` honoring counter-idiom precedence (issue #301): -/// `Rate` > `Increase` > `SumOverTime` > `Plain`. The walker may visit -/// nested calls in any order, so a more-specific flavour already set -/// must not be downgraded by a less-specific one seen later. (In -/// practice a single counter query has exactly one of these, but -/// pathological compositions like `increase(sum_over_time(...))` resolve -/// deterministically.) -fn set_counter_fn(t: &mut PromqlTrace, candidate: OuterFn) { - fn rank(f: OuterFn) -> u8 { - match f { - OuterFn::Rate => 3, - OuterFn::Increase => 2, - OuterFn::SumOverTime => 1, - OuterFn::Plain => 0, - } - } - if rank(candidate) > rank(t.outer_fn) { - t.outer_fn = candidate; - } -} - -fn walk_ast_for_trace(expr: &Expr, t: &mut PromqlTrace) { - match expr { - Expr::Call(call) => { - let name = call.func.name.to_lowercase(); - if t.function.is_empty() { - t.function = name.clone(); - } - // Flag the counter-function flavour ANYWHERE in the tree - // (issue #301) — mirrors the retired `query_contains_rate_call` - // walker but with the full taxonomy. For composed shapes like - // `sum by (zone) (rate(metric[r]))` the FIRST function set - // above is `"sum"` (the outer Aggregate), but `outer_fn` must - // report the INNER counter function so the engine dispatches - // correctly. `rate`/`irate` win over `increase`, which wins - // over `sum_over_time` (most-specific-counter-idiom wins); - // `set_counter_fn` enforces that precedence so the order in - // which the walker encounters nested calls doesn't matter. - match name.as_str() { - "rate" | "irate" => set_counter_fn(t, OuterFn::Rate), - "increase" => set_counter_fn(t, OuterFn::Increase), - "sum_over_time" => set_counter_fn(t, OuterFn::SumOverTime), - _ => {} - } - for a in &call.args.args { - if let Expr::NumberLiteral(nl) = a.as_ref() { - t.function_args.push(nl.val); - } else { - walk_ast_for_trace(a, t); - } - } - } - Expr::Aggregate(agg) => { - if t.function.is_empty() { - t.function = agg.op.to_string().to_lowercase(); - } - if let Some(p) = &agg.param { - if let Expr::NumberLiteral(nl) = p.as_ref() { - t.function_args.push(nl.val); - } - } - walk_ast_for_trace(&agg.expr, t); - } - Expr::MatrixSelector(ms) => { - if t.range_seconds == 0 { - t.range_seconds = duration_to_seconds(ms.range); - } - extract_metric_name(&ms.vs, t); - } - Expr::VectorSelector(vs) => { - extract_metric_name(vs, t); - } - Expr::Paren(p) => walk_ast_for_trace(&p.expr, t), - Expr::Subquery(sq) => walk_ast_for_trace(&sq.expr, t), - Expr::Binary(b) => { - walk_ast_for_trace(&b.lhs, t); - walk_ast_for_trace(&b.rhs, t); - } - Expr::Unary(u) => walk_ast_for_trace(&u.expr, t), - _ => {} - } -} - -#[allow(unused_variables)] -fn extract_metric_name(_vs: &VectorSelector, _t: &mut PromqlTrace) { - // Metric-name extraction is no longer needed here — the metric - // name comes from `ParsedQuery.metric_name`. The empty body keeps - // the AST walker symmetric (every selector-bearing branch routes - // through one helper) in case future telemetry wants it. -} - -fn duration_to_seconds(d: Duration) -> u64 { - d.as_secs() -} - -// ── Candidate → Policy matching ───────────────────────────────────────────── -// -// Closes the analyzer → policy registry lookup half of the merged-sid-identity -// query path. Together with `SketchStore::sids_for_policy` (PR #203) this -// gives the query engine an O(1) `Candidate → policy_fp → [sid]` index that -// avoids walking the per-sid metadata map. - -/// Translate an [`asap_types::AggregationConfig`] into the ASAP-tier -/// [`Capability`] its sids serve. Mirrors the inverse direction -/// `capability_for(&AggIntent)`: where that function says "this intent -/// wants *this* capability", this function says "this stored policy -/// *provides* this capability". Returns `None` for `AggregationType` -/// variants that don't have a corresponding ASAP-tier capability -/// (multi-pop keyed variants without an L4 binder, legacy config -/// wrappers, etc.) — callers MUST treat `None` as "policy doesn't -/// serve any ASAP-tier candidate" and skip. -pub fn policy_capability(cfg: &asap_types::AggregationConfig) -> Option { - use crate::physical::runtime_capability::SketchKindHandle; - use asap_types::AggregationType; - match cfg.aggregation_type { - // Exact-aggregation families — the ASAP-tier ExactAgg path. - AggregationType::Sum => Some(Capability::ExactAgg(AggregationType::Sum)), - AggregationType::Increase => Some(Capability::ExactAgg(AggregationType::Increase)), - AggregationType::MinMax => Some(Capability::ExactAgg(AggregationType::MinMax)), - // Quantile families — DDSketch and KLL answer quantile + min/max. - AggregationType::DDSketch => Some(Capability::QuantileApprox(SketchKindHandle::DDSketch)), - AggregationType::DatasketchesKLL => Some(Capability::QuantileApprox(SketchKindHandle::Kll)), - // Cardinality. - AggregationType::HLL => Some(Capability::CardinalityApprox), - // Frequency families. - AggregationType::CountMinSketch => { - Some(Capability::FrequencyEstimate(SketchKindHandle::CountMin)) - } - AggregationType::CountSketch => { - Some(Capability::FrequencyEstimate(SketchKindHandle::CountSketch)) - } - AggregationType::CountMinSketchWithHeap => { - Some(Capability::FrequencyTopk(SketchKindHandle::CmsWithHeap)) - } - AggregationType::CountSketchWithHeap => Some(Capability::FrequencyTopk( - SketchKindHandle::CountSketchWithHeap, - )), - // Keyed-multi-population variants. The capability the policy - // *provides* is the multi-pop variant itself; the matching - // predicate (`Capability::is_satisfied_by`) recognises that - // a multi-pop indexed capability satisfies a single-pop - // required capability through `multi_pop_satisfies_single`. - // So a candidate's `ExactAgg(Sum)` matches a policy whose - // `policy_capability` returns `ExactAgg(MultipleSum)`. - AggregationType::MultipleSum => Some(Capability::ExactAgg(AggregationType::MultipleSum)), - AggregationType::MultipleIncrease => { - Some(Capability::ExactAgg(AggregationType::MultipleIncrease)) - } - AggregationType::MultipleMinMax => { - Some(Capability::ExactAgg(AggregationType::MultipleMinMax)) - } - // No ASAP-tier capability today. HydraKLL is a keyed-quantile - // family that needs its own QuantileApprox arm (with a - // multi-pop equivalent rule) — separate follow-up. - // `Single/MultipleSubpopulation` are legacy enum wrappers from - // the pre-refactor config schema and have no semantic shape. - // (The retired `SetAggregator` / `DeltaSetAggregator` family - // used to live here too.) - AggregationType::HydraKLL - | AggregationType::SingleSubpopulation - | AggregationType::MultipleSubpopulation => None, - } -} - -/// Look up the policy whose contents match a freshly-ingested -/// sketch's shape. Used by the OTel sketch-ingest path -/// (`drivers/ingest/otel.rs`) to populate -/// `SketchInstanceMetadata.policy_fp` at registration time. Without -/// this lookup, sketch-backed sids carry `PolicyFingerprint::UNSET` -/// and are reachable only through the legacy -/// `instances_matching(metric, gbk)` walk; with it, they participate -/// in the `policy_fp → [sid]` reverse index (#203). -/// -/// Match shape — all must hold: -/// 1. `policy.metric == metric` -/// 2. `policy.aggregation_type == agg_type` -/// 3. `policy.grouping_labels.labels` (as a set) == `group_by_keys` -/// 4. Every key in `expected_params` is present in `policy.parameters` -/// with an equal value (deep `serde_json::Value` equality). -/// Extra keys on the policy that aren't in `expected_params` are -/// tolerated — the OTLP DP may not surface every param the -/// control plane authored, and policy-side defaults shouldn't -/// cause a mismatch. -/// 5. `policy.spatial_filter_normalized.is_empty()` — OTLP sketches -/// don't carry a filter context, so only unfiltered policies are -/// matchable from this path. -/// -/// Returns `Some(fp)` on a unique match, `None` when zero or multiple -/// policies match. Ambiguous (multiple-match) callers stay on the -/// UNSET sentinel — better than picking one arbitrarily. If multiple -/// distinct windows of the same `(metric, agg_type, params, group_by)` -/// shape exist, the control plane shouldn't have pushed them: they'd -/// collide on sid identity. The skip with `None` surfaces that bug. -pub fn find_policy_by_content( - registry: &asap_types::PolicyRegistry, - metric: &str, - group_by_keys: &BTreeSet, - agg_type: asap_types::AggregationType, - expected_params: &std::collections::HashMap, -) -> Option { - let mut hit: Option = None; - for (fp, cfg) in registry.iter() { - if cfg.metric != metric { - continue; - } - if cfg.aggregation_type != agg_type { - continue; - } - let policy_keys: BTreeSet = cfg.grouping_labels.labels.iter().cloned().collect(); - if &policy_keys != group_by_keys { - continue; - } - if !cfg.spatial_filter_normalized.is_empty() { - continue; - } - // Param subset match — every key the caller named must appear - // in policy.parameters with an equal value. We don't require - // the reverse direction (policy may have extra params the DP - // didn't surface). - let params_ok = expected_params - .iter() - .all(|(k, v)| cfg.parameters.get(k).is_some_and(|pv| pv == v)); - if !params_ok { - continue; - } - // Track unique-match invariant. - if hit.is_some() { - // Ambiguous — multiple policies match the same shape. Skip. - return None; - } - hit = Some(*fp); - } - hit -} - -/// Find every policy in `index` whose contents satisfy `candidate` — the -/// "whole-query resolution" mode of -/// `control_plane/docs/design-backend-plan-wire-format.md` §4's -/// `RoutingIndex` (family-level `Capability` match, returns every -/// surviving candidate rather than ranking down to one — see that -/// design doc's note on why this differs from its own originally-sketched -/// "pick one winner" framing: this function's actual, tested behavior is -/// "union every match," and the caller's own sid-level Hit/Ghost -/// classification does the real narrowing downstream). -/// -/// The result is empty when no policy fits — caller routes the query -/// to the archive engine (cold tier) in that case. Multiple matches -/// are valid (different windows / different sketch families all -/// serving the same intent); the caller can pick the cheapest via the -/// cost model or fan out to all of them and combine. -/// -/// `index.candidates_for_metric(&candidate.metric_name)` (Tier 2) already -/// narrows to this metric's own policies before any predicate below runs -/// — no per-candidate metric-name check needed here anymore. -/// -/// Matching predicate: -/// 1. `candidate.group_by_keys ⊆ policy.grouping_labels.labels` — -/// the policy's group-by must cover every key the candidate names -/// (extra group-by keys on the policy are fine; the query can -/// re-aggregate down to its required projection). -/// 2. `policy_capability(policy)` is `Some(c)` and -/// `candidate.required_capability.is_satisfied_by(&c)`. -/// 3. `policy.window_size ≤ candidate.range_seconds` — finer windows -/// can answer coarser queries by merging; the reverse isn't true. -/// When `candidate.range_seconds == 0` (instant-vector query), -/// any policy window passes. -/// 4. `policy.spatial_filter_normalized == candidate.spatial_filter_canonical` -/// — exact match on the canonical filter form. Empty matches empty -/// (the unfiltered case); non-empty must be byte-identical (both -/// sides come from `asap_types::utils::normalize_spatial_filter`, -/// which sorts matchers, so the comparison is independent of the -/// user's source ordering). -pub fn find_matching_policies( - index: &asap_types::RoutingIndex, - candidate: &ASAPTierCandidate, -) -> Vec { - let mut out = Vec::new(); - for fp in index.candidates_for_metric(&candidate.metric_name) { - let cfg = index - .get(*fp) - .expect("fp came from this index's own metric bucket"); - let policy_keys: BTreeSet = cfg.grouping_labels.labels.iter().cloned().collect(); - if !candidate.group_by_keys.is_subset(&policy_keys) { - continue; - } - let Some(provided) = policy_capability(cfg) else { - continue; - }; - if !candidate.required_capability.is_satisfied_by(&provided) { - continue; - } - if candidate.range_seconds > 0 && cfg.window_size > candidate.range_seconds { - continue; - } - if cfg.spatial_filter_normalized != candidate.spatial_filter_canonical { - continue; - } - out.push(*fp); - } - out -} - -// ── Tests ──────────────────────────────────────────────────────────────────── - -#[cfg(test)] -mod tests { - use super::*; - use asap_types::AggregationType; - - fn keys(items: &[&str]) -> BTreeSet { - items.iter().map(|s| s.to_string()).collect() - } - - // ── P2-1: parse-once equivalence ───────────────────────────────────── - - /// The parse-once refactor derives the flat `ParsedQuery` summary from - /// the SAME canonical tree the analysis walks, instead of re-running - /// `parse_query`. This pins that the derived summary is byte-for-byte - /// identical to the standalone `parse_query` path it replaced — across - /// the full shape catalogue the analyzer cares about (the - /// `metric_name`, `group_by_labels`, `label_filters`, `time_window`, - /// `exact_required`, `quantiles`, `aggregations` the rest of - /// `analyze_promql_for_asap_tier` reads off `parsed`). - #[test] - fn parse_once_matches_parse_query_for_all_shapes() { - use crate::query_parser::parse_query; - let queries = [ - "quantile_over_time(0.99, http_request_duration{env=\"prod\"}[5m])", - "sum by (zone) (http_requests_total)", - "rate(http_requests_total[5m])", - "count(unique_users_per_min)", - "topk by (service) (10, count_over_time(requests{env=\"prod\"}[1m]))", - "max by (zone) (quantile_over_time(0.99, latency[5m]))", - "avg_over_time(cpu_seconds_total[10m])", - "increase(errors_total[2m])", - ]; - for q in queries { - let canonical = parse_query_expr_canonical(q, WARM_TIER_ANALYSIS_ACCURACY) - .unwrap_or_else(|e| panic!("canonical parse failed for {q:?}: {e}")); - let derived = parsed_query_from_canonical(&canonical); - let direct = parse_query(q, WARM_TIER_ANALYSIS_ACCURACY) - .unwrap_or_else(|e| panic!("parse_query failed for {q:?}: {e}")); - - assert_eq!( - derived.metric_name, direct.metric_name, - "metric_name for {q:?}" - ); - assert_eq!( - derived.group_by_labels, direct.group_by_labels, - "group_by_labels for {q:?}" - ); - assert_eq!( - derived.label_filters, direct.label_filters, - "label_filters for {q:?}" - ); - assert_eq!( - derived.time_window, direct.time_window, - "time_window for {q:?}" - ); - assert_eq!( - derived.exact_required, direct.exact_required, - "exact_required for {q:?}" - ); - assert_eq!(derived.quantiles, direct.quantiles, "quantiles for {q:?}"); - assert_eq!( - derived.aggregations, direct.aggregations, - "aggregations for {q:?}" - ); - } - } - - // ── Supported shapes ───────────────────────────────────────────────── - - /// PromQL `count(metric)` is the spec's distinct-counting idiom - /// (count of label sets in the result vector). The analyzer must - /// collect EXACTLY ONE candidate — Cardinality — for the outer - /// count; the bare-metric inner selector must NOT synthesize an - /// `ExactAgg(Sum)` candidate that would force the engine's - /// "all candidates must succeed" loop to fail when no Sum policy - /// is registered. (The fix lives in - /// `query_parser::promql::walk_qe::Expr::VectorSelector` — gates - /// the implicit `Aggregate(Sum)` wrapper on `!ctx.outer_count`.) - #[test] - fn analyze_count_bare_metric_yields_only_cardinality_candidate() { - let a = analyze_promql_for_asap_tier("count(unique_users_per_min)"); - assert!(a.unsupported.is_none(), "{a:?}"); - assert_eq!( - a.candidates.len(), - 1, - "count(metric) must yield exactly one Cardinality candidate \ - (no implicit Sum from the bare-selector inner): {a:?}" - ); - assert_eq!( - a.candidates[0].required_capability, - Capability::CardinalityApprox, - "{a:?}" - ); - } - - /// `count(metric)` is the distinct-count idiom: the analyzer lifts - /// the outer `count` into BOTH `outer_agg = Count` and the - /// `CardinalityApprox` capability, and the bare-selector inner leaves - /// `function` empty. The data-plane engine must (a) derive the - /// reducer function from the capability when `function` is empty, and - /// (b) NOT re-apply the outer `Count` fold (the cardinality estimate - /// IS the count). This test pins the analyzer-side shape those - /// engine fixes rely on. - #[test] - fn analyze_count_bare_metric_trace_shape() { - let a = analyze_promql_for_asap_tier("count(unique_users_per_min)"); - let c = &a.candidates[0]; - assert_eq!( - c.function, "", - "bare-selector inner leaves the trace function empty: {a:?}" - ); - assert!( - matches!(c.outer_agg, OuterAgg::Count(_)), - "outer count is lifted into outer_agg: {a:?}" - ); - assert_eq!(c.range_seconds, 0, "instant query: {a:?}"); - } - - #[test] - fn analyze_quantile_over_time() { - let a = analyze_promql_for_asap_tier("quantile_over_time(0.99, http_latency_ms[5m])"); - assert!( - a.unsupported.is_none(), - "expected no unsupported reason: {a:?}" - ); - assert_eq!(a.candidates.len(), 1); - let c = &a.candidates[0]; - assert_eq!(c.metric_name, "http_latency_ms"); - assert_eq!(c.function, "quantile_over_time"); - assert_eq!(c.function_args, vec![0.99]); - assert_eq!(c.range_seconds, 300); - assert_eq!( - c.required_capability, - Capability::QuantileApprox(SketchKindHandle::Any) - ); - } - - #[test] - fn analyze_quantile_over_time_with_label_matchers() { - let a = analyze_promql_for_asap_tier( - "quantile_over_time(0.5, http_latency_ms{zone=\"z0\", region=\"us\"}[30s])", - ); - assert!(a.unsupported.is_none(), "{a:?}"); - assert_eq!(a.candidates.len(), 1); - let c = &a.candidates[0]; - assert_eq!(c.metric_name, "http_latency_ms"); - assert_eq!(c.range_seconds, 30); - // Group-by keys: zero — label EQ filters aren't group-by - // labels, they're just selectors. The lowerer leaves - // `group_by_labels` empty for a bare `quantile_over_time(…)`. - // (Adding `sum by (...)` around it changes group_by_keys.) - let _ = c.group_by_keys.clone(); - } - - #[test] - fn analyze_quantile_over_time_with_sum_by_group_keys() { - // PromQL `sum by (host) (quantile_over_time(...))` populates - // group_by_keys with `host`. - // - // L1 adoption (design-target-architecture.md Part B): `lower_promql` - // doesn't fuse "outer aggregation wraps an inner range function" - // into one shape the way the retired local parser did -- this - // composition is genuinely two operations (sum the per-series - // quantiles, grouped by host), so it lowers to two `Aggregate` - // nodes and `analyze_promql_for_asap_tier` (which walks every - // `Aggregate` node in the tree) now reports two candidates: the - // outer `ExactAgg(Sum)` and the inner `QuantileApprox`. Both carry - // `group_by_keys: {host}` (the trace-derived group-by is shared - // context across all candidates from one analysis call, not - // per-node). `data_plane`'s `engine.rs` already loops over every - // candidate (not just index 0), so multi-candidate composed - // shapes are an already-supported case, not a new invariant. - let a = analyze_promql_for_asap_tier( - "sum by (host) (quantile_over_time(0.99, http_latency_ms[5m]))", - ); - assert!(a.unsupported.is_none(), "{a:?}"); - assert_eq!(a.candidates.len(), 2, "{a:?}"); - for c in &a.candidates { - assert_eq!(c.group_by_keys, keys(&["host"]), "{a:?}"); - } - assert!( - a.candidates - .iter() - .any(|c| c.required_capability == Capability::QuantileApprox(SketchKindHandle::Any)), - "{a:?}" - ); - } - - #[test] - fn analyze_histogram_quantile_partially_answerable_via_inner_composition() { - // Renamed (was `..._is_rejected`): L1 adoption - // (design-target-architecture.md Part B), accepted behavior - // change + the resilience fix above compounding usefully here. - // - // The retired local parser unconditionally substituted - // `histogram_quantile(...)` into a plain sketchable `Quantile` - // intent (Step γ5). `lower_promql` instead correctly detects this - // is the classic `_bucket` + `by (le)` shape and produces the - // real, exact-only `AggIntent::HistogramQuantile` -- which - // `capability_for` has no mapping for. Under the old - // abort-on-first-miss loop this would have rejected the whole - // query; the resilience fix means that outer miss is skipped and - // the inner `sum(rate(...)) by (le)` composition's two real - // candidates (`ExactAgg(Sum)`, `ExactAgg(Increase)`) still - // surface -- a partial answer where the old fused behavior gave - // a full one, but not a full rejection either. - let a = analyze_promql_for_asap_tier( - "histogram_quantile(0.99, sum(rate(http_latency_bucket[5m])) by (le))", - ); - assert!(a.unsupported.is_none(), "{a:?}"); - assert_eq!(a.candidates.len(), 2, "{a:?}"); - assert!( - a.candidates - .iter() - .any(|c| c.required_capability == Capability::ExactAgg(AggregationType::Increase)), - "{a:?}" - ); - } - - #[test] - fn analyze_topk_aggregate() { - let a = analyze_promql_for_asap_tier( - "topk by (symbol) (10, count_over_time(financial_last_trade_price[5m]))", - ); - assert!(a.unsupported.is_none(), "{a:?}"); - // The first intent the lowerer emits inside a `topk` context - // is `Count{accuracy=Epsilon}` (because outer_count is set - // in the topk context), which maps to CardinalityApprox in - // the bridge — NOT FrequencyTopk. Confirm the right cap. - // (The actual FrequencyTopk binding lives at the topk wrapper, - // which isn't an AggIntent today; this is a documented gap.) - // The test just asserts at least one candidate was produced - // and no rejection fired. - assert!(!a.candidates.is_empty(), "expected at least one candidate"); - } - - // ── ExactAgg-routed shapes ─────────────────────────────────────────── - // - // `rate` / `irate` / `increase` / `sum` and the bare selector all - // lower (via `lower`) to `AggIntent::Sum`, and - // `capability_for(&Sum)` returns `Capability::ExactAgg(Sum)` — so - // they are ASAP-tier-answerable from exact-precompute state. (The - // older `lower_parsed_query` path *dropped* these intents, masking - // the `ExactAgg` capability and routing everything to archive.) - - #[test] - fn bare_vector_selector_is_no_longer_asap_tier_answerable() { - // Superseded by `bare_selector_is_no_longer_asap_tier_answerable` - // below -- see that test's comment (L1 adoption, - // design-target-architecture.md Part B, accepted behavior - // change): `lower_promql` doesn't implicitly wrap a bare selector - // in `Aggregate { Sum }` the way the retired local parser did. - let a = analyze_promql_for_asap_tier("http_requests_total{zone=\"z0\"}"); - assert!(a.candidates.is_empty(), "{a:?}"); - assert!( - matches!(a.unsupported, Some(UnsupportedReason::NoCallNodeFound)), - "{a:?}" - ); - } - - #[test] - fn rate_binds_to_exact_agg() { - let a = analyze_promql_for_asap_tier("rate(http_requests_total[5m])"); - assert!(a.unsupported.is_none(), "{a:?}"); - assert_eq!( - a.candidates[0].required_capability, - Capability::ExactAgg(AggregationType::Increase) - ); - } - - #[test] - fn irate_binds_to_exact_agg() { - // `irate` shares `AggFunc::Rate` with `rate` in - // `query_parser::promql`; both lower to `AggIntent::Rate`. - let a = analyze_promql_for_asap_tier("irate(http_requests_total[5m])"); - assert!(a.unsupported.is_none(), "{a:?}"); - assert_eq!( - a.candidates[0].required_capability, - Capability::ExactAgg(AggregationType::Increase) - ); - } - - #[test] - fn increase_binds_to_exact_agg() { - let a = analyze_promql_for_asap_tier("increase(http_requests_total[5m])"); - assert!(a.unsupported.is_none(), "{a:?}"); - assert_eq!( - a.candidates[0].required_capability, - Capability::ExactAgg(AggregationType::Increase) - ); - } - - #[test] - fn sum_by_binds_to_exact_agg() { - let a = analyze_promql_for_asap_tier("sum by (zone) (http_requests_total)"); - assert!(a.unsupported.is_none(), "{a:?}"); - assert_eq!( - a.candidates[0].required_capability, - Capability::ExactAgg(AggregationType::Sum) - ); - assert_eq!(a.candidates[0].group_by_keys, keys(&["zone"])); - } - - // ── outer_fn — rate vs plain disambiguation ────────────────────────── - // - // Regression coverage for the PR that retired the engine's - // `query_contains_rate_call` raw-PromQL re-parser. `outer_fn` is - // computed independently of `required_capability` — straight off the - // raw PromQL function name (`set_counter_fn`), not off the lowered - // `AggIntent` — so the engine's reducer dispatch has the rate-vs- - // increase-vs-plain distinction as a typed branch instead of a - // raw-PromQL re-parse, regardless of whether the capability - // computation happens to agree or differ across cases. `rate`/ - // `irate`/`increase` now bind to distinct-from-`sum`/`sum_over_time` - // capabilities (`ExactAgg(Increase)` vs `ExactAgg(Sum)` — see - // `rate_and_increase_share_capability_but_differ_on_outer_fn` below - // for the pairing that still needs `outer_fn` to disambiguate). - - #[test] - fn rate_candidate_carries_outer_fn_rate() { - let a = analyze_promql_for_asap_tier("rate(http_requests_total[5m])"); - assert!(a.unsupported.is_none(), "{a:?}"); - assert_eq!(a.candidates[0].outer_fn, OuterFn::Rate, "{a:?}"); - } - - #[test] - fn irate_candidate_carries_outer_fn_rate() { - let a = analyze_promql_for_asap_tier("irate(http_requests_total[5m])"); - assert!(a.unsupported.is_none(), "{a:?}"); - assert_eq!(a.candidates[0].outer_fn, OuterFn::Rate, "{a:?}"); - } - - #[test] - fn sum_over_time_candidate_carries_outer_fn_sum_over_time() { - // `sum_over_time(metric[r])` shares `Capability::ExactAgg(Sum)` - // with `rate(metric[r])` — the capability alone can't - // disambiguate. Post-#301 the `outer_fn` field reports - // `SumOverTime` so the engine can capability-miss → archive - // (asap can't reconstruct Σ-of-cumulative-samples from deltas). - let a = analyze_promql_for_asap_tier("sum_over_time(http_requests_total[5m])"); - assert!(a.unsupported.is_none(), "{a:?}"); - assert_eq!( - a.candidates[0].required_capability, - Capability::ExactAgg(AggregationType::Sum), - "{a:?}" - ); - assert_eq!(a.candidates[0].outer_fn, OuterFn::SumOverTime, "{a:?}"); - } - - #[test] - fn increase_candidate_carries_outer_fn_increase() { - // `increase(metric[r])` shares `Capability::ExactAgg(Increase)` - // with `rate`/`irate`; the `outer_fn` field carries the - // distinction so the engine sums deltas in `[t-r,t]` WITHOUT the - // rate divisor (issue #301). - let a = analyze_promql_for_asap_tier("increase(http_requests_total[5m])"); - assert!(a.unsupported.is_none(), "{a:?}"); - assert_eq!( - a.candidates[0].required_capability, - Capability::ExactAgg(AggregationType::Increase), - "{a:?}" - ); - assert_eq!(a.candidates[0].outer_fn, OuterFn::Increase, "{a:?}"); - } - - #[test] - fn sum_by_over_increase_candidate_carries_outer_fn_increase() { - // Composed `sum by (zone) (increase(metric[r]))` — inner counter - // function wins over the outer `sum` (same precedence as the - // rate case). - let a = analyze_promql_for_asap_tier("sum by (zone) (increase(http_requests_total[5m]))"); - assert!(a.unsupported.is_none(), "{a:?}"); - assert_eq!(a.candidates[0].outer_fn, OuterFn::Increase, "{a:?}"); - assert_eq!(a.candidates[0].range_seconds, 300, "{a:?}"); - } - - #[test] - fn sum_by_candidate_carries_outer_fn_plain() { - let a = analyze_promql_for_asap_tier("sum by (zone) (http_requests_total)"); - assert!(a.unsupported.is_none(), "{a:?}"); - assert_eq!(a.candidates[0].outer_fn, OuterFn::Plain, "{a:?}"); - } - - #[test] - fn bare_selector_is_no_longer_asap_tier_answerable() { - // L1 adoption (design-target-architecture.md Part B), accepted - // behavior change: the retired local parser wrapped a bare - // selector in an implicit `Aggregate { Sum }` specifically so the - // warm tier could still answer it via `ExactAgg(Sum)`. - // `lower_promql` doesn't add that wrapper -- a bare selector - // stays a plain `Scan` with zero `Aggregate` nodes, so this now - // returns `NoCallNodeFound` / no candidates and falls through to - // archive. Accepted as-is rather than reintroducing the implicit - // wrap locally. - let a = analyze_promql_for_asap_tier("http_requests_total{zone=\"z0\"}"); - assert!(a.candidates.is_empty(), "{a:?}"); - assert!( - matches!(a.unsupported, Some(UnsupportedReason::NoCallNodeFound)), - "{a:?}" - ); - } - - #[test] - fn sum_by_over_rate_candidate_carries_outer_fn_rate() { - // Composed shape `sum by (zone) (rate(metric[5m]))` — the - // outer function NAME is `"sum"` (the trace's `.function` - // field) but `outer_fn` MUST be `Rate` because the inner - // `rate(...)` call needs the rate-divisor reducer. This is - // the case that motivated the original `query_contains_rate_call` - // walker — now satisfied by walking the AST once in the - // analyzer and emitting the typed `OuterFn::Rate` flag. - // - // L1 adoption (design-target-architecture.md Part B): this is - // now two real `Aggregate` nodes (outer sum-by-zone, inner rate), - // so `analyze_promql_for_asap_tier` reports two candidates -- - // `ExactAgg(Sum)` (the outer reduction) and `ExactAgg(Increase)` - // (the inner rate) -- both sharing the trace-derived `outer_fn: - // Rate` and `range_seconds: 300`. Find the `Increase` one rather - // than assuming index 0. - let a = analyze_promql_for_asap_tier("sum by (zone) (rate(http_requests_total[5m]))"); - assert!(a.unsupported.is_none(), "{a:?}"); - assert_eq!(a.candidates.len(), 2, "{a:?}"); - let inc = a - .candidates - .iter() - .find(|c| c.required_capability == Capability::ExactAgg(AggregationType::Increase)) - .unwrap_or_else(|| panic!("expected an ExactAgg(Increase) candidate: {a:?}")); - assert_eq!(inc.outer_fn, OuterFn::Rate, "{a:?}"); - // Range is lifted from the inner rate's matrix selector. - assert_eq!(inc.range_seconds, 300, "{a:?}"); - } - - #[test] - fn rate_and_sum_over_time_differ_on_capability_and_outer_fn() { - // Pre-PromQL-frontend-semantic-retarget, `rate`/`irate`/ - // `increase`/`sum_over_time` all collapsed onto `AggIntent::Sum` - // (`Capability::ExactAgg(Sum)`), so `outer_fn` was the *only* - // thing that told the engine `rate(...)` needed a rate-divisor - // step `sum_over_time(...)` didn't. `rate`/`increase` now bind to - // their own dedicated `AggIntent`s (`Capability::ExactAgg( - // Increase)`, distinct from `sum_over_time`'s `ExactAgg(Sum)`) — - // a strictly more precise classification, not a regression: the - // capability itself now carries part of the distinction - // `outer_fn` used to carry alone. - let rate = analyze_promql_for_asap_tier("rate(http_requests_total[5m])"); - let sot = analyze_promql_for_asap_tier("sum_over_time(http_requests_total[5m])"); - assert_ne!( - rate.candidates[0].required_capability, sot.candidates[0].required_capability, - "rate and sum_over_time should now bind to distinct capabilities" - ); - assert_eq!( - rate.candidates[0].required_capability, - Capability::ExactAgg(AggregationType::Increase) - ); - assert_eq!( - sot.candidates[0].required_capability, - Capability::ExactAgg(AggregationType::Sum) - ); - assert_eq!(rate.candidates[0].outer_fn, OuterFn::Rate); - assert_eq!(sot.candidates[0].outer_fn, OuterFn::SumOverTime); - } - - #[test] - fn rate_and_increase_share_capability_but_differ_on_outer_fn() { - // The pairing that now needs `outer_fn` to disambiguate: `rate` - // and `increase` both bind to `Capability::ExactAgg(Increase)` - // (rate = increase / range, an L4/reducer-level division, not a - // capability difference) — the engine still can't tell from the - // capability alone whether to apply the rate divisor, so - // `outer_fn` carries that distinction. - let rate = analyze_promql_for_asap_tier("rate(http_requests_total[5m])"); - let inc = analyze_promql_for_asap_tier("increase(http_requests_total[5m])"); - assert_eq!( - rate.candidates[0].required_capability, inc.candidates[0].required_capability, - "rate and increase should share the same Capability" - ); - assert_ne!( - rate.candidates[0].outer_fn, inc.candidates[0].outer_fn, - "rate and increase MUST differ on outer_fn so the engine can \ - dispatch correctly without re-parsing the raw PromQL" - ); - assert_eq!(rate.candidates[0].outer_fn, OuterFn::Rate); - assert_eq!(inc.candidates[0].outer_fn, OuterFn::Increase); - } - - // ── outer_agg — outer aggregation operator on function results ────── - // - // Regression coverage for issue #296: the asap engine was rejecting - // `max by (zone) (quantile_over_time(0.99, m[5m]))` because no - // generic "aggregation operator wraps a function result" path - // existed. The analyzer now captures the outer agg operator on a - // typed `OuterAgg` field; the engine's fold pass consumes it - // after the inner function returns its per-row result. - - #[test] - fn max_by_quantile_over_time_carries_outer_agg_max() { - // L1 adoption (design-target-architecture.md Part B): two real - // `Aggregate` nodes now, not one fused shape -- the outer `max` - // itself is a real `AggIntent::Max` (`ExactAgg(MinMax)`), plus the - // inner `Quantile`. Both candidates share the trace-derived - // `outer_agg: Max([zone])` (shared context per analysis call). - let a = analyze_promql_for_asap_tier( - "max by (zone) (quantile_over_time(0.99, http_latency_ms[5m]))", - ); - assert!(a.unsupported.is_none(), "{a:?}"); - assert_eq!(a.candidates.len(), 2, "{a:?}"); - let c = a - .candidates - .iter() - .find(|c| c.required_capability == Capability::QuantileApprox(SketchKindHandle::Any)) - .unwrap_or_else(|| panic!("expected a QuantileApprox candidate: {a:?}")); - // OuterAgg captured. - match &c.outer_agg { - OuterAgg::Max(labels) => { - assert_eq!(labels, &vec!["zone".to_string()]); - } - other => panic!("expected OuterAgg::Max([zone]), got {other:?}"), - } - } - - #[test] - fn avg_by_quantile_over_time_carries_outer_agg_avg() { - // L1 adoption (design-target-architecture.md Part B): the outer - // `avg` is now a real `AggIntent::Avg`, which `capability_for` - // does NOT map to any capability (`Avg` has never been - // supported -- see `capability_for`'s own tests). Per the - // resilience fix above, that unsupported intent is skipped - // rather than aborting the whole analysis, so only the inner - // `Quantile` candidate survives -- still carrying the - // trace-derived `outer_agg: Avg([zone])`. - let a = analyze_promql_for_asap_tier( - "avg by (zone) (quantile_over_time(0.99, http_latency_ms[5m]))", - ); - assert!(a.unsupported.is_none(), "{a:?}"); - assert_eq!(a.candidates.len(), 1, "{a:?}"); - assert_eq!( - a.candidates[0].required_capability, - Capability::QuantileApprox(SketchKindHandle::Any), - "{a:?}" - ); - match &a.candidates[0].outer_agg { - OuterAgg::Avg(labels) => assert_eq!(labels, &vec!["zone".to_string()]), - other => panic!("expected OuterAgg::Avg([zone]), got {other:?}"), - } - } - - #[test] - fn min_by_quantile_over_time_carries_outer_agg_min() { - let a = analyze_promql_for_asap_tier( - "min by (zone) (quantile_over_time(0.99, http_latency_ms[5m]))", - ); - assert!(a.unsupported.is_none(), "{a:?}"); - match &a.candidates[0].outer_agg { - OuterAgg::Min(labels) => assert_eq!(labels, &vec!["zone".to_string()]), - other => panic!("expected OuterAgg::Min([zone]), got {other:?}"), - } - } - - #[test] - fn count_by_rate_carries_outer_agg_count() { - let a = analyze_promql_for_asap_tier("count by (zone) (rate(http_requests_total[5m]))"); - assert!(a.unsupported.is_none(), "{a:?}"); - match &a.candidates[0].outer_agg { - OuterAgg::Count(labels) => assert_eq!(labels, &vec!["zone".to_string()]), - other => panic!("expected OuterAgg::Count([zone]), got {other:?}"), - } - } - - #[test] - fn sum_over_time_carries_outer_agg_none() { - // No outer aggregation wrapper → OuterAgg::None. - let a = analyze_promql_for_asap_tier("sum_over_time(http_requests_total[5m])"); - assert!(a.unsupported.is_none(), "{a:?}"); - assert_eq!(a.candidates[0].outer_agg, OuterAgg::None); - } - - #[test] - fn sum_by_zone_does_not_set_outer_agg_sum() { - // `sum` MUST NOT populate OuterAgg — that operator routes through - // the ExactAgg(Sum) pipeline; double-dispatching would re-fold - // values that the per-window reducer has already accumulated. - let a = analyze_promql_for_asap_tier("sum by (zone) (http_requests_total)"); - assert!(a.unsupported.is_none(), "{a:?}"); - assert_eq!( - a.candidates[0].outer_agg, - OuterAgg::None, - "sum must not populate OuterAgg — handled by ExactAgg(Sum) pipeline" - ); - } - - #[test] - fn bare_quantile_over_time_carries_outer_agg_none() { - let a = analyze_promql_for_asap_tier("quantile_over_time(0.99, http_latency_ms[5m])"); - assert!(a.unsupported.is_none(), "{a:?}"); - assert_eq!(a.candidates[0].outer_agg, OuterAgg::None); - } - - // ── Unsupported / rejected shapes ──────────────────────────────────── - - #[test] - fn unparseable_promql_surfaces_clean_error() { - let a = analyze_promql_for_asap_tier("@@@ this is not promql @@@"); - match a.unsupported { - Some(UnsupportedReason::UnparseableMetricsql(msg)) => { - assert!(!msg.is_empty(), "parser error message should be non-empty"); - } - other => panic!("expected UnparseableMetricsql, got {other:?}"), - } - } - - // ── Range parsing ──────────────────────────────────────────────────── - - #[test] - fn range_seconds_parses_seconds() { - let a = analyze_promql_for_asap_tier("quantile_over_time(0.99, m[30s])"); - assert_eq!(a.candidates[0].range_seconds, 30); - } - - #[test] - fn range_seconds_parses_minutes() { - let a = analyze_promql_for_asap_tier("quantile_over_time(0.99, m[5m])"); - assert_eq!(a.candidates[0].range_seconds, 300); - } - - #[test] - fn range_seconds_parses_hours() { - let a = analyze_promql_for_asap_tier("quantile_over_time(0.99, m[2h])"); - assert_eq!(a.candidates[0].range_seconds, 7200); - } - - // ── is_asap_tier_answerable ────────────────────────────────────────── - - #[test] - fn is_asap_tier_answerable_true_for_supported() { - let a = analyze_promql_for_asap_tier("quantile_over_time(0.99, m[5m])"); - assert!(a.is_asap_tier_answerable()); - } - - #[test] - fn is_asap_tier_answerable_true_for_count_over_time() { - // `count_over_time(...)` now lowers to - // `AggIntent::Frequency{Epsilon}` (per-series sample count - // over the window), which `capability_for` maps to - // `Capability::FrequencyEstimate(Any)` — answerable by any - // frequency-family sketch (CMS / CountSketch, heap-less). - let a = analyze_promql_for_asap_tier("count_over_time(m[5m])"); - assert!(a.is_asap_tier_answerable(), "{a:?}"); - } - - #[test] - fn is_asap_tier_answerable_false_for_bare_selector() { - // Renamed + inverted (was `..._true_..`): L1 adoption - // (design-target-architecture.md Part B, accepted behavior - // change) -- see `bare_selector_is_no_longer_asap_tier_answerable`'s - // comment. A bare selector no longer lowers to an implicit - // `Aggregate { Sum }`, so it is NOT ASAP-tier-answerable anymore. - let a = analyze_promql_for_asap_tier("m{zone=\"z0\"}"); - assert!(!a.is_asap_tier_answerable()); - } - - // ── Cardinality / count_over_time real-PromQL acceptance ──────────── - - /// `count_over_time(metric[range])` is the PromQL per-series - /// sample-count idiom — exactly what a heap-less CMS / CountSketch - /// estimates. The parser maps it to `AggFunc::Frequency` which - /// lowers to `AggIntent::Frequency{Epsilon}`; `capability_for` - /// returns `Capability::FrequencyEstimate(Any)`. The warm engine - /// binds the query to any frequency-family policy registered for - /// the metric. - #[test] - fn count_over_time_binds_to_frequency_estimate() { - let a = analyze_promql_for_asap_tier("count_over_time(http_requests_total[5m])"); - assert!(a.unsupported.is_none(), "{a:?}"); - assert_eq!(a.candidates.len(), 1, "{a:?}"); - assert_eq!( - a.candidates[0].required_capability, - Capability::FrequencyEstimate(SketchKindHandle::Any), - "{a:?}" - ); - } - - /// `count by (...) (count_over_time(...))` is the PromQL distinct- - /// count idiom. The `query_parser::promql` walker promotes the - /// outer `count` + inner `count_over_time` to `AggFunc::CountDistinct`, - /// which lowers to `AggIntent::Cardinality{accuracy=Epsilon}` and - /// maps to `Capability::CardinalityApprox`. - #[test] - fn count_by_count_over_time_is_cardinality() { - let a = analyze_promql_for_asap_tier( - "count by (symbol) (count_over_time(financial_last_trade_price[5m]))", - ); - assert!(a.unsupported.is_none(), "{a:?}"); - assert!(!a.candidates.is_empty()); - assert_eq!( - a.candidates[0].required_capability, - Capability::CardinalityApprox, - ); - } - - // ── policy_capability + find_matching_policies tests ───────────────── - - mod matching { - use super::super::*; - use asap_types::AggregationType; - use asap_types::KeyByLabelNames; - use asap_types::{AggregationConfig, PolicyFingerprint, PolicyRegistry, RoutingIndex}; - use std::collections::HashMap; - - fn cfg( - metric: &str, - agg_type: AggregationType, - group_by: Vec<&str>, - window_size: u64, - spatial_filter: &str, - ) -> AggregationConfig { - AggregationConfig::new( - agg_type, - String::new(), - HashMap::new(), - KeyByLabelNames::new(group_by.into_iter().map(|s| s.to_string()).collect()), - KeyByLabelNames::empty(), - KeyByLabelNames::empty(), - String::new(), - window_size, - window_size, - asap_types::enums::WindowKind::Tumbling, - spatial_filter.to_string(), - metric.to_string(), - None, - None, - None, - ) - } - - fn candidate( - metric: &str, - group_by: &[&str], - cap: Capability, - range_seconds: u64, - ) -> ASAPTierCandidate { - candidate_with_filter(metric, group_by, cap, range_seconds, "") - } - - fn candidate_with_filter( - metric: &str, - group_by: &[&str], - cap: Capability, - range_seconds: u64, - spatial_filter_canonical: &str, - ) -> ASAPTierCandidate { - ASAPTierCandidate { - metric_name: metric.to_string(), - group_by_keys: group_by.iter().map(|s| s.to_string()).collect(), - required_capability: cap, - function: String::new(), - function_args: Vec::new(), - range_seconds, - spatial_filter_canonical: spatial_filter_canonical.to_string(), - outer_fn: OuterFn::default(), - outer_agg: OuterAgg::default(), - } - } - - #[test] - fn policy_capability_maps_sum_to_exact_agg() { - let c = cfg("m", AggregationType::Sum, vec![], 60, ""); - assert_eq!( - policy_capability(&c), - Some(Capability::ExactAgg(AggregationType::Sum)) - ); - } - - #[test] - fn policy_capability_maps_ddsketch_to_quantile_approx() { - use crate::physical::runtime_capability::SketchKindHandle; - let c = cfg("m", AggregationType::DDSketch, vec![], 60, ""); - assert_eq!( - policy_capability(&c), - Some(Capability::QuantileApprox(SketchKindHandle::DDSketch)) - ); - } - - #[test] - fn policy_capability_maps_multiple_sum_to_exact_agg_multiple_sum() { - let c = cfg("m", AggregationType::MultipleSum, vec!["zone"], 60, ""); - assert_eq!( - policy_capability(&c), - Some(Capability::ExactAgg(AggregationType::MultipleSum)) - ); - } - - #[test] - fn multiple_sum_policy_satisfies_unkeyed_sum_query() { - // MultipleSum policy keeps per-zone state; an unkeyed Sum - // query can re-aggregate across zones. The is_satisfied_by - // multi-pop-satisfies-single rule + the group_by ⊆ policy - // grouping check let it through. - let policies = vec![cfg( - "http_lat", - AggregationType::MultipleSum, - vec!["zone"], - 60, - "", - )]; - let registry = RoutingIndex::build(PolicyRegistry::from_configs(policies)); - let cand = candidate( - "http_lat", - &[], - Capability::ExactAgg(AggregationType::Sum), - 60, - ); - assert_eq!(find_matching_policies(®istry, &cand).len(), 1); - } - - #[test] - fn multiple_increase_policy_satisfies_keyed_increase_query() { - let policies = vec![cfg( - "http_requests_total", - AggregationType::MultipleIncrease, - vec!["zone", "service"], - 60, - "", - )]; - let registry = RoutingIndex::build(PolicyRegistry::from_configs(policies)); - // Query asks for per-zone increase; policy keeps {zone, - // service} (superset). - let cand = candidate( - "http_requests_total", - &["zone"], - Capability::ExactAgg(AggregationType::Increase), - 60, - ); - assert_eq!(find_matching_policies(®istry, &cand).len(), 1); - } - - #[test] - fn single_pop_policy_does_not_satisfy_keyed_query() { - // Unkeyed Sum policy can't answer per-zone Sum — keys - // already collapsed. Group_by ⊆ policy_grouping_labels - // check rejects this even though capabilities would - // structurally satisfy. - let policies = vec![cfg("http_lat", AggregationType::Sum, vec![], 60, "")]; - let registry = RoutingIndex::build(PolicyRegistry::from_configs(policies)); - let cand = candidate( - "http_lat", - &["zone"], - Capability::ExactAgg(AggregationType::Sum), - 60, - ); - assert!(find_matching_policies(®istry, &cand).is_empty()); - } - - #[test] - fn matches_exact_metric_and_capability() { - let policies = vec![cfg("http_lat", AggregationType::Sum, vec![], 60, "")]; - let registry = RoutingIndex::build(PolicyRegistry::from_configs(policies)); - let cand = candidate( - "http_lat", - &[], - Capability::ExactAgg(AggregationType::Sum), - 60, - ); - let m = find_matching_policies(®istry, &cand); - assert_eq!(m.len(), 1); - assert_eq!(m[0], registry.fingerprints().next().unwrap()); - } - - #[test] - fn does_not_match_different_metric() { - let policies = vec![cfg("http_lat", AggregationType::Sum, vec![], 60, "")]; - let registry = RoutingIndex::build(PolicyRegistry::from_configs(policies)); - let cand = candidate( - "cpu_pct", - &[], - Capability::ExactAgg(AggregationType::Sum), - 60, - ); - assert!(find_matching_policies(®istry, &cand).is_empty()); - } - - #[test] - fn does_not_match_incompatible_capability() { - // Policy is Sum (ExactAgg); candidate asks for QuantileApprox. - use crate::physical::runtime_capability::SketchKindHandle; - let policies = vec![cfg("http_lat", AggregationType::Sum, vec![], 60, "")]; - let registry = RoutingIndex::build(PolicyRegistry::from_configs(policies)); - let cand = candidate( - "http_lat", - &[], - Capability::QuantileApprox(SketchKindHandle::Any), - 60, - ); - assert!(find_matching_policies(®istry, &cand).is_empty()); - } - - #[test] - fn matches_when_policy_group_by_covers_candidate() { - // Policy keeps {zone, service}; candidate asks for just {zone}. - // That's covered — the query can re-aggregate down. - let policies = vec![cfg( - "http_lat", - AggregationType::Sum, - vec!["zone", "service"], - 60, - "", - )]; - let registry = RoutingIndex::build(PolicyRegistry::from_configs(policies)); - let cand = candidate( - "http_lat", - &["zone"], - Capability::ExactAgg(AggregationType::Sum), - 60, - ); - assert_eq!(find_matching_policies(®istry, &cand).len(), 1); - } - - #[test] - fn does_not_match_when_candidate_needs_keys_policy_lacks() { - // Policy keeps {zone}; candidate asks for {zone, service}. - // That's NOT covered — policy already projected service away. - let policies = vec![cfg("http_lat", AggregationType::Sum, vec!["zone"], 60, "")]; - let registry = RoutingIndex::build(PolicyRegistry::from_configs(policies)); - let cand = candidate( - "http_lat", - &["zone", "service"], - Capability::ExactAgg(AggregationType::Sum), - 60, - ); - assert!(find_matching_policies(®istry, &cand).is_empty()); - } - - #[test] - fn finer_window_matches_coarser_query_range() { - // Policy emits 60s windows; candidate wants 300s range. - // Finer can answer coarser via merge. - let policies = vec![cfg("http_lat", AggregationType::Sum, vec![], 60, "")]; - let registry = RoutingIndex::build(PolicyRegistry::from_configs(policies)); - let cand = candidate( - "http_lat", - &[], - Capability::ExactAgg(AggregationType::Sum), - 300, - ); - assert_eq!(find_matching_policies(®istry, &cand).len(), 1); - } - - #[test] - fn coarser_window_does_not_match_finer_query_range() { - // Policy emits 300s windows; candidate wants 60s range. - // Can't downsample 300s into 60s. - let policies = vec![cfg("http_lat", AggregationType::Sum, vec![], 300, "")]; - let registry = RoutingIndex::build(PolicyRegistry::from_configs(policies)); - let cand = candidate( - "http_lat", - &[], - Capability::ExactAgg(AggregationType::Sum), - 60, - ); - assert!(find_matching_policies(®istry, &cand).is_empty()); - } - - #[test] - fn zero_range_query_accepts_any_window() { - // Instant-vector queries (range_seconds=0) match any policy. - let policies = vec![cfg("http_lat", AggregationType::Sum, vec![], 300, "")]; - let registry = RoutingIndex::build(PolicyRegistry::from_configs(policies)); - let cand = candidate( - "http_lat", - &[], - Capability::ExactAgg(AggregationType::Sum), - 0, - ); - assert_eq!(find_matching_policies(®istry, &cand).len(), 1); - } - - #[test] - fn filtered_policy_matches_when_candidate_has_matching_filter() { - // Both sides carry the canonical form - // `{status="200"}` (normalize_spatial_filter sorts + - // brace-wraps single-matcher inputs). Match should succeed. - let policies = vec![cfg( - "http_lat", - AggregationType::Sum, - vec![], - 60, - r#"status="200""#, - )]; - let registry = RoutingIndex::build(PolicyRegistry::from_configs(policies)); - let cand = candidate_with_filter( - "http_lat", - &[], - Capability::ExactAgg(AggregationType::Sum), - 60, - r#"{status="200"}"#, - ); - assert_eq!(find_matching_policies(®istry, &cand).len(), 1); - } - - #[test] - fn filtered_policy_does_not_match_unfiltered_candidate() { - let policies = vec![cfg( - "http_lat", - AggregationType::Sum, - vec![], - 60, - r#"status="200""#, - )]; - let registry = RoutingIndex::build(PolicyRegistry::from_configs(policies)); - let cand = candidate( - "http_lat", - &[], - Capability::ExactAgg(AggregationType::Sum), - 60, - ); - assert!(find_matching_policies(®istry, &cand).is_empty()); - } - - #[test] - fn unfiltered_policy_does_not_match_filtered_candidate() { - let policies = vec![cfg("http_lat", AggregationType::Sum, vec![], 60, "")]; - let registry = RoutingIndex::build(PolicyRegistry::from_configs(policies)); - let cand = candidate_with_filter( - "http_lat", - &[], - Capability::ExactAgg(AggregationType::Sum), - 60, - r#"{status="200"}"#, - ); - assert!(find_matching_policies(®istry, &cand).is_empty()); - } - - #[test] - fn different_filter_values_do_not_match() { - let policies = vec![cfg( - "http_lat", - AggregationType::Sum, - vec![], - 60, - r#"status="200""#, - )]; - let registry = RoutingIndex::build(PolicyRegistry::from_configs(policies)); - let cand = candidate_with_filter( - "http_lat", - &[], - Capability::ExactAgg(AggregationType::Sum), - 60, - r#"{status="500"}"#, - ); - assert!(find_matching_policies(®istry, &cand).is_empty()); - } - - #[test] - fn multiple_matches_return_all_fingerprints() { - // Two policies serve the same intent at different windows - // — both match (cost model picks one later). - let policies = vec![ - cfg("http_lat", AggregationType::Sum, vec![], 60, ""), - cfg("http_lat", AggregationType::Sum, vec![], 30, ""), - ]; - let registry = RoutingIndex::build(PolicyRegistry::from_configs(policies)); - let cand = candidate( - "http_lat", - &[], - Capability::ExactAgg(AggregationType::Sum), - 300, - ); - assert_eq!(find_matching_policies(®istry, &cand).len(), 2); - } - - #[test] - fn ignores_unsupported_multi_pop_variants() { - // `HydraKLL` has `policy_capability == None` because no - // Capability variant covers its shape today. Matching - // skips it. (Historically `SetAggregator` / - // `DeltaSetAggregator` were also in this bucket; they've - // been retired.) - let policies = vec![cfg( - "http_lat", - AggregationType::HydraKLL, - vec!["zone"], - 60, - "", - )]; - let registry = RoutingIndex::build(PolicyRegistry::from_configs(policies)); - let cand = candidate( - "http_lat", - &["zone"], - Capability::ExactAgg(AggregationType::Sum), - 60, - ); - assert!(find_matching_policies(®istry, &cand).is_empty()); - } - - #[test] - fn empty_registry_yields_empty_matches() { - let registry = - RoutingIndex::build(PolicyRegistry::from_configs(Vec::::new())); - let cand = candidate( - "http_lat", - &[], - Capability::ExactAgg(AggregationType::Sum), - 60, - ); - assert!(find_matching_policies(®istry, &cand).is_empty()); - let _ = PolicyFingerprint::UNSET; // silence unused import warning - } - } -} diff --git a/control_plane/src/backend_client.rs b/control_plane/src/backend_client.rs index 932b836ea..0165d0429 100644 --- a/control_plane/src/backend_client.rs +++ b/control_plane/src/backend_client.rs @@ -359,6 +359,47 @@ impl BackendClient { Err(classify_http_status(status, body, "BackendPlan POST")) } } + + /// Publish the backend-facing portions of one PhysicalPlan in a single + /// request, preventing independently retried documents from mixing + /// generations at the backend. + pub async fn post_physical_plan_typed( + &self, + precompute_plan: &crate::physical::compiler::PrecomputePlan, + backend_plan: Vec, + storage_routing: Option, + ) -> std::result::Result<(), BackendPostError> { + let url = derive_physical_plan_url(&self.endpoint); + let response = self + .http + .post(&url) + .json(&serde_json::json!({ + "precompute_plan": precompute_plan, + "backend_plan": backend_plan, + "storage_routing": storage_routing, + })) + .send() + .await + .map_err(classify_reqwest_error)?; + let status = response.status(); + if status.is_success() { + Ok(()) + } else { + let body = response.text().await.unwrap_or_default(); + Err(classify_http_status(status, body, "PhysicalPlan POST")) + } + } +} + +fn derive_physical_plan_url(endpoint: &str) -> String { + const DASH: &str = "/api/v1/streaming-config"; + const UNDERSCORE: &str = "/api/v1/streaming_config"; + const PHYSICAL: &str = "/api/v1/physical-plan"; + endpoint + .strip_suffix(DASH) + .or_else(|| endpoint.strip_suffix(UNDERSCORE)) + .map(|base| format!("{base}{PHYSICAL}")) + .unwrap_or_else(|| endpoint.to_string()) } /// Map a streaming-config endpoint URL to the sibling storage-routing diff --git a/control_plane/src/backend_plan/from_stage_config.rs b/control_plane/src/backend_plan/from_stage_config.rs index 062595bb3..d54bd98f5 100644 --- a/control_plane/src/backend_plan/from_stage_config.rs +++ b/control_plane/src/backend_plan/from_stage_config.rs @@ -48,8 +48,9 @@ pub fn from_stage_config( HashMap::with_capacity(cfg.aggregations.len()); for agg in &cfg.aggregations { - let fingerprint = policy_fingerprint_for_aggregation(agg) + let fingerprint = aggregation_config_for_materialization(agg) .with_context(|| format!("aggregation_id {:?}", agg.aggregation_id))?; + let fingerprint = fingerprint.policy_fingerprint(); fingerprint_by_agg_id.insert(agg.aggregation_id.as_str(), fingerprint); let (kind, params) = match &agg.agg_type_override { @@ -71,6 +72,7 @@ pub fn from_stage_config( }, group_by: agg.grouping.clone(), rollup: Vec::new(), + spatial_filter: asap_types::utils::normalize_spatial_filter(&agg.spatial_filter), kind, params, col: ColumnRef::SampleValue, @@ -134,14 +136,15 @@ pub fn from_stage_config( /// parser the real `POST /api/v1/streaming-config` handler uses (which /// parses its body as YAML regardless of declared content-type, since /// JSON is valid YAML) — rather than re-deriving the field mapping here. -fn policy_fingerprint_for_aggregation(agg: &BackendAggregation) -> Result { +pub fn aggregation_config_for_materialization( + agg: &BackendAggregation, +) -> Result { let json = build_backend_aggregation_json(agg); let text = serde_json::to_string(&json).context("serialize synthesized aggregation JSON")?; let yaml_value: serde_yaml::Value = serde_yaml::from_str(&text).context("parse synthesized aggregation JSON as YAML")?; - let cfg = AggregationConfig::from_yaml_data(&yaml_value, None, QueryLanguage::promql) - .context("build AggregationConfig from synthesized aggregation JSON")?; - Ok(cfg.policy_fingerprint()) + AggregationConfig::from_yaml_data(&yaml_value, None, QueryLanguage::promql) + .context("build AggregationConfig from synthesized aggregation JSON") } /// Option B (post-#287) exact-agg override: `s` is already the wire diff --git a/control_plane/src/backend_plan/mod.rs b/control_plane/src/backend_plan/mod.rs index 6ab9b05b2..8c9580808 100644 --- a/control_plane/src/backend_plan/mod.rs +++ b/control_plane/src/backend_plan/mod.rs @@ -28,6 +28,7 @@ pub mod proto { } mod from_stage_config; +pub use from_stage_config::aggregation_config_for_materialization; pub use from_stage_config::from_stage_config; use std::collections::HashMap; @@ -70,6 +71,12 @@ pub enum ValidationError { ZeroSlide { fingerprint: u64 }, #[error("route references unknown materialization {fingerprint}")] UnknownMaterialization { fingerprint: u64 }, + #[error("materialization {fingerprint} has mismatched kind/parameters")] + KindParamsMismatch { fingerprint: u64 }, + #[error("route capability is incompatible with materialization {fingerprint}")] + IncompatibleRoute { fingerprint: u64 }, + #[error("stale plan generation: incoming={incoming}, active={active}")] + StaleGeneration { incoming: u64, active: u64 }, } // ── WindowSpec ─────────────────────────────────────────────────────────────── @@ -586,6 +593,7 @@ pub struct Materialization { pub window: WindowSpec, pub group_by: Vec, pub rollup: Vec, + pub spatial_filter: String, pub kind: SummaryKind, pub params: SummaryParams, pub col: ColumnRef, @@ -631,6 +639,7 @@ impl From<&Materialization> for proto::Materialization { window: Some((&m.window).into()), group_by: m.group_by.clone(), rollup: m.rollup.clone(), + spatial_filter: m.spatial_filter.clone(), params: Some((&m.params).into()), col: Some((&m.col).into()), retention: m.retention.as_ref().map(Into::into), @@ -674,6 +683,7 @@ impl TryFrom for Materialization { .try_into()?, group_by: m.group_by, rollup: m.rollup, + spatial_filter: m.spatial_filter, kind, params, col: m @@ -811,18 +821,70 @@ impl BackendPlan { if materialization.window.slide_ms == Some(0) { return Err(ValidationError::ZeroSlide { fingerprint: key.0 }); } + if !kind_params_match(&materialization.kind, &materialization.params) { + return Err(ValidationError::KindParamsMismatch { fingerprint: key.0 }); + } } for route in &self.routing { - if !self.materializations.contains_key(&route.materialization) { + let Some(materialization) = self.materializations.get(&route.materialization) else { return Err(ValidationError::UnknownMaterialization { fingerprint: route.materialization.0, }); + }; + if !materialization_satisfies(&route.satisfies, materialization) { + return Err(ValidationError::IncompatibleRoute { + fingerprint: route.materialization.0, + }); } } Ok(()) } } +fn kind_params_match(kind: &SummaryKind, params: &SummaryParams) -> bool { + matches!( + (kind, params), + (SummaryKind::Sum, SummaryParams::Sum) + | (SummaryKind::Count, SummaryParams::Count) + | (SummaryKind::MinMax, SummaryParams::MinMax) + | (SummaryKind::Increase, SummaryParams::Increase) + | (SummaryKind::Rate, SummaryParams::Rate) + | (SummaryKind::Kll, SummaryParams::Kll { .. }) + | (SummaryKind::Cms, SummaryParams::Cms { .. }) + | (SummaryKind::Hll, SummaryParams::Hll { .. }) + | (SummaryKind::DDSketch, SummaryParams::DDSketch { .. }) + | (SummaryKind::CmsWithHeap, SummaryParams::CmsWithHeap { .. }) + | (SummaryKind::Kmv, SummaryParams::Kmv { .. }) + | (SummaryKind::Theta, SummaryParams::Theta { .. }) + | (SummaryKind::CountSketch, SummaryParams::CountSketch { .. }) + | ( + SummaryKind::CountSketchWithHeap, + SummaryParams::CountSketchWithHeap { .. } + ) + ) +} + +fn materialization_satisfies(required: &Capability, m: &Materialization) -> bool { + let available = match m.kind { + SummaryKind::DDSketch => Capability::QuantileApprox(SketchKindHandle::DDSketch), + SummaryKind::Kll => Capability::QuantileApprox(SketchKindHandle::Kll), + SummaryKind::Hll => Capability::CardinalityApprox, + SummaryKind::Cms => Capability::FrequencyEstimate(SketchKindHandle::CountMin), + SummaryKind::CountSketch => Capability::FrequencyEstimate(SketchKindHandle::CountSketch), + SummaryKind::CmsWithHeap => Capability::FrequencyTopk(SketchKindHandle::CmsWithHeap), + SummaryKind::CountSketchWithHeap => { + Capability::FrequencyTopk(SketchKindHandle::CountSketchWithHeap) + } + SummaryKind::Sum => Capability::ExactAgg(AggregationType::Sum), + SummaryKind::MinMax => Capability::ExactAgg(AggregationType::MinMax), + SummaryKind::Increase => Capability::ExactAgg(AggregationType::Increase), + SummaryKind::Count | SummaryKind::Rate | SummaryKind::Kmv | SummaryKind::Theta => { + return false + } + }; + required.is_satisfied_by(&available) +} + #[cfg(test)] mod tests { use super::*; @@ -845,6 +907,7 @@ mod tests { }, group_by: vec!["zone".to_string()], rollup: vec![], + spatial_filter: String::new(), kind, params, col: ColumnRef::SampleValue, diff --git a/control_plane/src/emit/backend_push.rs b/control_plane/src/emit/backend_push.rs index 6154a2b58..2da63f794 100644 --- a/control_plane/src/emit/backend_push.rs +++ b/control_plane/src/emit/backend_push.rs @@ -49,11 +49,12 @@ use std::sync::Arc; use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use tokio::sync::Mutex; -use tracing::{debug, info, warn}; +use tracing::{info, warn}; use crate::backend_client::{BackendClient, BackendPostError}; -use crate::emit::{emit_backend_storage_routing, emit_backend_streaming_config_json}; +use crate::emit::emit_backend_storage_routing; use crate::physical::colored_dag::emitter::BackendStageConfig; +use crate::physical::compiler::{PlanEnvelope, PrecomputePlan}; use crate::workload::AggRole; /// Retry policy for transient POST failures. Tuned to bridge the @@ -227,66 +228,40 @@ pub enum PushOutcome { /// Returns the per-side success flags and the total attempts spent. async fn push_documents_coupled( client: &Arc, - streaming_body: String, + precompute_plan: &PrecomputePlan, routing_body: String, + plan_bytes: Vec, ) -> (bool, bool, u32) { let start = Instant::now(); - let mut streaming_ok = false; - let mut routing_ok = false; - // A permanent failure on a side disables further attempts on that side - // (retrying a 400 just floods the logs). - let mut streaming_permanent = false; - let mut routing_permanent = false; + let routing: serde_json::Value = match serde_json::from_str(&routing_body) { + Ok(value) => value, + Err(error) => { + warn!(%error, "invalid generated storage routing"); + return (false, false, 0); + } + }; for attempt in 1..=RETRY_MAX_ATTEMPTS { - // POST whichever side is still outstanding (not yet ok, not - // permanently failed). Re-POSTing an already-applied side is safe - // (idempotent swap) but wasteful, so we skip it. - if !streaming_ok && !streaming_permanent { - match client - .post_streaming_config_json_typed(streaming_body.clone()) - .await - { - Ok(()) => streaming_ok = true, - Err(BackendPostError::Permanent(e)) => { - streaming_permanent = true; - warn!(op = "streaming-config", error = %e, "permanent backend POST failure; will not retry this side"); - } - Err(BackendPostError::Transient(e)) => { - warn!(op = "streaming-config", attempt, error = %e, "transient backend POST failure (coupled)"); - } + match client + .post_physical_plan_typed(precompute_plan, plan_bytes.clone(), Some(routing.clone())) + .await + { + Ok(()) => return (true, true, attempt), + Err(BackendPostError::Permanent(error)) => { + warn!(%error, "permanent physical-plan publication failure"); + return (false, false, attempt); } - } - if !routing_ok && !routing_permanent { - match client - .post_storage_routing_json_typed(routing_body.clone()) - .await - { - Ok(()) => routing_ok = true, - Err(BackendPostError::Permanent(e)) => { - routing_permanent = true; - warn!(op = "storage-routing", error = %e, "permanent backend POST failure; will not retry this side"); - } - Err(BackendPostError::Transient(e)) => { - warn!(op = "storage-routing", attempt, error = %e, "transient backend POST failure (coupled)"); - } + Err(BackendPostError::Transient(error)) => { + warn!(attempt, %error, "transient physical-plan publication failure") } } - - // Both confirmed → done. Both terminal (ok or permanent) → no point - // sleeping. Otherwise back off and retry the outstanding side(s). - let streaming_done = streaming_ok || streaming_permanent; - let routing_done = routing_ok || routing_permanent; - if streaming_done && routing_done { - return (streaming_ok, routing_ok, attempt); - } if attempt < RETRY_MAX_ATTEMPTS { let delay = backoff_delay(attempt, start); tokio::time::sleep(delay).await; } } - (streaming_ok, routing_ok, RETRY_MAX_ATTEMPTS) + (false, false, RETRY_MAX_ATTEMPTS) } /// Required push of the encoded `BackendPlan` — no @@ -296,24 +271,6 @@ async fn push_documents_coupled( /// so the next cycle is itself the retry backstop — same contract /// [`push_or_log`] already establishes for the legacy YAML path. Logs at /// WARN on failure and return it to the coupled publication outcome. -async fn push_backend_plan_required(client: &Arc, bytes: Vec) -> bool { - match client.post_backend_plan_typed(bytes).await { - Ok(()) => { - debug!(stage = "backend", endpoint = %client.endpoint(), "BackendPlan push succeeded"); - true - } - Err(e) => { - warn!( - stage = "backend", - endpoint = %client.endpoint(), - error = %e, - "BackendPlan push failed; publication generation is incomplete" - ); - false - } - } -} - /// Update the cumulative cache with `be` for `(metric, role)` and /// POST the cumulative streaming-config + storage-routing JSON /// documents to the backend. @@ -461,24 +418,18 @@ async fn push_cumulative_entries( .flat_map(|(_, c)| c.readouts.iter().cloned()) .collect(), }; - let streaming_body = match emit_backend_streaming_config_json(&cumulative_be, monitors) { - Ok(doc) => doc.to_string(), - Err(e) => { - warn!(error = %e, "emit_backend_streaming_config_json failed; skipping coupled push"); - return PushOutcome::EmitFailed; - } - }; - // BackendPlan (design-backend-plan-wire-format.md): built from the // SAME `cumulative_be` snapshot as the legacy documents above, so all // three describe one consistent generation of planning state. Until // BackendPlan fully replaces the compatibility documents, publication // succeeds only when all three are accepted. + let plan_id = PLAN_ID_COUNTER.fetch_add(1, Ordering::Relaxed); + let generated_at_unix_ms = now_unix_ms(); let plan_bytes = match crate::backend_plan::from_stage_config( &cumulative_be, monitors, - PLAN_ID_COUNTER.fetch_add(1, Ordering::Relaxed), - now_unix_ms(), + plan_id, + generated_at_unix_ms, ) { Ok(plan) => plan.encode_to_vec(), Err(e) => { @@ -486,6 +437,26 @@ async fn push_cumulative_entries( return PushOutcome::EmitFailed; } }; + let precompute_plan = PrecomputePlan { + envelope: PlanEnvelope { + plan_id, + generated_at_unix_ms, + planner_revision: crate::physical::compiler::PLANNER_REVISION.into(), + capability_snapshot_id: "replanner".into(), + }, + materializations: match cumulative_be + .aggregations + .iter() + .map(crate::backend_plan::aggregation_config_for_materialization) + .collect::>>() + { + Ok(materializations) => materializations, + Err(error) => { + warn!(%error, "failed to build typed PrecomputePlan"); + return PushOutcome::EmitFailed; + } + }, + }; // Storage-routing: the routing classifier (`build_routing_entry` in // `emit/stage_config.rs`) reads `cfg.aggregations` to derive shape @@ -545,9 +516,8 @@ async fn push_cumulative_entries( }; let (streaming_ok, routing_ok, attempts) = - push_documents_coupled(client, streaming_body, routing_body).await; - - let plan_ok = push_backend_plan_required(client, plan_bytes).await; + push_documents_coupled(client, &precompute_plan, routing_body, plan_bytes).await; + let plan_ok = streaming_ok; if streaming_ok && routing_ok && plan_ok { info!( @@ -844,29 +814,17 @@ mod tests { }; let app = Router::new() .route( - "/api/v1/streaming-config", + "/api/v1/physical-plan", post( |State(m): State, _body: axum::body::Bytes| async move { m.streaming_hits.fetch_add(1, StdOrdering::SeqCst); - m.streaming_status - }, - ), - ) - .route( - "/api/v1/storage_routing", - post( - |State(m): State, _body: axum::body::Bytes| async move { m.routing_hits.fetch_add(1, StdOrdering::SeqCst); - m.routing_status - }, - ), - ) - .route( - "/api/v1/backend-plan", - post( - |State(m): State, _body: axum::body::Bytes| async move { m.plan_hits.fetch_add(1, StdOrdering::SeqCst); - axum::http::StatusCode::OK + if !m.streaming_status.is_success() { + m.streaming_status + } else { + m.routing_status + } }, ), ) @@ -928,17 +886,8 @@ mod tests { /// explicitly incomplete even if both compatibility documents landed. #[tokio::test] async fn backend_plan_push_failure_is_reported_as_desync() { - // A mock that only serves the legacy endpoints (no - // `/api/v1/backend-plan` route) — the plan push 404s. - let app = Router::new() - .route( - "/api/v1/streaming-config", - post(|_body: axum::body::Bytes| async { axum::http::StatusCode::OK }), - ) - .route( - "/api/v1/storage_routing", - post(|_body: axum::body::Bytes| async { axum::http::StatusCode::OK }), - ); + // A mock without the atomic endpoint: the whole generation fails. + let app = Router::new(); let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let addr = listener.local_addr().unwrap(); tokio::spawn(async move { @@ -962,8 +911,8 @@ mod tests { assert_eq!( outcome, PushOutcome::Desynced { - streaming_ok: true, - routing_ok: true, + streaming_ok: false, + routing_ok: false, plan_ok: false, }, "publication must not report success when the authoritative plan is missing" @@ -997,9 +946,9 @@ mod tests { assert_eq!( outcome, PushOutcome::Desynced { - streaming_ok: true, + streaming_ok: false, routing_ok: false, - plan_ok: true, + plan_ok: false, }, "one-sided failure must surface as Desynced, not silent success" ); diff --git a/control_plane/src/epsilon_alloc.rs b/control_plane/src/epsilon_alloc.rs index c6395b680..7a7fc7a79 100644 --- a/control_plane/src/epsilon_alloc.rs +++ b/control_plane/src/epsilon_alloc.rs @@ -450,7 +450,7 @@ mod tests { 0.05, &["quantile_over_time(0.99, latency_ms[5m])".to_string()], 1.0, - |_m, sks| metric_rate_from_telemetry(&store, sks), + |_m, _sks| metric_rate_from_telemetry(&store, &[SketchType::DDSketch]), &monitors, ); let m = resp diff --git a/control_plane/src/lib.rs b/control_plane/src/lib.rs index a71a6db57..0c443f670 100644 --- a/control_plane/src/lib.rs +++ b/control_plane/src/lib.rs @@ -86,12 +86,6 @@ pub mod workload; // (`emit`, `pipeline`, `intent_algebra::relational`, `optimizer`, // `physical`, `physical::colored_dag`, etc.) per the layered-cleanup // follow-up task. -/// PromQL → ASAP-tier candidate analyzer. Phase-9 unification of the -/// per-`Capability` dispatch knowledge that previously lived in -/// `asap-query-engine/src/query-engines/asap_query/warm_tier/promql_extract.rs`. See -/// the module docs for the full PromQL shape coverage matrix. -pub mod asap_tier_analysis; - /// PromQL → ASAPPlanner's canonical post-ASAP plan via /// `asap_aware_mapping::bind::implement_tree`. pub mod asap_tier_implement; diff --git a/control_plane/src/main.rs b/control_plane/src/main.rs index eba92f8e5..78f462f16 100644 --- a/control_plane/src/main.rs +++ b/control_plane/src/main.rs @@ -640,33 +640,12 @@ async fn handle_compile_and_publish_physical_plan( ) .into_response(); } - let precompute_stage_config = BackendStageConfig { - aggregations: bundle.precompute_plan.materializations.clone(), - readouts: Vec::new(), - }; - let precompute_config = - match emit::emit_backend_streaming_config_json(&precompute_stage_config, &[]) { - Ok(config) => config.to_string(), - Err(error) => { - return ( - StatusCode::INTERNAL_SERVER_ERROR, - format!("failed to emit precompute physical plan: {error}"), - ) - .into_response() - } - }; if let Err(error) = backend - .post_streaming_config_json_typed(precompute_config) - .await - { - return ( - StatusCode::BAD_GATEWAY, - format!("backend rejected precompute physical plan: {error}"), + .post_physical_plan_typed( + &bundle.precompute_plan, + bundle.backend_plan.encode_to_vec(), + None, ) - .into_response(); - } - if let Err(error) = backend - .post_backend_plan_typed(bundle.backend_plan.encode_to_vec()) .await { return ( @@ -732,9 +711,18 @@ fn compile_physical_plan_request( Ok(expr) => expr, Err(error) => return Err((StatusCode::UNPROCESSABLE_ENTITY, error.to_string())), }; + let post_asap = match physical::compiler::select_post_asap( + &expr, + query.accuracy.clone(), + &query.lifecycle, + request.evidence.get(&query.query_id), + ) { + Ok(plan) => plan, + Err(error) => return Err((StatusCode::UNPROCESSABLE_ENTITY, error.to_string())), + }; queries.push(physical::compiler::PlanningQuery { query_id: query.query_id, - expr, + post_asap, source: planner_types::pre_asap::Source::TimeSeries { metric: query.metric, }, @@ -3336,7 +3324,7 @@ mod api_tests { let sink: Arc = Arc::new(std::sync::Mutex::new(Vec::new())); let sink_capture = Arc::clone(&sink); let mock_app = axum::Router::new().route( - "/api/v1/storage_routing", + "/api/v1/physical-plan", axum::routing::post(move |body: axum::body::Bytes| { let sink = Arc::clone(&sink_capture); async move { @@ -3412,7 +3400,7 @@ mod api_tests { // MinIO and Thanos has them indexed. let last: serde_json::Value = serde_json::from_str(bodies.last().unwrap()).expect("last body is valid JSON"); - let metric_names: std::collections::BTreeSet = last["metrics"] + let metric_names: std::collections::BTreeSet = last["storage_routing"]["metrics"] .as_array() .expect("metrics array") .iter() @@ -3432,7 +3420,7 @@ mod api_tests { // queries to Thanos. Without this target the metric falls back // to `default_engine: sketch_store` and the archive miss // reproduces. - for m in last["metrics"].as_array().unwrap() { + for m in last["storage_routing"]["metrics"].as_array().unwrap() { let targets = m["targets"].as_array().expect("targets array"); let engines: Vec<&str> = targets .iter() @@ -3491,26 +3479,17 @@ mod api_tests { type SinkInner = std::sync::Mutex>; let sink: Arc = Arc::new(std::sync::Mutex::new(Vec::new())); let sink_capture = Arc::clone(&sink); - let mock_app = axum::Router::new() - .route( - "/api/v1/streaming-config", - axum::routing::post(move |body: axum::body::Bytes| { - let sink = Arc::clone(&sink_capture); - async move { - let s = String::from_utf8_lossy(&body).to_string(); - sink.lock().unwrap().push(s); - axum::http::StatusCode::OK - } - }), - ) - // Sibling storage-routing endpoint stubbed so the - // `handle_plan` cycle's second POST doesn't 404 and - // pollute the test log (the assertion only inspects the - // streaming-config sink). - .route( - "/api/v1/storage_routing", - axum::routing::post(|| async { axum::http::StatusCode::OK }), - ); + let mock_app = axum::Router::new().route( + "/api/v1/physical-plan", + axum::routing::post(move |body: axum::body::Bytes| { + let sink = Arc::clone(&sink_capture); + async move { + let s = String::from_utf8_lossy(&body).to_string(); + sink.lock().unwrap().push(s); + axum::http::StatusCode::OK + } + }), + ); let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let addr = listener.local_addr().unwrap(); tokio::spawn(async move { @@ -3576,7 +3555,7 @@ mod api_tests { // `archive_miss` failure documented on the sibling test. let last: serde_json::Value = serde_json::from_str(bodies.last().unwrap()).expect("last body is valid JSON"); - let aggs = last["aggregations"] + let aggs = last["precompute_plan"]["materializations"] .as_array() .expect("aggregations array on cumulative streaming-config body"); // Wire-format note: `build_backend_aggregation_json` writes diff --git a/control_plane/src/physical/colored_dag/emitter.rs b/control_plane/src/physical/colored_dag/emitter.rs index b1eba0638..f2f95221f 100644 --- a/control_plane/src/physical/colored_dag/emitter.rs +++ b/control_plane/src/physical/colored_dag/emitter.rs @@ -672,7 +672,7 @@ pub struct BackendStageConfig { /// payload is built by `emit::stage_config::build_backend_aggregation_json` /// (a hand-written JSON builder reading these fields), never a whole-struct /// serialize. -#[derive(Debug, Clone, PartialEq)] +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub struct BackendAggregation { /// Internal-only id (see struct doc). Not on the wire. pub aggregation_id: String, diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index 847816cad..65540589f 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -41,7 +41,9 @@ pub const PLANNER_REVISION: &str = "5d0b6f6edcac65edc89a72051f37977ab0c83031"; #[derive(Debug, Clone)] pub struct PlanningQuery { pub query_id: String, - pub expr: QueryExpr, + /// Planner-selected post-ASAP DAG. The physical compiler must not + /// re-select a summary family from pre-ASAP input. + pub post_asap: Rc, pub source: Source, pub window_secs: u64, /// Label names are deployment metadata because Planner's canonical IR @@ -137,10 +139,10 @@ pub struct CollectorPlan { /// PromQL string or ad-hoc scheduler job. The aggregation definitions are /// emitted to `/api/v1/streaming-config`, where the runtime matches incoming /// series, maintains windows, and writes content-addressed materializations. -#[derive(Debug, Clone)] +#[derive(Debug, Clone, Serialize, Deserialize)] pub struct PrecomputePlan { pub envelope: PlanEnvelope, - pub materializations: Vec, + pub materializations: Vec, } /// Complete physical projection of one post-ASAP planning decision. @@ -235,17 +237,7 @@ impl PhysicalCompiler { delete: false, }, ); - let node = crate::planner_selection::select_summary_with_evidence( - &query.expr, - &model, - &DefaultAccuracyModel, - &EqualSplitAllocator, - &QueryEvidence(evidence), - ) - .map_err(|error| CompileError::Query { - query_id: query.query_id.clone(), - reason: error.to_string(), - })?; + let node = query.post_asap.clone(); let selected = extract_selected(&node).ok_or_else(|| CompileError::Query { query_id: query.query_id.clone(), reason: "selected plan has no executable sketch materialization/readout".into(), @@ -327,7 +319,10 @@ impl PhysicalCompiler { .collect(); let precompute_plan = PrecomputePlan { envelope: envelope.clone(), - materializations: aggregations, + materializations: aggregations + .iter() + .map(backend_plan::aggregation_config_for_materialization) + .collect::, _>>()?, }; Ok(PhysicalPlan { envelope, @@ -338,6 +333,38 @@ impl PhysicalCompiler { } } +/// Planner-adapter selection step used before physical compilation. Keeping +/// this separate makes the ownership boundary explicit: callers supply the +/// selected post-ASAP DAG to [`PhysicalCompiler::compile`]. +pub fn select_post_asap( + expr: &QueryExpr, + accuracy: AccuracyTarget, + lifecycle: &LifecyclePlanningInput, + evidence: Option<&TopKMembershipEvidence>, +) -> Result, crate::planner_selection::SelectionError> { + let model = ControlPlaneCostModel::new(accuracy).with_summary_maintenance( + SummaryMaintenanceLifecycleCostInputs { + build_cost: Some(Cost(lifecycle.costs.build)), + maintenance_cost_per_update: Some(Cost(lifecycle.costs.maintenance_per_update)), + summary_read_cost: Some(Cost(lifecycle.costs.read)), + retention_cost_rate: Some(CostRate(lifecycle.costs.retention_per_second)), + retirement_cost: Some(Cost(lifecycle.costs.retirement)), + }, + SummaryMaintenanceCapabilities { + incremental_update: true, + merge: true, + delete: false, + }, + ); + crate::planner_selection::select_summary_with_evidence( + expr, + &model, + &DefaultAccuracyModel, + &EqualSplitAllocator, + &QueryEvidence(evidence), + ) +} + fn validate_evidence( query_id: &str, evidence: &TopKMembershipEvidence, @@ -551,40 +578,53 @@ mod tests { } } - fn request(query_id: &str, promql: &str) -> PlanningRequest { + fn request_with_evidence( + query_id: &str, + promql: &str, + evidence: Option, + ) -> Result { let accuracy = AccuracyTarget::EpsilonDelta { epsilon: 0.01, delta: 0.01, }; let parsed = crate::query_parser::parse_query_expr_canonical(promql, accuracy.clone()) .expect("canonical query"); - let expr = parsed; - PlanningRequest { + let lifecycle = LifecyclePlanningInput { + evaluation_interval_ms: 10_000, + ingestion_rate_per_second: 100.0, + evidence_observed_at_unix_ms: 9_500, + evidence_valid_for_ms: 60_000, + horizon_seconds: 300.0, + costs: LifecycleCostEvidence { + build: 10.0, + maintenance_per_update: 0.001, + read: 0.1, + retention_per_second: 0.001, + retirement: 1.0, + }, + }; + let post_asap = select_post_asap(&parsed, accuracy.clone(), &lifecycle, evidence.as_ref())?; + let mut evidence_by_query = HashMap::new(); + if let Some(evidence) = evidence { + evidence_by_query.insert(query_id.to_string(), evidence); + } + Ok(PlanningRequest { queries: vec![PlanningQuery { query_id: query_id.into(), - expr, + post_asap, source: Source::TimeSeries { metric: "m".into() }, window_secs: 60, group_by: vec![], accuracy, - lifecycle: LifecyclePlanningInput { - evaluation_interval_ms: 10_000, - ingestion_rate_per_second: 100.0, - evidence_observed_at_unix_ms: 9_500, - evidence_valid_for_ms: 60_000, - horizon_seconds: 300.0, - costs: LifecycleCostEvidence { - build: 10.0, - maintenance_per_update: 0.001, - read: 0.1, - retention_per_second: 0.001, - retirement: 1.0, - }, - }, + lifecycle, }], - evidence: HashMap::new(), + evidence: evidence_by_query, planner_revision: PLANNER_REVISION.into(), - } + }) + } + + fn request(query_id: &str, promql: &str) -> PlanningRequest { + request_with_evidence(query_id, promql, None).expect("post-ASAP selection") } #[test] @@ -635,25 +675,23 @@ mod tests { #[test] fn topk_fails_closed_without_membership_evidence() { - let error = PhysicalCompiler - .compile(request("q-topk", "topk(5, m)"), environment(10_000)) - .expect_err("missing certificate must fail"); - assert!(matches!(error, CompileError::Query { .. })); + assert!(request_with_evidence("q-topk", "topk(5, m)", None).is_err()); } #[test] fn stale_topk_evidence_is_rejected_before_planner_selection() { - let mut request = request("q-topk", "topk(5, count_over_time(m[1m]))"); - request.evidence.insert( - "q-topk".into(), - TopKMembershipEvidence { + let request = request_with_evidence( + "q-topk", + "topk(5, count_over_time(m[1m]))", + Some(TopKMembershipEvidence { selected_lower_bound: 101.0, excluded_upper_bound: 100.0, interval_failure_probability: 0.005, observed_at_unix_ms: 1, source: "runtime-margin-monitor".into(), - }, - ); + }), + ) + .expect("selection accepts evidence before freshness validation"); let error = PhysicalCompiler .compile(request, environment(100_000)) .expect_err("stale certificate must fail"); @@ -662,17 +700,18 @@ mod tests { #[test] fn fresh_topk_evidence_enables_physical_compilation() { - let mut request = request("q-topk", "topk(5, count_over_time(m[1m]))"); - request.evidence.insert( - "q-topk".into(), - TopKMembershipEvidence { + let request = request_with_evidence( + "q-topk", + "topk(5, count_over_time(m[1m]))", + Some(TopKMembershipEvidence { selected_lower_bound: 101.0, excluded_upper_bound: 100.0, interval_failure_probability: 0.005, observed_at_unix_ms: 9_500, source: "runtime-margin-monitor".into(), - }, - ); + }), + ) + .expect("selection accepts valid evidence"); let bundle = PhysicalCompiler .compile(request, environment(10_000)) .expect("certified TopK compiles"); diff --git a/control_plane/src/query_parser/mod.rs b/control_plane/src/query_parser/mod.rs index c9f2f8f75..e494a0daa 100644 --- a/control_plane/src/query_parser/mod.rs +++ b/control_plane/src/query_parser/mod.rs @@ -103,8 +103,7 @@ pub fn parse_query_expr_canonical( /// Parse a PromQL query string into a [`ParsedQuery`]. /// -/// This is the backward-compatible entry point for the existing -/// [`crate::analyzer::Analyzer`]. Internally it parses via +/// Flat compatibility entry point for workload metadata extraction. Internally it parses via /// [`parse_query_expr_canonical`] and extracts the flat summary by walking /// the canonical [`QueryExpr`] tree. pub fn parse_query(query: &str, accuracy: AccuracyTarget) -> anyhow::Result { @@ -112,20 +111,6 @@ pub fn parse_query(query: &str, accuracy: AccuracyTarget) -> anyhow::Result ParsedQuery { - qe_to_parsed_query(qe) -} - /// Extract a flat [`ParsedQuery`] by walking a canonical [`QueryExpr`] tree. /// /// Step γ7: the collector walks the canonical IR. `Aggregate` carries diff --git a/control_plane/src/query_planning.rs b/control_plane/src/query_planning.rs index 3cdec931d..48205b09f 100644 --- a/control_plane/src/query_planning.rs +++ b/control_plane/src/query_planning.rs @@ -2,10 +2,7 @@ //! plan**. //! //! This composes two existing pieces with slice 1's inverse map: -//! 1. [`analyze_promql_for_asap_tier`] lowers each query to -//! `ASAPTierCandidate`s, each already carrying a `metric_name` and the -//! `required_capability` the query needs (and an `unsupported` reason for -//! the parts that only the cold tier can answer). +//! 1. ASAPPlanner lowers each query to its canonical post-ASAP summary DAG. //! 2. [`required_sketches_for_capabilities`](crate::sketch_selection::required_sketches_for_capabilities) //! (slice 1) maps a capability set to the `SketchType` families that satisfy it. //! @@ -16,7 +13,6 @@ use std::collections::BTreeMap; -use crate::asap_tier_analysis::{analyze_promql_for_asap_tier, UnsupportedReason}; use crate::physical::runtime_capability::Capability; use crate::sketch_selection::required_sketches_for_capabilities; use crate::types::SketchType; @@ -40,7 +36,7 @@ pub struct ColdQuery { pub query: String, /// Why it isn't ASAP-tier-answerable (`None` only in the degenerate case /// where the analyzer returned no candidates and no explicit reason). - pub reason: Option, + pub reason: Option, } /// The full allocation plan for a query set: what to allocate warm, and what @@ -91,21 +87,23 @@ where for q in queries { let q = q.as_ref(); - let analysis = analyze_promql_for_asap_tier(q); - - for cand in &analysis.candidates { - let caps = by_metric.entry(cand.metric_name.clone()).or_default(); - if !caps.contains(&cand.required_capability) { - caps.push(cand.required_capability.clone()); + let planned = crate::asap_tier_implement::implement_promql_for_asap_tier(q); + let mut found = false; + if let Ok(nodes) = &planned { + for node in nodes { + if let Some((metric, capability)) = planned_capability(node) { + found = true; + let caps = by_metric.entry(metric).or_default(); + if !caps.contains(&capability) { + caps.push(capability); + } + } } } - - // No answerable candidate at all, OR a partially-supported query with a - // residual unsupported reason → the (residual) query goes cold. - if analysis.candidates.is_empty() || analysis.unsupported.is_some() { + if !found { cold_only.push(ColdQuery { query: q.to_string(), - reason: analysis.unsupported, + reason: planned.err().map(|error| format!("{error:?}")), }); } } @@ -128,6 +126,93 @@ where } } +fn planned_capability( + node: &planner_types::post_asap::SummaryNode, +) -> Option<(String, Capability)> { + use crate::physical::runtime_capability::SketchKindHandle; + use planner_types::post_asap::{SketchQuery, SummaryExpr, SummaryFamilyType}; + + fn metric(node: &planner_types::post_asap::SummaryNode) -> Option { + fn from_query(expr: &planner_types::pre_asap::QueryExpr) -> Option { + use planner_types::pre_asap::{QueryExpr, Source}; + match expr { + QueryExpr::Scan { + source: Source::TimeSeries { metric }, + .. + } => Some(metric.clone()), + QueryExpr::Filter { child, .. } + | QueryExpr::Project { child, .. } + | QueryExpr::Aggregate { child, .. } + | QueryExpr::Dedup { child, .. } + | QueryExpr::Sort { child, .. } + | QueryExpr::Limit { child, .. } + | QueryExpr::PromqlSubquery { child, .. } + | QueryExpr::TimeRange { child, .. } + | QueryExpr::TimeShift { child, .. } + | QueryExpr::SQLWindowFunc { child, .. } => from_query(child), + _ => None, + } + } + match &node.expr { + SummaryExpr::KeepPreAsap(expr) => from_query(expr), + SummaryExpr::SummaryAgg { child, .. } => metric(child), + SummaryExpr::SummaryEstimate { summary_input, .. } => metric(summary_input), + SummaryExpr::SummaryMerge { children } => { + children.iter().find_map(|child| metric(child)) + } + _ => None, + } + } + + fn handle(family: &SummaryFamilyType) -> Option { + let SummaryFamilyType::Sketch(kind, _) = family else { + return None; + }; + Some(match asap_types::SummaryKind::from(kind.clone()) { + asap_types::SummaryKind::DDSketch => SketchKindHandle::DDSketch, + asap_types::SummaryKind::Kll => SketchKindHandle::Kll, + asap_types::SummaryKind::Hll => SketchKindHandle::Hll, + asap_types::SummaryKind::Cms => SketchKindHandle::CountMin, + asap_types::SummaryKind::CmsWithHeap => SketchKindHandle::CmsWithHeap, + asap_types::SummaryKind::CountSketch => SketchKindHandle::CountSketch, + asap_types::SummaryKind::CountSketchWithHeap => SketchKindHandle::CountSketchWithHeap, + _ => return None, + }) + } + + let metric = metric(node)?; + let capability = match &node.expr { + SummaryExpr::SummaryEstimate { + summary_input, + query, + } => { + let SummaryExpr::SummaryAgg { family, .. } = &summary_input.expr else { + return None; + }; + match query { + SketchQuery::Quantile { .. } => Capability::QuantileApprox(handle(family)?), + SketchQuery::Cardinality => Capability::CardinalityApprox, + SketchQuery::PointCount { .. } => Capability::FrequencyEstimate(handle(family)?), + SketchQuery::TopK { .. } => Capability::FrequencyTopk(handle(family)?), + } + } + SummaryExpr::SummaryAgg { + family: SummaryFamilyType::ExactAggregate(kind, _), + .. + } => { + let agg = match asap_types::SummaryKind::from(kind.clone()) { + asap_types::SummaryKind::Sum => asap_types::AggregationType::Sum, + asap_types::SummaryKind::Increase => asap_types::AggregationType::Increase, + asap_types::SummaryKind::MinMax => asap_types::AggregationType::MinMax, + _ => return None, + }; + Capability::ExactAgg(agg) + } + _ => return None, + }; + Some((metric, capability)) +} + #[cfg(test)] mod tests { use super::*; diff --git a/control_plane/src/replan.rs b/control_plane/src/replan.rs index 93cfb0746..c6631fc1b 100644 --- a/control_plane/src/replan.rs +++ b/control_plane/src/replan.rs @@ -1219,7 +1219,7 @@ mod tests { let hits = StdArc::new(AtomicU32::new(0)); let app = Router::new() .route( - "/api/v1/streaming-config", + "/api/v1/physical-plan", post( |State(h): State>, _b: axum::body::Bytes| async move { h.fetch_add(1, Ordering::SeqCst); @@ -1227,14 +1227,6 @@ mod tests { }, ), ) - .route( - "/api/v1/storage_routing", - post(|_b: axum::body::Bytes| async move { axum::http::StatusCode::OK }), - ) - .route( - "/api/v1/backend-plan", - post(|_b: axum::body::Bytes| async move { axum::http::StatusCode::OK }), - ) .with_state(StdArc::clone(&hits)); let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let addr = listener.local_addr().unwrap(); diff --git a/control_plane/src/workload.rs b/control_plane/src/workload.rs index 7660b60b3..52b65f9a0 100644 --- a/control_plane/src/workload.rs +++ b/control_plane/src/workload.rs @@ -152,7 +152,7 @@ pub fn derive_agg_role(entry: &WorkloadEntry) -> AggRole { return AggRole::Other; }; let mut intents: Vec = Vec::new(); - crate::asap_tier_analysis::collect_agg_intents(&expr, &mut intents); + collect_agg_intents(&expr, &mut intents); let Some(outer) = intents.first() else { // No Aggregate node at all — a bare metric selector (or a // window-only shape with no AggType to map onto). Bare @@ -176,6 +176,43 @@ pub fn derive_agg_role(entry: &WorkloadEntry) -> AggRole { } } +fn collect_agg_intents(expr: &planner_types::pre_asap::QueryExpr, out: &mut Vec) { + use planner_types::pre_asap::QueryExpr; + match expr { + QueryExpr::Aggregate { + measures, child, .. + } => { + out.extend(measures.iter().cloned()); + collect_agg_intents(child, out); + } + QueryExpr::Filter { child, .. } + | QueryExpr::Project { child, .. } + | QueryExpr::Dedup { child, .. } + | QueryExpr::Sort { child, .. } + | QueryExpr::Limit { child, .. } + | QueryExpr::PromqlSubquery { child, .. } + | QueryExpr::TimeRange { child, .. } + | QueryExpr::TimeShift { child, .. } + | QueryExpr::SQLWindowFunc { child, .. } => collect_agg_intents(child, out), + QueryExpr::Concat { children } => { + for child in children { + collect_agg_intents(child, out); + } + } + QueryExpr::Join { left, right, .. } + | QueryExpr::SetOp { left, right, .. } + | QueryExpr::BinaryOp { + lhs: left, + rhs: right, + .. + } => { + collect_agg_intents(left, out); + collect_agg_intents(right, out); + } + _ => {} + } +} + /// A single workload entry from the workloads YAML file. #[derive(Debug, Clone, Serialize, Deserialize)] pub struct WorkloadEntry { diff --git a/crates/asap_types/src/accumulator_spec.rs b/crates/asap_types/src/accumulator_spec.rs index 45caa0cf2..e57f23c2d 100644 --- a/crates/asap_types/src/accumulator_spec.rs +++ b/crates/asap_types/src/accumulator_spec.rs @@ -95,7 +95,8 @@ use crate::AggregationType; /// variant, e.g. `CmsWithHeap` vs `Cms`). Vendored (see module doc): same /// 14-variant shape ASAPController's pre-split `asap_sketch::SummaryKind` /// had. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)] +#[serde(rename_all = "snake_case")] pub enum SummaryKind { Sum, Count, @@ -132,7 +133,8 @@ impl SummaryKind { /// Typed tuning parameters matching a [`SummaryKind`]. Vendored /// alongside it (see module doc) — same shape as ASAPController's /// pre-split `asap_sketch::SummaryParams`. -#[derive(Debug, Clone, PartialEq)] +#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)] +#[serde(rename_all = "snake_case")] pub enum SummaryParams { Sum, Count, diff --git a/crates/asap_types/src/routing_index.rs b/crates/asap_types/src/routing_index.rs index 038d6dc68..098ae7db9 100644 --- a/crates/asap_types/src/routing_index.rs +++ b/crates/asap_types/src/routing_index.rs @@ -33,7 +33,7 @@ //! cached derived value through `HotReloadStreamingConfig`'s swap path) //! and is not done by this type on its own. -use std::collections::HashMap; +use std::collections::{BTreeSet, HashMap}; use crate::aggregation_config::AggregationConfig; use crate::policy_fingerprint::PolicyFingerprint; @@ -92,6 +92,36 @@ impl RoutingIndex { pub fn fingerprints(&self) -> impl Iterator + '_ { self.registry.fingerprints() } + + /// Resolve an ingest-side sketch shape to exactly one configured policy. + /// Ambiguous or absent matches fail closed. + pub fn find_policy_by_content( + &self, + metric: &str, + group_by_keys: &BTreeSet, + agg_type: crate::AggregationType, + expected_params: &HashMap, + ) -> Option { + let mut hit = None; + for fp in self.candidates_for_metric(metric) { + let cfg = self.get(*fp)?; + let policy_keys: BTreeSet<_> = cfg.grouping_labels.labels.iter().cloned().collect(); + if cfg.aggregation_type != agg_type + || &policy_keys != group_by_keys + || !cfg.spatial_filter_normalized.is_empty() + || !expected_params + .iter() + .all(|(key, value)| cfg.parameters.get(key) == Some(value)) + { + continue; + } + if hit.is_some() { + return None; + } + hit = Some(*fp); + } + hit + } } #[cfg(test)] diff --git a/data_plane/src/drivers/ingest/otel.rs b/data_plane/src/drivers/ingest/otel.rs index c62a077b6..6659ef725 100644 --- a/data_plane/src/drivers/ingest/otel.rs +++ b/data_plane/src/drivers/ingest/otel.rs @@ -1814,15 +1814,10 @@ fn derive_sketch_policy_fp( }; let params = sketch_config_to_params(cfg); let snap = ingest_state.config_snapshot(); - let registry = snap.policy_registry(); - control_plane::asap_tier_analysis::find_policy_by_content( - ®istry, - metric, - group_by_keys, - agg_type, - ¶ms, - ) - .unwrap_or(asap_types::PolicyFingerprint::UNSET) + let index = asap_types::RoutingIndex::build(snap.policy_registry()); + index + .find_policy_by_content(metric, group_by_keys, agg_type, ¶ms) + .unwrap_or(asap_types::PolicyFingerprint::UNSET) } /// Phase 5 helper — map a `ModifiedOtlpSketchDp` to the matching diff --git a/data_plane/src/drivers/query/servers/http.rs b/data_plane/src/drivers/query/servers/http.rs index 7edeb40b3..be50c40e1 100644 --- a/data_plane/src/drivers/query/servers/http.rs +++ b/data_plane/src/drivers/query/servers/http.rs @@ -146,6 +146,9 @@ pub struct HttpServer { /// normal routing-table path (which goes to Thanos for the cold /// archive and observes the 60–90 s flush gap). probe_cache: Option>, + /// Serializes multi-document physical-plan publication so two control + /// plane generations cannot interleave their config and BackendPlan. + physical_plan_lock: Arc>, } #[derive(Clone)] @@ -171,6 +174,7 @@ struct AppState { data_retention_ms: Option, /// See [`HttpServer::probe_cache`]. probe_cache: Option>, + physical_plan_lock: Arc>, } impl HttpServer { @@ -195,6 +199,7 @@ impl HttpServer { backfill: None, data_retention_ms: None, probe_cache: None, + physical_plan_lock: Arc::new(tokio::sync::Mutex::new(())), } } @@ -358,6 +363,7 @@ impl HttpServer { backfill: self.backfill.clone(), data_retention_ms: self.data_retention_ms, probe_cache: self.probe_cache.clone(), + physical_plan_lock: self.physical_plan_lock.clone(), }; let range_query_endpoint = adapter.get_range_query_endpoint(); @@ -383,6 +389,7 @@ impl HttpServer { "/api/v1/backend-plan", get(handle_get_backend_plan).post(handle_post_backend_plan), ) + .route("/api/v1/physical-plan", post(handle_post_physical_plan)) // Phase α (MVP): control-plane-pushed `BackendStorageRouting` // table. POST replaces the current table atomically; GET // returns a JSON snapshot for operator diagnostics. @@ -443,6 +450,7 @@ impl HttpServer { backfill: self.backfill.clone(), data_retention_ms: self.data_retention_ms, probe_cache: self.probe_cache.clone(), + physical_plan_lock: self.physical_plan_lock.clone(), }; let range_query_endpoint = adapter.get_range_query_endpoint(); @@ -464,6 +472,7 @@ impl HttpServer { "/api/v1/backend-plan", get(handle_get_backend_plan).post(handle_post_backend_plan), ) + .route("/api/v1/physical-plan", post(handle_post_physical_plan)) // Phase α (MVP): control-plane-pushed `BackendStorageRouting` // table. POST replaces the current table atomically; GET // returns a JSON snapshot for operator diagnostics. @@ -5335,6 +5344,130 @@ async fn handle_post_backend_plan( (StatusCode::OK, axum::Json(body)).into_response() } +#[derive(serde::Deserialize)] +struct PhysicalPlanInstallRequest { + precompute_plan: control_plane::physical::compiler::PrecomputePlan, + backend_plan: Vec, + storage_routing: Option, +} + +/// Install the two backend views of one PhysicalPlan as one validated +/// publication. The BackendPlan is installed first, so readers racing the +/// short swap interval fail closed on missing fingerprints rather than using +/// a new streaming configuration with stale routing authority. +async fn handle_post_physical_plan( + State(state): State, + axum::Json(request): axum::Json, +) -> axum::response::Response { + use axum::http::StatusCode; + use axum::response::IntoResponse; + use std::collections::BTreeSet; + + let (Some(config_handle), Some(plan_handle)) = ( + state.hot_reload_config.as_ref(), + state.hot_reload_backend_plan.as_ref(), + ) else { + return ( + StatusCode::SERVICE_UNAVAILABLE, + axum::Json(serde_json::json!({ + "status": "error", "error": "physical-plan hot-reload handles are not attached" + })), + ) + .into_response(); + }; + let new_config = crate::storage_engines::types::StreamingConfig::new( + request + .precompute_plan + .materializations + .iter() + .cloned() + .map(|config| (config.policy_fp_u64(), config)) + .collect(), + ); + let new_plan = match control_plane::backend_plan::BackendPlan::decode(&request.backend_plan) { + Ok(plan) => plan, + Err(error) => { + return ( + StatusCode::BAD_REQUEST, + axum::Json(serde_json::json!({ + "status": "error", "error": format!("BackendPlan decode error: {error}") + })), + ) + .into_response() + } + }; + if let Err(error) = new_plan.validate() { + return ( + StatusCode::UNPROCESSABLE_ENTITY, + axum::Json(serde_json::json!({ + "status": "error", "error": format!("BackendPlan validation error: {error}") + })), + ) + .into_response(); + } + let config_fps: BTreeSet = new_config.aggregation_configs.keys().copied().collect(); + let plan_fps: BTreeSet = new_plan.materializations.keys().map(|fp| fp.0).collect(); + if config_fps != plan_fps { + return ( + StatusCode::UNPROCESSABLE_ENTITY, + axum::Json(serde_json::json!({ + "status": "error", + "error": "streaming-config and BackendPlan materialization fingerprints differ" + })), + ) + .into_response(); + } + let new_routing = match request.storage_routing.as_ref() { + Some(value) => match crate::storage_engines::types::BackendStorageRouting::from_json_payload(value) { + Ok(routing) => Some(routing), + Err(error) => return (StatusCode::BAD_REQUEST, axum::Json(serde_json::json!({ + "status": "error", "error": format!("BackendStorageRouting build error: {error:#}") + }))).into_response(), + }, + None => None, + }; + if new_routing.is_some() && state.backend_storage_routing.is_none() { + return ( + StatusCode::SERVICE_UNAVAILABLE, + axum::Json(serde_json::json!({ + "status": "error", "error": "storage-routing hot-reload handle is not attached" + })), + ) + .into_response(); + } + + let _guard = state.physical_plan_lock.lock().await; + let plan_id = new_plan.plan_id; + if let Err(error) = plan_handle.install(new_plan) { + return ( + StatusCode::CONFLICT, + axum::Json(serde_json::json!({ + "status": "error", "error": format!("BackendPlan install error: {error}") + })), + ) + .into_response(); + } + config_handle.swap(new_config); + if let (Some(handle), Some(routing)) = (state.backend_storage_routing.as_ref(), new_routing) { + let tenant = routing.tenant().to_string(); + handle.swap_tenant(&tenant, routing); + } + let snap = config_handle.snapshot(); + let sid_summary = crate::storage_engines::sketch_db::lifecycle::reconcile_from_streaming_config( + state.sketch_index.as_ref(), + snap.as_ref(), + crate::storage_engines::sketch_db::DEFAULT_RETIREMENT_RETENTION, + ); + ( + StatusCode::OK, + axum::Json(serde_json::json!({ + "status": "success", "plan_id": plan_id, + "materialization_count": plan_fps.len(), "sids_retired": sid_summary.retired + })), + ) + .into_response() +} + // ── Phase α: BackendStorageRouting hot-reload endpoints ──────────── /// `GET /api/v1/storage_routing` — return a JSON snapshot of the diff --git a/data_plane/src/query_engines/asap_query_engine/post_asap_planner.rs b/data_plane/src/query_engines/asap_query_engine/post_asap_planner.rs index 9d5936282..aac5a685f 100644 --- a/data_plane/src/query_engines/asap_query_engine/post_asap_planner.rs +++ b/data_plane/src/query_engines/asap_query_engine/post_asap_planner.rs @@ -223,10 +223,10 @@ fn observed_family_for_metric( /// `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( +fn observed_families_for_metric_from_plan( plan: &control_plane::backend_plan::BackendPlan, metric: &str, -) -> Option<(SketchAlgorithm, SketchParams)> { +) -> Vec<(SketchAlgorithm, SketchParams)> { let mut warm_fingerprints: Vec<_> = plan .routing .iter() @@ -237,7 +237,7 @@ fn observed_family_for_metric_from_plan( .collect(); warm_fingerprints.sort_by_key(|fp| fp.0); warm_fingerprints.dedup(); - warm_fingerprints.into_iter().find_map(|fingerprint| { + warm_fingerprints.into_iter().filter_map(|fingerprint| { let m = plan.materializations.get(&fingerprint)?; if !matches!(&m.source, planner_types::pre_asap::Source::TimeSeries { metric: mm } if mm == metric) { @@ -249,16 +249,39 @@ fn observed_family_for_metric_from_plan( // materializations (mirrors `observed_family_for_metric`'s // "first sketch-typed one found" semantics). Some((m.kind.as_sketch_kind()?, m.params.as_sketch_params()?)) - }) + }).collect() } -fn backend_plan_covers_post_asap( +pub fn resolve_materializations_for_post_asap( plan: &control_plane::backend_plan::BackendPlan, node: &SummaryNode, -) -> Result<(), LoweringSkip> { + query: &str, + accuracy: AccuracyTarget, +) -> Result, LoweringSkip> { + let parsed = control_plane::query_parser::parse_query(query, accuracy) + .map_err(|e| LoweringSkip::ParseFailed(e.to_string()))?; + let spatial_filter = if parsed.label_filters.is_empty() { + String::new() + } else { + let rendered = parsed + .label_filters + .iter() + .map(|(key, value)| format!("{key}=\"{value}\"")) + .collect::>() + .join(","); + asap_types::utils::normalize_spatial_filter(&rendered) + }; + let required_groups: std::collections::BTreeSet<_> = + parsed.group_by_labels.into_iter().collect(); + let query_window_ms = parsed.time_window.as_millis() as u64; + fn visit( plan: &control_plane::backend_plan::BackendPlan, node: &SummaryNode, + spatial_filter: &str, + required_groups: &std::collections::BTreeSet, + query_window_ms: u64, + resolved: &mut std::collections::BTreeSet, ) -> Result<(), LoweringSkip> { match &node.expr { SummaryExpr::SummaryAgg { child, family, .. } => { @@ -281,25 +304,51 @@ fn backend_plan_covers_post_asap( ))) } }; - let covered = plan.routing.iter().any(|route| { - route.storage_backend == control_plane::backend_plan::StorageBackend::SketchStore + let matches: Vec<_> = plan.routing.iter().filter_map(|route| { + (route.storage_backend == control_plane::backend_plan::StorageBackend::SketchStore && plan.materializations.get(&route.materialization).is_some_and(|m| { matches!(&m.source, planner_types::pre_asap::Source::TimeSeries { metric: mm } if mm == &metric) && m.kind == kind && m.params == params - }) - }); - if !covered { + && m.spatial_filter == spatial_filter + && required_groups.iter().all(|key| m.group_by.contains(key)) + && m.window.size_ms <= query_window_ms + })) + .then_some(route.materialization) + }).collect(); + if matches.is_empty() { return Err(LoweringSkip::NoWarmRoute(format!( "no warm BackendPlan route for metric `{metric}` and family `{kind:?}`" ))); } - visit(plan, child) + resolved.extend(matches); + visit( + plan, + child, + spatial_filter, + required_groups, + query_window_ms, + resolved, + ) } - SummaryExpr::SummaryEstimate { summary_input, .. } => visit(plan, summary_input), + SummaryExpr::SummaryEstimate { summary_input, .. } => visit( + plan, + summary_input, + spatial_filter, + required_groups, + query_window_ms, + resolved, + ), SummaryExpr::SummaryMerge { children } => { for child in children { - visit(plan, child)?; + visit( + plan, + child, + spatial_filter, + required_groups, + query_window_ms, + resolved, + )?; } Ok(()) } @@ -309,7 +358,16 @@ fn backend_plan_covers_post_asap( )), } } - visit(plan, node) + let mut resolved = std::collections::BTreeSet::new(); + visit( + plan, + node, + &spatial_filter, + &required_groups, + query_window_ms, + &mut resolved, + )?; + Ok(resolved) } fn query_expr_contains_time_range(qe: &planner_types::pre_asap::QueryExpr) -> bool { @@ -580,35 +638,78 @@ pub fn plan_promql_to_post_asap( // 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).map_err(|e: BindingError| match e { - BindingError::Implement( + let metric = find_metric_in_query_expr(&qe); + let mut observed = metric + .as_deref() + .and_then(|metric| { + backend_plan.map(|plan| observed_families_for_metric_from_plan(plan, metric)) + }) + .unwrap_or_default(); + if observed.is_empty() { + if let Some(family) = metric + .as_deref() + .and_then(|metric| observed_family_for_metric(index, metric)) + { + observed.push(family); + } + } + // No installed family means the planner may use its accuracy-driven + // default. With a BackendPlan, try every family materialized for the + // metric: a fingerprint does not uniquely identify query semantics, + // and choosing the first family could incorrectly fall back while a + // later materialization is an exact match. + let candidates: Vec<_> = if observed.is_empty() { + vec![None] + } else { + observed.into_iter().map(Some).collect() + }; + let mut last_skip = LoweringSkip::NotRealized; + for observed in candidates { + let cost_model = ObservedFamilyCostModel::new(accuracy.clone(), observed); + let physical = match bind_query_expr_with_cost_model(&qe, &cost_model) { + Ok(physical) => physical, + Err(BindingError::Implement( control_plane::planner_selection::SelectionError::NoLegalCandidate, - ) => LoweringSkip::NotRealized, - other => LoweringSkip::Implement(other.to_string()), - })?; - - match physical { - PhysicalExpr::Committed(PostAsapPlan::Summary(node)) => { - if matches!(node.expr, SummaryExpr::KeepPreAsap(_)) { - Err(LoweringSkip::NotRealized) - } else { - ensure_warm_runtime_support(&node, source_has_filter)?; - if let Some(plan) = backend_plan { - backend_plan_covers_post_asap(plan, &node)?; + )) => { + last_skip = LoweringSkip::NotRealized; + continue; + } + Err(other) => { + last_skip = LoweringSkip::Implement(other.to_string()); + continue; + } + }; + + match physical { + PhysicalExpr::Committed(PostAsapPlan::Summary(node)) => { + if matches!(node.expr, SummaryExpr::KeepPreAsap(_)) { + last_skip = LoweringSkip::NotRealized; + } else { + if let Err(skip) = ensure_warm_runtime_support(&node, source_has_filter) { + last_skip = skip; + continue; + } + if let Some(plan) = backend_plan { + match resolve_materializations_for_post_asap( + plan, + &node, + query, + accuracy.clone(), + ) { + Ok(_) => return Ok(node), + Err(skip) => { + last_skip = skip; + continue; + } + } + } + return Ok(node); } - Ok(node) } + _ => last_skip = LoweringSkip::UnsupportedPhysicalShape, } - _ => Err(LoweringSkip::UnsupportedPhysicalShape), } + Err(last_skip) } #[cfg(test)] @@ -810,6 +911,7 @@ mod tests { }, group_by: Vec::new(), rollup: Vec::new(), + spatial_filter: String::new(), // `Materialization.kind`/`.params` span both exact // accumulators and sketches -- the flat // `asap_types::SummaryKind`, not this file's own diff --git a/data_plane/src/query_engines/asap_query_engine/post_asap_readout.rs b/data_plane/src/query_engines/asap_query_engine/post_asap_readout.rs index 50ff4297b..c8fd1845f 100644 --- a/data_plane/src/query_engines/asap_query_engine/post_asap_readout.rs +++ b/data_plane/src/query_engines/asap_query_engine/post_asap_readout.rs @@ -8,7 +8,7 @@ use crate::query_engines::asap_query_engine::summary_exec::{execute, ExecOutcome use control_plane::types_v2::AccuracyTarget; use crate::query_engines::asap_query_engine::post_asap_planner::{ - execution_hints, plan_promql_to_post_asap, LoweringSkip, + execution_hints, plan_promql_to_post_asap, resolve_materializations_for_post_asap, LoweringSkip, }; use crate::query_engines::asap_query_engine::summary_executor::{ QueryExecutionContext, SummaryValue, @@ -72,8 +72,17 @@ pub fn execute_post_asap_readout( accuracy: AccuracyTarget, backend_plan: Option<&control_plane::backend_plan::BackendPlan>, ) -> Result { - let node = plan_promql_to_post_asap(index, query, accuracy, backend_plan)?; - execute_planned_post_asap(index, &node, t0_ms, t1_ms, is_cumulative, backend_plan) + let node = plan_promql_to_post_asap(index, query, accuracy.clone(), backend_plan)?; + execute_planned_post_asap( + index, + &node, + query, + accuracy, + t0_ms, + t1_ms, + is_cumulative, + backend_plan, + ) } /// Plan and execute an instant query without consulting the legacy candidate @@ -87,7 +96,7 @@ pub fn execute_post_asap_instant( backend_plan: Option<&control_plane::backend_plan::BackendPlan>, ) -> Result<(PostAsapReadoutOutcome, u64), LoweringSkip> { const DEFAULT_LOOKBACK_MS: u64 = 5 * 60 * 1000; - let node = plan_promql_to_post_asap(index, query, accuracy, backend_plan)?; + let node = plan_promql_to_post_asap(index, query, accuracy.clone(), backend_plan)?; let hints = execution_hints(&node); let t0_ms = if hints.full_history { 0 @@ -97,6 +106,8 @@ pub fn execute_post_asap_instant( let outcome = execute_planned_post_asap( index, &node, + query, + accuracy, t0_ms, now_ms, hints.cumulative_readout, @@ -108,26 +119,25 @@ pub fn execute_post_asap_instant( fn execute_planned_post_asap( index: &SketchStore, node: &planner_types::post_asap::SummaryNode, + query: &str, + accuracy: AccuracyTarget, t0_ms: u64, t1_ms: u64, is_cumulative: bool, backend_plan: Option<&control_plane::backend_plan::BackendPlan>, ) -> Result { + let allowed_materializations = match backend_plan { + Some(plan) => Some(resolve_materializations_for_post_asap( + plan, node, query, accuracy, + )?), + None => None, + }; let ctx = QueryExecutionContext { index, t0_ms, t1_ms, is_cumulative, - allowed_materializations: backend_plan.map(|plan| { - plan.routing - .iter() - .filter(|route| { - route.storage_backend - == control_plane::backend_plan::StorageBackend::SketchStore - }) - .map(|route| route.materialization) - .collect() - }), + allowed_materializations, }; match execute(node, &ctx) { diff --git a/data_plane/src/query_engines/asap_query_engine/summary_executor.rs b/data_plane/src/query_engines/asap_query_engine/summary_executor.rs index 4846ae73b..0be2b3348 100644 --- a/data_plane/src/query_engines/asap_query_engine/summary_executor.rs +++ b/data_plane/src/query_engines/asap_query_engine/summary_executor.rs @@ -396,11 +396,7 @@ impl<'a> SummaryExecutor for QueryExecutionContext<'a> { if self .allowed_materializations .as_ref() - .is_some_and(|allowed| { - !allowed.contains(&m.policy_fp) - && !(m.policy_fp == asap_types::PolicyFingerprint::UNSET - && allowed.len() == 1) - }) + .is_some_and(|allowed| !allowed.contains(&m.policy_fp)) { return None; } diff --git a/data_plane/src/storage_engines/types/hot_reload_config.rs b/data_plane/src/storage_engines/types/hot_reload_config.rs index 9f184bc4c..3a8c316d0 100644 --- a/data_plane/src/storage_engines/types/hot_reload_config.rs +++ b/data_plane/src/storage_engines/types/hot_reload_config.rs @@ -93,18 +93,21 @@ use crate::storage_engines::types::StreamingConfig; #[derive(Clone)] pub struct HotReloadBackendPlan { inner: Arc>, + install_lock: Arc>, } impl HotReloadBackendPlan { pub fn new(initial: control_plane::backend_plan::BackendPlan) -> Self { Self { inner: Arc::new(ArcSwap::new(Arc::new(initial))), + install_lock: Arc::new(std::sync::Mutex::new(())), } } pub fn from_arc(initial: Arc) -> Self { Self { inner: Arc::new(ArcSwap::new(initial)), + install_lock: Arc::new(std::sync::Mutex::new(())), } } @@ -128,7 +131,20 @@ impl HotReloadBackendPlan { Arc, control_plane::backend_plan::ValidationError, > { + let _guard = self + .install_lock + .lock() + .expect("backend-plan install lock poisoned"); new.validate()?; + let active = self.snapshot(); + if new.generated_at_unix_ms < active.generated_at_unix_ms { + return Err( + control_plane::backend_plan::ValidationError::StaleGeneration { + incoming: new.generated_at_unix_ms, + active: active.generated_at_unix_ms, + }, + ); + } Ok(self.swap(new)) } } @@ -200,6 +216,20 @@ mod hot_reload_backend_plan_tests { assert!(hr.install(invalid).is_err()); assert_eq!(hr.snapshot().plan_id, 1); } + + #[test] + fn older_generation_is_rejected_without_replacing_snapshot() { + let mut current = plan(2); + current.generated_at_unix_ms = 200; + let hr = HotReloadBackendPlan::new(current); + let mut stale = plan(3); + stale.generated_at_unix_ms = 199; + assert!(matches!( + hr.install(stale), + Err(control_plane::backend_plan::ValidationError::StaleGeneration { .. }) + )); + assert_eq!(hr.snapshot().plan_id, 2); + } } /// Thin wrapper around `ArcSwap` with ergonomic