From 3af64c85638ba01bae668a6b21d8d2c491e951a2 Mon Sep 17 00:00:00 2001 From: zz_y Date: Mon, 27 Jul 2026 13:07:37 -0600 Subject: [PATCH 1/2] feat(engine): serve from SummaryExecutor when it is provably safe MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 2 of the SummaryExecutor rollout (Phase 1: shadow-mode, PR #419). Adds the actual serving cutover, gated behind ASAP_SUMMARY_EXECUTOR_LIVE (default off): when a query lowers and executes cleanly and isn't the known ASAPController#163 grouping-ambiguity shape, engine.rs now serves the answer directly from SummaryExecutor and skips the legacy SketchReducer call entirely for that candidate. Every other case falls back exactly as today, so None here is indistinguishable from Phase 1. - summary_executor.rs: GroupState::exact_coverage gives ExactAgg the same coverage story SummaryValue already has (needed so live-serve can report ASAPTierResult.coverage regardless of which family answered). - l4_readout.rs (new): shared lowering + execution + conversion into ASAPTierResult's (series, coverage) shape, with the mechanical ambiguous_merge_risk gate (empty root by + >1 group). Both shadow_compare.rs and live_serve.rs now call this instead of duplicating the conversion logic. - live_serve.rs (new): the actual cutover — flag check, ambiguity gate, Some(...) means "use this instead of the legacy reducer." - engine.rs: wired into the range- and instant-query dispatch loops at the points where the legacy reducer is called; skips the redundant apply_outer_agg_fold and shadow-mode comparison when a candidate was already served live. Verified: full lib suite (955 tests) and the e2e suite pass identically with the flag on and off, including two new e2e tests proving the live path actually answers a DDSketch quantile and correctly falls back (via legacy) on the known ambiguous multi-HLL-sid count() shape. Co-Authored-By: Claude Sonnet 5 --- .../query_engines/asap_query_engine/engine.rs | 288 +++++++++------ .../asap_query_engine/l4_lowering.rs | 19 +- .../asap_query_engine/l4_readout.rs | 338 ++++++++++++++++++ .../asap_query_engine/live_serve.rs | 251 +++++++++++++ .../query_engines/asap_query_engine/mod.rs | 2 + .../asap_query_engine/shadow_compare.rs | 105 +----- .../asap_query_engine/summary_executor.rs | 78 ++++ ...e2e_controller_plans_and_backend_serves.rs | 213 +++++++++++ 8 files changed, 1094 insertions(+), 200 deletions(-) create mode 100644 data_plane/src/query_engines/asap_query_engine/l4_readout.rs create mode 100644 data_plane/src/query_engines/asap_query_engine/live_serve.rs diff --git a/data_plane/src/query_engines/asap_query_engine/engine.rs b/data_plane/src/query_engines/asap_query_engine/engine.rs index 570d51b1..94d32a94 100644 --- a/data_plane/src/query_engines/asap_query_engine/engine.rs +++ b/data_plane/src/query_engines/asap_query_engine/engine.rs @@ -692,6 +692,12 @@ impl ASAPQueryEngine { let reducer = crate::storage_engines::sketch_db::query::SketchReducer::new(idx); let mut combined_result: Option = None; + // Set when ANY candidate this loop was served directly from + // `SummaryExecutor` (see `live_serve.rs`) rather than the legacy + // reducer — gates the post-loop shadow-mode comparison below, + // since there's no legacy answer left to diff a live-served + // result against. + let mut any_live_served = false; for candidate in &analysis.candidates { // Resolve candidate → {sids} via the sid catalog. Schema- @@ -773,40 +779,59 @@ impl ASAPQueryEngine { // sub-window sums and divides by the range to produce // events-per-second. See `SketchReducer::evaluate_exact_agg` // / `evaluate_exact_agg_rate` for the per-path semantics. - let result = match &candidate.required_capability { - crate::storage_engines::sketch_db::index::Capability::ExactAgg(agg_type) => { - // Counter-function dispatch (issue #301) — mirror the - // instant `execute(&str)` path's branching off the - // typed `candidate.outer_fn`. On this explicit - // RANGE (matrix) surface the per-window timeseries is - // the correct shape for `sum`/`increase` (the wire - // format wants a point per window), so - // `accumulate_windows = false`. `rate` still folds + - // divides; `sum_over_time` over a counter is refused - // (decision (a)) so the query routes to archive. - use control_plane::asap_tier_analysis::OuterFn; - let is_exact_sum_family = matches!( + // Try serving this candidate directly from `SummaryExecutor` + // (see `live_serve.rs`) before falling back to the legacy + // reducer dispatch below. `None` here covers the flag being + // off, a rate-shaped candidate (self-excludes via + // `LoweringSkip::RateShape` — the legacy rate branch below + // is untouched for those), and every other "can't safely + // serve this way" outcome — all indistinguishable from + // Phase 1's shadow-only behavior. + let live_served_result = + crate::query_engines::asap_query_engine::live_serve::try_serve_from_summary_executor( + idx, query, start_ms, end_ms, false, + ); + let served_live = live_served_result.is_some(); + if served_live { + any_live_served = true; + } + + let result = match live_served_result { + Some(result) => result, + None => match &candidate.required_capability { + crate::storage_engines::sketch_db::index::Capability::ExactAgg(agg_type) => { + // Counter-function dispatch (issue #301) — mirror the + // instant `execute(&str)` path's branching off the + // typed `candidate.outer_fn`. On this explicit + // RANGE (matrix) surface the per-window timeseries is + // the correct shape for `sum`/`increase` (the wire + // format wants a point per window), so + // `accumulate_windows = false`. `rate` still folds + + // divides; `sum_over_time` over a counter is refused + // (decision (a)) so the query routes to archive. + use control_plane::asap_tier_analysis::OuterFn; + let is_exact_sum_family = matches!( agg_type, crate::storage_engines::sketch_db::data::AggregationType::Sum | crate::storage_engines::sketch_db::data::AggregationType::MultipleSum | crate::storage_engines::sketch_db::data::AggregationType::Increase | crate::storage_engines::sketch_db::data::AggregationType::MultipleIncrease ); - if is_exact_sum_family && candidate.outer_fn == OuterFn::SumOverTime { - return Err(crate::query_engines::EngineError::capability_miss( - crate::storage_engines::types::StorageBackend::SketchStore - .data_source_id(), - format!( - "SketchStore cannot answer `sum_over_time` over counter \ + if is_exact_sum_family && candidate.outer_fn == OuterFn::SumOverTime { + return Err(crate::query_engines::EngineError::capability_miss( + crate::storage_engines::types::StorageBackend::SketchStore + .data_source_id(), + format!( + "SketchStore cannot answer `sum_over_time` over counter \ deltas for `{query}` (issue #301) — failing over to archive" - ), - )); - } - let use_rate_path = candidate.range_seconds > 0 - && candidate.outer_fn == OuterFn::Rate - && is_exact_sum_family; - if use_rate_path { - reducer + ), + )); + } + let use_rate_path = candidate.range_seconds > 0 + && candidate.outer_fn == OuterFn::Rate + && is_exact_sum_family; + if use_rate_path { + reducer .evaluate_exact_agg_rate( &hit_sids, *agg_type, @@ -824,63 +849,70 @@ impl ASAPQueryEngine { ), ) })? - } else { - reducer - .evaluate_exact_agg( - &hit_sids, - *agg_type, - &candidate.group_by_keys, - start_ms, - end_ms, - false, - ) - .map_err(|e| { - crate::query_engines::EngineError::capability_miss( - crate::storage_engines::types::StorageBackend::SketchStore - .data_source_id(), - format!( + } else { + reducer + .evaluate_exact_agg( + &hit_sids, + *agg_type, + &candidate.group_by_keys, + start_ms, + end_ms, + false, + ) + .map_err(|e| { + crate::query_engines::EngineError::capability_miss( + crate::storage_engines::types::StorageBackend::SketchStore + .data_source_id(), + format!( "SketchStore exact-agg reducer failed for `{query}` over \ [{start_ms}, {end_ms}]: {e:?} — failing over to archive" ), - ) - })? + ) + })? + } } - } - // P2-4 (typed dispatch): route off the analyzer's typed - // `required_capability` via `evaluate_for_capability` - // instead of round-tripping it through a function-name - // string the reducer re-parses. - _ => reducer - .evaluate_for_capability( - &candidate.required_capability, - &hit_sids, - &candidate.function_args, - // Per-item CMS estimate(key) is wired through the reducer - // but only dispatched once the engine resolves the item - // value against an item_label-mode sid (Phase 2b). Until - // then keyed CMS frequency safe-misses (see below), so the - // bucket-total path is correct here. - None, - effective_is_cumulative(candidate), - start_ms, - end_ms, - ) - .map_err(|e| { - crate::query_engines::EngineError::capability_miss( - crate::storage_engines::types::StorageBackend::SketchStore - .data_source_id(), - format!( - "SketchStore reducer failed for `{query}` over \ - [{start_ms}, {end_ms}]: {e:?} — failing over to archive" - ), + // P2-4 (typed dispatch): route off the analyzer's typed + // `required_capability` via `evaluate_for_capability` + // instead of round-tripping it through a function-name + // string the reducer re-parses. + _ => reducer + .evaluate_for_capability( + &candidate.required_capability, + &hit_sids, + &candidate.function_args, + // Per-item CMS estimate(key) is wired through the reducer + // but only dispatched once the engine resolves the item + // value against an item_label-mode sid (Phase 2b). Until + // then keyed CMS frequency safe-misses (see below), so the + // bucket-total path is correct here. + None, + effective_is_cumulative(candidate), + start_ms, + end_ms, ) - })?, + .map_err(|e| { + crate::query_engines::EngineError::capability_miss( + crate::storage_engines::types::StorageBackend::SketchStore + .data_source_id(), + format!( + "SketchStore reducer failed for `{query}` over \ + [{start_ms}, {end_ms}]: {e:?} — failing over to archive" + ), + ) + })?, + }, }; // Apply the analyzer's typed outer-aggregation operator on // the range-query path too (issue #296) — same identity // case + fold semantics as the instant-query trait - // adapter above. - let result = if candidate.outer_agg.is_some() && !outer_fold_already_consumed(candidate) + // adapter above. Skipped when `SummaryExecutor` already + // served this candidate: `bind_query_expr`'s lowering + // already realizes the full aggregation (including any + // `by (...)`) into the `L4Node` it executed, so re-folding + // here would double-apply it. + let result = if !served_live + && candidate.outer_agg.is_some() + && !outer_fold_already_consumed(candidate) { apply_outer_agg_fold(result, &candidate.outer_agg) } else { @@ -899,10 +931,15 @@ impl ASAPQueryEngine { // Shadow-mode comparison against the new SummaryExecutor path — // see `data_plane/docs/l4node-plan-executor-design.md`'s // "Rollout" section. No-op unless `ASAP_SHADOW_SUMMARY_EXECUTOR` - // is set; never affects `result`/the response below. - crate::query_engines::asap_query_engine::shadow_compare::maybe_shadow_compare( - idx, query, start_ms, end_ms, false, &result, - ); + // is set; never affects `result`/the response below. Skipped + // entirely when `result` was already served BY SummaryExecutor + // (`any_live_served`) — there's no separate legacy answer left + // to diff it against. + if !any_live_served { + crate::query_engines::asap_query_engine::shadow_compare::maybe_shadow_compare( + idx, query, start_ms, end_ms, false, &result, + ); + } // Matrix shape — the range_query wire format requires it. let warm_qr = asap_tier_result_to_query_result(result.clone(), end_ms, true); @@ -1395,6 +1432,12 @@ impl crate::query_engines::routing::query_engine_routing::QueryEngine for ASAPQu // vector`. Returning `Matrix` for an instant query // produces a 500 (adapter rejects the shape mismatch). let mut any_range_candidate = false; + // Set when ANY candidate this loop was served directly from + // `SummaryExecutor` (see `live_serve.rs`) rather than the + // legacy reducer — gates the post-loop shadow-mode + // comparison below, since there's no legacy answer left to + // diff a live-served result against. + let mut any_live_served = false; // Snapshot the streaming config once for this query's // policy lookups. Hot-reload swaps the underlying Arc; the @@ -1722,8 +1765,49 @@ impl crate::query_engines::routing::query_engine_routing::QueryEngine for ASAPQu combined_t0 = t0_ms; } - let reducer_result = - match &candidate.required_capability { + // Try serving this candidate directly from + // `SummaryExecutor` (see `live_serve.rs`) before falling + // back to the legacy reducer dispatch below. Skipped + // outright when the frequency-rate fallback already + // answered (`freq_rate_override`) — that path runs + // against an empty `hit_sids` sid set that the analyzer + // itself couldn't satisfy directly, so there's nothing + // for the new path to re-derive from the raw query + // that would be any more meaningful. `None` otherwise + // covers the flag being off, a rate-shaped candidate + // (self-excludes via `LoweringSkip::RateShape` — the + // legacy rate branch below is untouched for those), and + // every other "can't safely serve this way" outcome. + let live_served_result = if freq_rate_override.is_none() { + crate::query_engines::asap_query_engine::live_serve::try_serve_from_summary_executor( + idx, + query, + t0_ms, + now_ms, + effective_is_cumulative(candidate), + ) + } else { + None + }; + let served_live = live_served_result.is_some(); + if served_live { + any_live_served = true; + } + + // P1-1: if the frequency-rate fallback produced a result + // (hit_sids was empty for an ExactAgg(Sum)+Rate candidate + // but a warm FrequencyEstimate sid answered), use it + // directly; the ExactAgg dispatch below would run + // against an empty `hit_sids` and is moot. Otherwise, if + // `SummaryExecutor` already served this candidate, use + // that. Only compute + dispatch the legacy reducer when + // neither of the above applies. + let result = if let Some(r) = freq_rate_override { + r + } else if let Some(r) = live_served_result { + r + } else { + let reducer_result = match &candidate.required_capability { crate::storage_engines::sketch_db::index::Capability::ExactAgg( agg_type, ) if use_rate_path => reducer.evaluate_exact_agg_rate( @@ -1776,15 +1860,6 @@ impl crate::query_engines::routing::query_engine_routing::QueryEngine for ASAPQu now_ms, ), }; - - // P1-1: if the frequency-rate fallback produced a result - // (hit_sids was empty for an ExactAgg(Sum)+Rate candidate - // but a warm FrequencyEstimate sid answered), use it - // directly; the ExactAgg dispatch above ran against an - // empty `hit_sids` and is moot. - let result = if let Some(r) = freq_rate_override { - r - } else { match reducer_result { Ok(r) => r, Err( @@ -1862,12 +1937,20 @@ impl crate::query_engines::routing::query_engine_routing::QueryEngine for ASAPQu // a single-value group, returning the same value // unchanged. No special case needed; the general fold // handles it. - let result = - if candidate.outer_agg.is_some() && !outer_fold_already_consumed(candidate) { - apply_outer_agg_fold(result, &candidate.outer_agg) - } else { - result - }; + // + // Skipped when `SummaryExecutor` already served this + // candidate (`served_live`): `bind_query_expr`'s + // lowering already realizes the full aggregation + // (including any `by (...)`) into the `L4Node` it + // executed, so re-folding here would double-apply it. + let result = if !served_live + && candidate.outer_agg.is_some() + && !outer_fold_already_consumed(candidate) + { + apply_outer_agg_fold(result, &candidate.outer_agg) + } else { + result + }; combined_result = Some(result); } @@ -1901,10 +1984,15 @@ impl crate::query_engines::routing::query_engine_routing::QueryEngine for ASAPQu // `data_plane/docs/l4node-plan-executor-design.md`'s // "Rollout" section. No-op unless // `ASAP_SHADOW_SUMMARY_EXECUTOR` is set; never affects - // `result`/the response below. - crate::query_engines::asap_query_engine::shadow_compare::maybe_shadow_compare( - idx, query, stitch_t0, now_ms, true, &result, - ); + // `result`/the response below. Skipped entirely when + // `result` was already served BY SummaryExecutor + // (`any_live_served`) — there's no separate legacy + // answer left to diff it against. + if !any_live_served { + crate::query_engines::asap_query_engine::shadow_compare::maybe_shadow_compare( + idx, query, stitch_t0, now_ms, true, &result, + ); + } let warm_qr = asap_tier_result_to_query_result(result.clone(), now_ms, false); if let (Some((cov_lo, cov_hi)), Some(archive)) = (result.coverage, self.archive_engine.as_ref()) diff --git a/data_plane/src/query_engines/asap_query_engine/l4_lowering.rs b/data_plane/src/query_engines/asap_query_engine/l4_lowering.rs index 56f0364d..035a254c 100644 --- a/data_plane/src/query_engines/asap_query_engine/l4_lowering.rs +++ b/data_plane/src/query_engines/asap_query_engine/l4_lowering.rs @@ -1,5 +1,6 @@ -//! PromQL string → `asap_sketch::L4Node` bridge for shadow-mode -//! `SummaryExecutor` comparison. See +//! PromQL string → `asap_sketch::L4Node` bridge, shared by both +//! shadow-mode comparison (`shadow_compare.rs`) and the actual serving +//! cutover (`live_serve.rs`, via `l4_readout.rs`). See //! `data_plane/docs/l4node-plan-executor-design.md`'s "Rollout" section //! for the full design and why this calls //! `control_plane::sketch_algebra::lower::bind_query_expr` @@ -20,10 +21,13 @@ use control_plane::sketch_algebra::capability::OuterFn; use control_plane::sketch_algebra::{BindingError, L4Plan, PhysicalExpr}; use control_plane::types_v2::AccuracyTarget; -/// Why `lower_promql_to_l4node` didn't produce a comparable `L4Node`. +/// Why a query couldn't be answered through the `L4Node`/`SummaryExecutor` +/// path — covers both `lower_promql_to_l4node`'s own failure to produce a +/// tree, AND (via `l4_readout.rs`'s `execute_l4_readout`) a failure of +/// `asap_sketch::exec::execute()` on a tree that DID lower successfully. /// None of these are errors in the alarming sense — every variant is an /// expected, frequent outcome for *some* fraction of live traffic; the -/// caller's only obligation is "don't attempt a shadow comparison," never +/// caller's only obligation is "fall back to the legacy path," never /// "log this as a problem." #[derive(Debug)] pub enum LoweringSkip { @@ -60,6 +64,13 @@ pub enum LoweringSkip { /// outcome, just detected one step earlier so the caller can skip /// without even constructing a `QueryExecutionContext`. NotRealized, + /// The tree lowered successfully, but `asap_sketch::exec::execute()` + /// itself returned `Err` (`NoCandidates`, `MergeKindParamsMismatch`, + /// a decode/merge failure surfaced from `summary_executor.rs`, ...). + /// Always safe to just fall back — this means "can't answer this way + /// right now" (e.g. the sid catalog doesn't have an exact + /// `(SummaryKind, SummaryParams)` match), never "answered wrong." + ExecuteFailed(String), } /// Lower a raw PromQL query string to the `L4Node` tree diff --git a/data_plane/src/query_engines/asap_query_engine/l4_readout.rs b/data_plane/src/query_engines/asap_query_engine/l4_readout.rs new file mode 100644 index 00000000..63f9f6d4 --- /dev/null +++ b/data_plane/src/query_engines/asap_query_engine/l4_readout.rs @@ -0,0 +1,338 @@ +//! Shared `L4Node` lowering + execution + conversion into +//! `ASAPTierResult`'s `(series, coverage)` shape — the common core of both +//! `shadow_compare.rs` (diagnostic only, never affects serving) and +//! `live_serve.rs` (the actual cutover). See +//! `data_plane/docs/l4node-plan-executor-design.md` for the design. + +use std::collections::BTreeMap; + +use asap_sketch::exec::{execute, ExecOutcome}; +use asap_sketch::{L4Node, SummaryExpr}; +use control_plane::types_v2::AccuracyTarget; + +use crate::query_engines::asap_query_engine::l4_lowering::{lower_promql_to_l4node, LoweringSkip}; +use crate::query_engines::asap_query_engine::summary_executor::{ + QueryExecutionContext, SummaryValue, +}; +use crate::storage_engines::sketch_db::index::SketchStore; + +/// Mirrors `ASAPTierResult.series`'s row shape — `(label_values, samples)` +/// where `samples` is `(window_end_unix_ms, value)`. +pub type SeriesRows = Vec<(BTreeMap, Vec<(i64, f64)>)>; + +/// The result of lowering + executing a query through +/// `SummaryExecutor`, converted into the same shape `ASAPTierResult` +/// uses, regardless of whether the answer came from the sketch +/// (`ExecOutcome::Value`) or `ExactAgg` (`ExecOutcome::State`) side — +/// callers that only care about "did this answer the query, and is it +/// safe to trust" don't need to know which. +pub struct L4ReadoutOutcome { + pub series: SeriesRows, + pub coverage: Option<(u64, u64)>, + /// `true` only for a sketch-family (`ExecOutcome::Value`) outcome + /// whose tree's root `SummaryAgg` had an empty `by` AND produced more + /// than one group. This is exactly the ambiguous shape + /// [ASAPController#163](https://github.com/ProjectASAP/ASAPController/issues/163) + /// describes — an empty `by` is indistinguishable between "no + /// grouping concept applies" (this group split is correct) and "an + /// aggregation operator asked to reduce everything" (these groups + /// should have been merged into one). Always `false` for `ExactAgg` + /// (`Sum`/`Increase` map only from genuine aggregation operators, so + /// their empty `by` is unambiguous — see the design doc's "Grouping + /// semantics"/"ExactAgg" sections) and for any sketch outcome with a + /// non-empty `by` or ≤1 resulting group (nothing to disagree about). + pub ambiguous_merge_risk: bool, +} + +/// Lower `query`, execute it against `index` over `[t0_ms, t1_ms]`, and +/// convert the result into `L4ReadoutOutcome`. `Err` covers every reason +/// this couldn't produce a trustworthy answer — see `LoweringSkip`'s +/// variants; every one of them means "fall back to the legacy path," +/// never "the legacy path is wrong." +pub fn execute_l4_readout( + index: &SketchStore, + query: &str, + t0_ms: u64, + t1_ms: u64, + is_cumulative: bool, + accuracy: AccuracyTarget, +) -> Result { + let node = lower_promql_to_l4node(query, accuracy)?; + let by_is_empty = root_summary_agg_by_is_empty(&node); + + let ctx = QueryExecutionContext { + index, + t0_ms, + t1_ms, + is_cumulative, + }; + + match execute(&node, &ctx) { + Ok(ExecOutcome::Value(values)) => { + let mut coverage: Option<(u64, u64)> = None; + let mut series = Vec::new(); + for (group_key, value) in &values { + fold_coverage(&mut coverage, value.coverage()); + series.extend(summary_value_to_series(group_key, value)); + } + let ambiguous_merge_risk = by_is_empty.unwrap_or(false) && values.len() > 1; + Ok(L4ReadoutOutcome { + series, + coverage, + ambiguous_merge_risk, + }) + } + Ok(ExecOutcome::State(groups)) => { + let mut coverage: Option<(u64, u64)> = None; + let mut series = Vec::new(); + for (group_key, state, _kind, _params) in &groups { + fold_coverage(&mut coverage, state.exact_coverage()); + let Some(value) = state.exact_value(&None) else { + continue; + }; + series.push((group_key.clone(), vec![(t1_ms as i64, value)])); + } + Ok(L4ReadoutOutcome { + series, + coverage, + ambiguous_merge_risk: false, + }) + } + Err(e) => Err(LoweringSkip::ExecuteFailed(format!("{e:?}"))), + } +} + +/// Walk down to the tree's `SummaryAgg` node (through a `SummaryEstimate` +/// wrapper if present, and through the first child of a `SummaryMerge` — +/// its children agree on `(SummaryKind, SummaryParams)` by construction, +/// so their `by` agrees too) and report whether its `by` list is empty. +/// `None` for a bare `Logical` root (shouldn't happen here — +/// `lower_promql_to_l4node` already rejects that via `NotRealized` — kept +/// exhaustive and defensive rather than assumed unreachable). +fn root_summary_agg_by_is_empty(node: &L4Node) -> Option { + match &node.expr { + SummaryExpr::SummaryAgg { by, .. } => Some(by.is_empty()), + SummaryExpr::SummaryEstimate { sketch_input, .. } => { + root_summary_agg_by_is_empty(sketch_input) + } + SummaryExpr::SummaryMerge { children } => children + .first() + .and_then(|c| root_summary_agg_by_is_empty(c)), + SummaryExpr::Logical(_) => None, + _ => None, + } +} + +/// `SummaryValue::Points`/`TopK` -> `ASAPTierResult.series`'s row shape. +/// `TopK`'s ranked-list-per-timestamp shape is pivoted into one row per +/// item (each row = the group's label map plus an `item` label, one point +/// per timestamp that item appeared in the ranked list) -- the SAME +/// convention `sketch_reducer.rs`'s own topk arm already uses, not a new +/// one invented here. +fn summary_value_to_series( + group_key: &BTreeMap, + value: &SummaryValue, +) -> SeriesRows { + match value { + SummaryValue::Points(points, _coverage) => { + vec![(group_key.clone(), points.clone())] + } + SummaryValue::TopK(ranked_per_ts, _coverage) => { + let mut by_item: BTreeMap> = BTreeMap::new(); + for (ts, items) in ranked_per_ts { + for (item, val) in items { + by_item.entry(item.clone()).or_default().push((*ts, *val)); + } + } + by_item + .into_iter() + .map(|(item, points)| { + let mut lv = group_key.clone(); + lv.insert("item".to_string(), item); + (lv, points) + }) + .collect() + } + } +} + +pub(crate) fn fold_coverage(coverage: &mut Option<(u64, u64)>, next: Option<(u64, u64)>) { + let Some((lo, hi)) = next else { return }; + *coverage = Some(match *coverage { + Some((clo, chi)) => (clo.min(lo), chi.max(hi)), + None => (lo, hi), + }); +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::storage_engines::sketch_db::data::{AggKind, SketchConfig}; + use crate::storage_engines::sketch_db::index::{ + AccuracyBound, Capability, SketchInstanceMetadata, SketchKindHandle, SketchSampleState, + }; + + fn accuracy() -> AccuracyTarget { + AccuracyTarget::Epsilon(0.01) + } + + fn register_hll(idx: &SketchStore, sid: u64, service: &str, items: &[&str]) { + // precision 14 -- what `ControlPlaneCostModel` actually picks for + // `AccuracyTarget::Epsilon(0.01)` (confirmed by inspecting the + // bound tree directly); `find_candidates`'s exact-match contract + // means a mismatched precision here would just silently produce + // `NoCandidates`, not a wrong answer -- but that's not what these + // tests are checking. + let cfg = SketchConfig::Hll { precision: 14 }; + let mut group_by_keys = std::collections::BTreeSet::new(); + group_by_keys.insert("service".to_string()); + idx.register(SketchInstanceMetadata { + sid, + metric_name: "unique_users".to_string(), + group_by_keys, + capability: Some(Capability::CardinalityApprox), + agg_kind: AggKind::Sketch { + kind: SketchKindHandle::Hll, + config: cfg.clone(), + spatial_filter_canonical: String::new(), + }, + accuracy: Some(AccuracyBound::from_config(&cfg)), + first_seen_unix_ms: 0, + retired_at_ms: None, + expires_at_ms: None, + policy_fp: asap_types::PolicyFingerprint::UNSET, + }); + use asap_sketchlib::{HllSketch, HllVariant, MessagePackCodec}; + let mut sk = HllSketch::new(HllVariant::Regular, 14); + for item in items { + sk.update(item.as_bytes()); + } + let mut labels = BTreeMap::new(); + labels.insert("service".to_string(), service.to_string()); + idx.append_sample( + sid, + labels, + (1_000, 2_000), + SketchSampleState { + bytes: sk.to_msgpack().expect("encode HLL"), + encoding: crate::storage_engines::sketch_db::index::SketchEncoding::MsgpackFull, + }, + ); + } + + fn ddsketch_fixture() -> SketchStore { + let idx = SketchStore::new(); + // alpha 0.01 -- what `ControlPlaneCostModel` actually picks for + // `quantile_over_time` at `AccuracyTarget::Epsilon(0.01)` (DDSketch, + // not KLL -- confirmed by inspecting the bound tree directly). + let cfg = SketchConfig::DDSketch { + relative_accuracy: 0.01, + }; + idx.register(SketchInstanceMetadata { + sid: 1, + metric_name: "latency_ms".to_string(), + group_by_keys: std::collections::BTreeSet::new(), + capability: Some(Capability::QuantileApprox(SketchKindHandle::DDSketch)), + agg_kind: AggKind::Sketch { + kind: SketchKindHandle::DDSketch, + config: cfg.clone(), + spatial_filter_canonical: String::new(), + }, + accuracy: Some(AccuracyBound::from_config(&cfg)), + first_seen_unix_ms: 0, + retired_at_ms: None, + expires_at_ms: None, + policy_fp: asap_types::PolicyFingerprint::UNSET, + }); + use asap_sketchlib::{DdSketch, MessagePackCodec}; + let mut sk = DdSketch::new(0.01); + for i in 1..=100 { + sk.update(i as f64); + } + idx.append_sample( + 1, + BTreeMap::new(), + (1_000, 2_000), + SketchSampleState { + bytes: sk.to_msgpack().expect("encode DDSketch"), + encoding: crate::storage_engines::sketch_db::index::SketchEncoding::MsgpackFull, + }, + ); + idx + } + + #[test] + fn unambiguous_sketch_query_is_not_flagged() { + let idx = ddsketch_fixture(); + let outcome = execute_l4_readout( + &idx, + "quantile_over_time(0.99, latency_ms[1m])", + 1_000, + 2_000, + true, + accuracy(), + ) + .expect("should execute"); + assert!( + !outcome.ambiguous_merge_risk, + "a single-series bare range function must not be flagged ambiguous" + ); + assert_eq!(outcome.series.len(), 1); + } + + #[test] + fn ambiguous_global_merge_shape_is_flagged() { + // The exact ASAPController#163 shape: two HLL sids, no explicit + // by(), an aggregation-operator query -- find_candidates can't + // tell whether these two groups should have been merged. + let idx = SketchStore::new(); + register_hll(&idx, 1, "svc-a", &["a", "b", "c"]); + register_hll(&idx, 2, "svc-b", &["d", "e", "f"]); + let outcome = + execute_l4_readout(&idx, "count(unique_users)", 1_000, 2_000, true, accuracy()) + .expect("should execute"); + assert!( + outcome.ambiguous_merge_risk, + "two distinct-service HLL groups under a by-less count() must be flagged, got {:?}", + outcome.series + ); + assert_eq!(outcome.series.len(), 2); + } + + #[test] + fn exact_agg_outcome_is_never_flagged_ambiguous() { + let idx = SketchStore::new(); + idx.register( + crate::storage_engines::sketch_db::index::SketchInstanceMetadata { + sid: 1, + metric_name: "bytes_total".to_string(), + group_by_keys: std::collections::BTreeSet::new(), + capability: Some(Capability::ExactAgg(asap_types::AggregationType::Sum)), + agg_kind: AggKind::ExactAgg { + agg_type: asap_types::AggregationType::Sum, + parameters_canonical: String::new(), + spatial_filter_canonical: String::new(), + }, + accuracy: None, + first_seen_unix_ms: 0, + retired_at_ms: None, + expires_at_ms: None, + policy_fp: asap_types::PolicyFingerprint::UNSET, + }, + ); + idx.append_precompute( + 1, + BTreeMap::new(), + (1_000, 2_000), + Box::new(crate::precompute_engine::operators::SumAccumulator::with_sum(42.0)), + ); + let outcome = execute_l4_readout(&idx, "sum(bytes_total)", 1_000, 2_000, true, accuracy()) + .expect("should execute"); + assert!(!outcome.ambiguous_merge_risk); + // Window-end-only coverage: a single window (1_000, 2_000) is + // keyed by its end (2_000) alone, so both bounds equal 2_000 -- + // same semantics as `SummaryValue::coverage()`, reconfirmed for + // `exact_coverage` by this module's A0 test in `summary_executor.rs`. + assert_eq!(outcome.coverage, Some((2_000, 2_000))); + } +} diff --git a/data_plane/src/query_engines/asap_query_engine/live_serve.rs b/data_plane/src/query_engines/asap_query_engine/live_serve.rs new file mode 100644 index 00000000..f0c52397 --- /dev/null +++ b/data_plane/src/query_engines/asap_query_engine/live_serve.rs @@ -0,0 +1,251 @@ +//! The actual `SummaryExecutor` serving cutover — unlike +//! `shadow_compare.rs` (diagnostic only, never affects what's served), +//! `try_serve_from_summary_executor` returning `Some(...)` means the +//! caller uses THIS answer instead of calling the legacy +//! `SketchReducer` path. See +//! `data_plane/docs/l4node-plan-executor-design.md` and the Phase 2 +//! plan's "What 'safe to serve' means, precisely" section for the exact +//! gate this applies. + +use control_plane::types_v2::AccuracyTarget; + +use crate::query_engines::asap_query_engine::l4_readout::execute_l4_readout; +use crate::storage_engines::sketch_db::index::SketchStore; +use crate::storage_engines::sketch_db::query::ASAPTierResult; + +/// Fixed accuracy target for this phase — mirrors +/// `shadow_compare::SHADOW_ACCURACY`; `data_plane` doesn't carry a +/// per-workload `AccuracyTarget` today (see the design doc's "Rollout" +/// section). +const LIVE_ACCURACY: AccuracyTarget = AccuracyTarget::Epsilon(0.01); + +/// Whether the actual serving cutover is enabled for this process. +/// Mirrors `shadow_compare::shadow_summary_executor_enabled`'s exact +/// mechanics, own flag, own default (off) — this is a materially +/// riskier switch than shadow mode (it changes what's served, not just +/// what's logged), so it must never be implied by the shadow flag. +pub fn summary_executor_live_enabled() -> bool { + std::env::var("ASAP_SUMMARY_EXECUTOR_LIVE") + .map(|v| { + let v = v.trim(); + v == "1" || v.eq_ignore_ascii_case("true") || v.eq_ignore_ascii_case("on") + }) + .unwrap_or(false) +} + +/// Try to serve `query` entirely from `SummaryExecutor`. Returns `None` +/// whenever the caller should fall back to the legacy path exactly as +/// it does today (flag off, lowering/execution failed, or the +/// grouping-ambiguity gate tripped — see `L4ReadoutOutcome::ambiguous_merge_risk`'s +/// doc) — `None` here is indistinguishable from Phase 1's shadow-only +/// behavior. `Some(...)` means the new path answered and the caller +/// must NOT also call the legacy reducer for this candidate. +pub fn try_serve_from_summary_executor( + index: &SketchStore, + query: &str, + t0_ms: u64, + t1_ms: u64, + is_cumulative: bool, +) -> Option { + if !summary_executor_live_enabled() { + return None; + } + + let outcome = match execute_l4_readout(index, query, t0_ms, t1_ms, is_cumulative, LIVE_ACCURACY) + { + Ok(outcome) => outcome, + Err(skip) => { + tracing::debug!( + query, + ?skip, + "live: query not servable from SummaryExecutor, falling back" + ); + return None; + } + }; + + if outcome.ambiguous_merge_risk { + tracing::debug!( + query, + "live: ambiguous global-merge shape (ASAPController#163), falling back to legacy path" + ); + return None; + } + + tracing::debug!(query, "live: served from SummaryExecutor"); + Some(ASAPTierResult { + series: outcome.series, + coverage: outcome.coverage, + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::collections::BTreeMap; + + use crate::storage_engines::sketch_db::data::{AggKind, SketchConfig}; + use crate::storage_engines::sketch_db::index::{ + AccuracyBound, Capability, SketchInstanceMetadata, SketchKindHandle, SketchSampleState, + }; + + /// Mirrors `shadow_compare.rs`'s `ENV_VAR_LOCK`/`ShadowEnvGuard` + /// pattern exactly, own env var — `std::env::set_var`/`remove_var` + /// mutate process-global state and `cargo test` runs this module's + /// tests on multiple threads in the same process. + static ENV_VAR_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(()); + + #[allow(dead_code)] + struct LiveEnvGuard(std::sync::MutexGuard<'static, ()>); + + impl Drop for LiveEnvGuard { + fn drop(&mut self) { + std::env::remove_var("ASAP_SUMMARY_EXECUTOR_LIVE"); + } + } + + fn set_live_env(value: &str) -> LiveEnvGuard { + let guard = ENV_VAR_LOCK + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + std::env::set_var("ASAP_SUMMARY_EXECUTOR_LIVE", value); + LiveEnvGuard(guard) + } + + fn clear_live_env() -> LiveEnvGuard { + let guard = ENV_VAR_LOCK + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + std::env::remove_var("ASAP_SUMMARY_EXECUTOR_LIVE"); + LiveEnvGuard(guard) + } + + fn ddsketch_fixture() -> SketchStore { + let idx = SketchStore::new(); + let cfg = SketchConfig::DDSketch { + relative_accuracy: 0.01, + }; + idx.register(SketchInstanceMetadata { + sid: 1, + metric_name: "latency_ms".to_string(), + group_by_keys: std::collections::BTreeSet::new(), + capability: Some(Capability::QuantileApprox(SketchKindHandle::DDSketch)), + agg_kind: AggKind::Sketch { + kind: SketchKindHandle::DDSketch, + config: cfg.clone(), + spatial_filter_canonical: String::new(), + }, + accuracy: Some(AccuracyBound::from_config(&cfg)), + first_seen_unix_ms: 0, + retired_at_ms: None, + expires_at_ms: None, + policy_fp: asap_types::PolicyFingerprint::UNSET, + }); + use asap_sketchlib::{DdSketch, MessagePackCodec}; + let mut sk = DdSketch::new(0.01); + for i in 1..=100 { + sk.update(i as f64); + } + idx.append_sample( + 1, + BTreeMap::new(), + (1_000, 2_000), + SketchSampleState { + bytes: sk.to_msgpack().expect("encode DDSketch"), + encoding: crate::storage_engines::sketch_db::index::SketchEncoding::MsgpackFull, + }, + ); + idx + } + + fn register_hll(idx: &SketchStore, sid: u64, service: &str, items: &[&str]) { + let cfg = SketchConfig::Hll { precision: 14 }; + let mut group_by_keys = std::collections::BTreeSet::new(); + group_by_keys.insert("service".to_string()); + idx.register(SketchInstanceMetadata { + sid, + metric_name: "unique_users".to_string(), + group_by_keys, + capability: Some(Capability::CardinalityApprox), + agg_kind: AggKind::Sketch { + kind: SketchKindHandle::Hll, + config: cfg.clone(), + spatial_filter_canonical: String::new(), + }, + accuracy: Some(AccuracyBound::from_config(&cfg)), + first_seen_unix_ms: 0, + retired_at_ms: None, + expires_at_ms: None, + policy_fp: asap_types::PolicyFingerprint::UNSET, + }); + use asap_sketchlib::{HllSketch, HllVariant, MessagePackCodec}; + let mut sk = HllSketch::new(HllVariant::Regular, 14); + for item in items { + sk.update(item.as_bytes()); + } + let mut labels = BTreeMap::new(); + labels.insert("service".to_string(), service.to_string()); + idx.append_sample( + sid, + labels, + (1_000, 2_000), + SketchSampleState { + bytes: sk.to_msgpack().expect("encode HLL"), + encoding: crate::storage_engines::sketch_db::index::SketchEncoding::MsgpackFull, + }, + ); + } + + #[test] + fn flag_off_never_serves() { + let _guard = clear_live_env(); + let idx = ddsketch_fixture(); + let result = try_serve_from_summary_executor( + &idx, + "quantile_over_time(0.99, latency_ms[1m])", + 1_000, + 2_000, + true, + ); + assert!(result.is_none(), "flag off must never serve"); + } + + #[test] + fn flag_on_safe_shape_serves() { + let _guard = set_live_env("1"); + let idx = ddsketch_fixture(); + let result = try_serve_from_summary_executor( + &idx, + "quantile_over_time(0.99, latency_ms[1m])", + 1_000, + 2_000, + true, + ); + let result = result.expect("unambiguous single-series quantile must serve"); + assert_eq!(result.series.len(), 1); + assert!(!result.is_empty()); + } + + #[test] + fn flag_on_ambiguous_shape_falls_back() { + let _guard = set_live_env("1"); + let idx = SketchStore::new(); + register_hll(&idx, 1, "svc-a", &["a", "b", "c"]); + register_hll(&idx, 2, "svc-b", &["d", "e", "f"]); + let result = + try_serve_from_summary_executor(&idx, "count(unique_users)", 1_000, 2_000, true); + assert!( + result.is_none(), + "ambiguous global-merge shape must fall back to legacy, not serve a possibly-wrong answer" + ); + } + + #[test] + fn flag_on_unservable_query_falls_back() { + let _guard = set_live_env("1"); + let idx = SketchStore::new(); + let result = + try_serve_from_summary_executor(&idx, "rate(http_requests_total[5m])", 0, 1000, true); + assert!(result.is_none()); + } +} diff --git a/data_plane/src/query_engines/asap_query_engine/mod.rs b/data_plane/src/query_engines/asap_query_engine/mod.rs index 5ff8b3ae..e75d9b02 100644 --- a/data_plane/src/query_engines/asap_query_engine/mod.rs +++ b/data_plane/src/query_engines/asap_query_engine/mod.rs @@ -11,6 +11,8 @@ pub mod engine; pub mod l4_lowering; +pub mod l4_readout; +pub mod live_serve; pub mod shadow_compare; pub mod summary_executor; diff --git a/data_plane/src/query_engines/asap_query_engine/shadow_compare.rs b/data_plane/src/query_engines/asap_query_engine/shadow_compare.rs index 30dfc9ea..3669461b 100644 --- a/data_plane/src/query_engines/asap_query_engine/shadow_compare.rs +++ b/data_plane/src/query_engines/asap_query_engine/shadow_compare.rs @@ -8,12 +8,7 @@ use std::collections::BTreeMap; -use asap_sketch::exec::{execute, ExecOutcome}; - -use crate::query_engines::asap_query_engine::l4_lowering::lower_promql_to_l4node; -use crate::query_engines::asap_query_engine::summary_executor::{ - QueryExecutionContext, SummaryValue, -}; +use crate::query_engines::asap_query_engine::l4_readout::{execute_l4_readout, SeriesRows}; use crate::storage_engines::sketch_db::index::SketchStore; use crate::storage_engines::sketch_db::query::ASAPTierResult; @@ -32,10 +27,6 @@ const SHADOW_ACCURACY: control_plane::types_v2::AccuracyTarget = /// tolerance absorbs floating-point summation-order differences only. const RELATIVE_TOLERANCE: f64 = 1e-6; -/// Mirrors `ASAPTierResult.series`'s row shape -- `(label_values, samples)` -/// where `samples` is `(window_end_unix_ms, value)`. -type SeriesRows = Vec<(BTreeMap, Vec<(i64, f64)>)>; - /// Whether shadow-mode comparison is enabled for this process. Mirrors /// `ASAP_LEGACY_DUAL_WRITE`'s exact mechanics (`drivers/ingest/otel.rs`) -- /// trimmed, case-insensitive `1`/`true`/`on`, default off. @@ -66,94 +57,16 @@ pub fn maybe_shadow_compare( return; } - let node = match lower_promql_to_l4node(query, SHADOW_ACCURACY) { - Ok(node) => node, - Err(skip) => { - tracing::debug!(query, ?skip, "shadow: query not comparable, skipping"); - return; - } - }; - - let ctx = QueryExecutionContext { - index, - t0_ms, - t1_ms, - is_cumulative, - }; - - let new_series = match execute(&node, &ctx) { - Ok(ExecOutcome::Value(values)) => { - let mut coverage: Option<(u64, u64)> = None; - let mut series = Vec::new(); - for (group_key, value) in &values { - fold_coverage(&mut coverage, value.coverage()); - series.extend(summary_value_to_series(group_key, value)); - } - (series, coverage) - } - Ok(ExecOutcome::State(groups)) => { - let mut series = Vec::new(); - for (group_key, state, _kind, _params) in &groups { - let Some(value) = state.exact_value(&None) else { - tracing::debug!( - query, - ?group_key, - "shadow: ExactAgg group had no comparable value, skipping group" - ); - continue; - }; - series.push((group_key.clone(), vec![(t1_ms as i64, value)])); + let outcome = + match execute_l4_readout(index, query, t0_ms, t1_ms, is_cumulative, SHADOW_ACCURACY) { + Ok(outcome) => outcome, + Err(skip) => { + tracing::debug!(query, ?skip, "shadow: query not comparable, skipping"); + return; } - (series, None) - } - Err(e) => { - tracing::debug!(query, error = ?e, "shadow: execute() failed, skipping"); - return; - } - }; - - diff_and_log(query, old, &new_series.0, new_series.1); -} - -/// `SummaryValue::Points`/`TopK` -> `ASAPTierResult.series`'s row shape. -/// `TopK`'s ranked-list-per-timestamp shape is pivoted into one row per -/// item (each row = the group's label map plus an `item` label, one point -/// per timestamp that item appeared in the ranked list) -- the SAME -/// convention `sketch_reducer.rs`'s own topk arm already uses, not a new -/// one invented here. -fn summary_value_to_series( - group_key: &BTreeMap, - value: &SummaryValue, -) -> SeriesRows { - match value { - SummaryValue::Points(points, _coverage) => { - vec![(group_key.clone(), points.clone())] - } - SummaryValue::TopK(ranked_per_ts, _coverage) => { - let mut by_item: BTreeMap> = BTreeMap::new(); - for (ts, items) in ranked_per_ts { - for (item, val) in items { - by_item.entry(item.clone()).or_default().push((*ts, *val)); - } - } - by_item - .into_iter() - .map(|(item, points)| { - let mut lv = group_key.clone(); - lv.insert("item".to_string(), item); - (lv, points) - }) - .collect() - } - } -} + }; -fn fold_coverage(coverage: &mut Option<(u64, u64)>, next: Option<(u64, u64)>) { - let Some((lo, hi)) = next else { return }; - *coverage = Some(match *coverage { - Some((clo, chi)) => (clo.min(lo), chi.max(hi)), - None => (lo, hi), - }); + diff_and_log(query, old, &outcome.series, outcome.coverage); } /// Diff the new path's series/coverage against the old `ASAPTierResult` 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 c5667fbd..59424f1f 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 @@ -213,6 +213,29 @@ impl GroupState { .query_statistic(stat, key, &std::collections::HashMap::new()) .ok() } + + /// Coverage analog of `exact_value` — folds `(min_window_end_ms, + /// max_window_end_ms)` from every window observed across the group's + /// entries, via the same `fold_coverage` helper + /// `readout_cumulative`/`readout_per_window` already use for the + /// sketch family (same window-end-only caveat — see `SummaryValue`'s + /// doc). Without this, a caller reading an `ExactAgg` group's value + /// via `exact_value` would have no coverage signal at all to decide + /// whether an archive tier also needs to be consulted — unlike + /// `SummaryValue::coverage()` on the sketch side. `None` for a + /// `Sketch` state or a group with no windows in range. + pub fn exact_coverage(&self) -> Option<(u64, u64)> { + let GroupState::ExactAgg { entries, .. } = self else { + return None; + }; + let mut coverage: Option<(u64, u64)> = None; + for windows in entries { + for &w_end in windows.keys() { + fold_coverage(&mut coverage, w_end); + } + } + coverage + } } #[derive(Debug)] @@ -2356,6 +2379,61 @@ mod tests { Some(25.0), "exact_value must merge both windows' sums (10 + 15)" ); + assert_eq!( + state.exact_coverage(), + Some((T0 + 1000, T0 + 2000)), + "exact_coverage must bracket both windows' end timestamps, mirroring \ + SummaryValue::coverage()'s sketch-family behavior" + ); + } + + #[test] + fn exact_coverage_is_none_for_a_sketch_state() { + // `exact_coverage` is the `ExactAgg`-only counterpart to + // `SummaryValue::coverage()` -- must not silently return something + // for a `Sketch` state. + let idx = SketchStore::new(); + let sid = 1u64; + idx.register(kll_meta(sid, "latency_ms", &[])); + idx.append_sample( + sid, + BTreeMap::new(), + (T0, T0 + 1000), + SketchSampleState { + bytes: encode_kll_items_proto(200, &[1.0, 2.0, 3.0]), + encoding: crate::storage_engines::sketch_db::index::SketchEncoding::ProtoFull, + }, + ); + let child = scan_node("latency_ms", None); + let tree = estimate_node( + kll_agg_node(child, vec![]), + SketchQuery::Quantile { q: 0.5 }, + ); + let exec = ctx(&idx); + let ExecOutcome::Value(_) = execute(&tree, &exec).expect("execute should succeed") else { + panic!("expected a value"); + }; + // Build the Sketch GroupState directly via fetch_state to exercise + // exact_coverage's defensive None arm (readout() already proved + // this tree resolves to a real Sketch Value above). + let handles = exec + .find_candidates( + &SummaryKind::Kll, + &SummaryParams::Kll { k: 200 }, + &ColumnRef::SampleValue, + &[], + &scan_node("latency_ms", None), + ) + .expect("find_candidates should succeed"); + let (_key, handle) = &handles[0]; + let state = exec + .fetch_state(handle) + .expect("fetch_state should succeed"); + assert_eq!( + state.exact_coverage(), + None, + "exact_coverage must be None for a Sketch state, not silently Some" + ); } #[test] diff --git a/data_plane/tests/e2e_controller_plans_and_backend_serves.rs b/data_plane/tests/e2e_controller_plans_and_backend_serves.rs index b9b22569..2184467e 100644 --- a/data_plane/tests/e2e_controller_plans_and_backend_serves.rs +++ b/data_plane/tests/e2e_controller_plans_and_backend_serves.rs @@ -2441,3 +2441,216 @@ async fn shadow_mode_does_not_change_served_ddsketch_quantile() { "shadow mode must not corrupt the served quantile value, got {value}" ); } + +// ── Test — the live serving cutover actually answers from SummaryExecutor ─── +// +// Phase 2 of the rollout (see the plan's "What 'safe to serve' means, +// precisely" section): with `ASAP_SUMMARY_EXECUTOR_LIVE` set, an +// unambiguous single-series query must be answered by `SummaryExecutor` +// directly (`engine.rs` skips the legacy `SketchReducer` call for it +// entirely), not merely shadow-compared. Same fixture as +// `shadow_mode_does_not_change_served_ddsketch_quantile` -- this test's +// job is proving the cutover serves a correct answer via the NEW code +// path, not re-checking quantile accuracy. + +/// RAII guard for `ASAP_SUMMARY_EXECUTOR_LIVE`. Own lock, own var -- +/// mirrors `ShadowEnvGuard` exactly (same reason: `std::env::set_var`/ +/// `remove_var` mutate process-global state and `cargo test` runs tests +/// in the same process across threads by default). +#[allow(dead_code)] +struct LiveServeEnvGuard(std::sync::MutexGuard<'static, ()>); + +impl LiveServeEnvGuard { + fn enable() -> Self { + static LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(()); + let guard = LOCK.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + std::env::set_var("ASAP_SUMMARY_EXECUTOR_LIVE", "1"); + Self(guard) + } +} + +impl Drop for LiveServeEnvGuard { + fn drop(&mut self) { + std::env::remove_var("ASAP_SUMMARY_EXECUTOR_LIVE"); + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn live_serve_actually_answers_ddsketch_quantile() { + let _live = LiveServeEnvGuard::enable(); + + let stack = start_full_stack(19_593, 19_594).await; + let client = reqwest::Client::new(); + + let workload = build_workload( + "http_latency_ms", + vec![AggType::Quantile], + 0.01, + Duration::from_secs(1), + vec!["service".to_string()], + vec![0.99], + ); + let streaming_config_json = plan_streaming_config_json(&workload); + post_streaming_config(&client, stack.backend_port, &streaming_config_json).await; + + let alpha = 0.01; + let store_counts = vec![5u64, 10, 15, 20]; + let dd_state = build_dd_sketch_state(alpha, store_counts, -1); + let sketch_bytes = dd_state.encode_to_vec(); + + let now_ns = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .expect("system time before UNIX epoch") + .as_nanos() as u64; + let sketch_t_ns = now_ns.saturating_sub(3_000_000_000); + let watermark_t_ns = now_ns.saturating_sub(1_000_000_000); + + let req = build_dd_sketch_export( + "http_latency_ms", + &[("service", "e2e-test")], + sketch_t_ns, + sketch_bytes, + alpha, + ); + post_otlp_http(&client, stack.otlp_http_port, req).await; + + let watermark_state = build_dd_sketch_state(alpha, Vec::new(), 0); + let watermark_req = build_dd_sketch_export( + "http_latency_ms", + &[("service", "e2e-test")], + watermark_t_ns, + watermark_state.encode_to_vec(), + alpha, + ); + post_otlp_http(&client, stack.otlp_http_port, watermark_req).await; + + tokio::time::sleep(Duration::from_millis(800)).await; + + let query_url = format!("http://127.0.0.1:{}/api/v1/query", stack.backend_port); + let response: JsonValue = client + .get(&query_url) + .query(&[("query", "quantile_over_time(0.99, http_latency_ms[10s])")]) + .send() + .await + .expect("PromQL query failed to send") + .json() + .await + .expect("PromQL response was not JSON"); + + assert_eq!( + response["status"].as_str().unwrap_or("(missing)"), + "success", + "live-serve must answer this unambiguous single-series quantile. Response:\n{}", + serde_json::to_string_pretty(&response).unwrap_or_default() + ); + + let value = response["data"]["result"] + .as_array() + .and_then(|r| extract_first_scalar(&JsonValue::Array(r.clone()))) + .expect("expected a scalar quantile result"); + assert!( + value.is_finite() && value > 0.0, + "live-serve must produce a correct served quantile value, got {value}" + ); +} + +// ── Test — the live serving cutover falls back correctly on the known ────── +// ambiguous global-merge shape (ASAPController#163) +// +// `count(hll_metric)` with NO `by (...)` and MULTIPLE distinct-service HLL +// sids is exactly the ambiguous shape the design doc's "Grouping +// semantics" section describes: `SummaryAgg{by: []}` can't tell "no +// grouping concept" from "reduce everything." `live_serve.rs`'s +// `ambiguous_merge_risk` gate must decline to serve this from the new +// path even with the flag on, falling back to the legacy +// `evaluate_cardinality_global` special case (which already merges the +// registers correctly) -- so the end-to-end answer must still succeed. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn live_serve_ambiguous_hll_global_count_falls_back_correctly() { + let _live = LiveServeEnvGuard::enable(); + + let stack = start_full_stack(19_595, 19_596).await; + let client = reqwest::Client::new(); + + let workload = build_workload_with_override( + "unique_users_per_min", + vec![AggType::Cardinality], + 0.05, + Duration::from_secs(1), + vec!["service".to_string()], + Vec::new(), + Some(SketchType::HLL), + ); + let streaming_config_json = plan_streaming_config_json(&workload); + post_streaming_config(&client, stack.backend_port, &streaming_config_json).await; + + let precision = 10u32; + let num_registers = 1usize << precision; + + let now_ns = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .expect("system time before UNIX epoch") + .as_nanos() as u64; + let sketch_t_ns = now_ns.saturating_sub(3_000_000_000); + let watermark_t_ns = now_ns.saturating_sub(1_000_000_000); + + // Two distinct services, DISJOINT non-zero registers -- two separate + // sids the analyzer's `count(unique_users_per_min)` candidate resolves + // to together (empty group_by_keys), the exact shape ASAPController#163 + // describes. + for (service, reg_idx) in [("svc-a", 0usize), ("svc-b", 500usize)] { + let mut registers = vec![0u8; num_registers]; + registers[reg_idx] = 6; + let hll_state = build_hll_state(precision, registers); + let req = build_hll_export( + "unique_users_per_min", + &[("service", service)], + sketch_t_ns, + hll_state.encode_to_vec(), + precision, + ); + post_otlp_http(&client, stack.otlp_http_port, req).await; + + let watermark_state = build_hll_state(precision, vec![0u8; num_registers]); + let watermark_req = build_hll_export( + "unique_users_per_min", + &[("service", service)], + watermark_t_ns, + watermark_state.encode_to_vec(), + precision, + ); + post_otlp_http(&client, stack.otlp_http_port, watermark_req).await; + } + + tokio::time::sleep(Duration::from_millis(800)).await; + + let response: JsonValue = client + .get(format!( + "http://127.0.0.1:{}/api/v1/query", + stack.backend_port + )) + .query(&[("query", "count(unique_users_per_min)")]) + .send() + .await + .expect("query failed") + .json() + .await + .expect("response not JSON"); + + assert_eq!( + response["status"].as_str().unwrap_or("(missing)"), + "success", + "the ambiguous global-merge shape must still succeed via legacy fallback \ + with live-serve on. Response:\n{}", + serde_json::to_string_pretty(&response).unwrap_or_default() + ); + + let value = response["data"]["result"] + .as_array() + .and_then(|r| extract_first_scalar(&JsonValue::Array(r.clone()))) + .expect("expected a scalar cardinality result"); + assert!( + value.is_finite() && value > 0.0, + "expected a valid merged cardinality estimate, got {value}" + ); +} From 0848fe9a6c5fbeda42c3807b94646f7671aab051 Mon Sep 17 00:00:00 2001 From: zz_y Date: Tue, 28 Jul 2026 16:07:32 -0600 Subject: [PATCH 2/2] refactor(live_serve): remove the grouping-ambiguity gate, now obsolete `L4ReadoutOutcome::ambiguous_merge_risk` and `live_serve.rs`'s use of it existed purely as a workaround for ASAPController#163: an empty `by: Vec` was indistinguishable between "no grouping concept applies" (the group split is correct) and "an aggregation operator asked to reduce everything" (the groups should have merged). Unable to tell which, `try_serve_from_summary_executor` declined to serve ANY empty-`by` shape that produced more than one group, falling back to the legacy path. That ambiguity no longer exists. ASAPController#165 made the reduction kind explicit (`Reduction::{PerEntity, Reduce(GroupKeys)}`) and the previous commit made `summary_executor.rs::resolve_group_key` act on it, so both branches are already resolved correctly before the gate ran: * `PerEntity` -- the multi-group split is definitionally correct (one row per entity, never merged). Never a "risk"; the gate could only ever DECLINE a correct answer here. * `Reduce([])` -- every candidate shares one group key, so the outcome has exactly one group and the `values.len() > 1` trigger cannot fire at all. The flag would therefore be unconditionally `false` today. Keeping it would mean keeping a heuristic whose only remaining effect is spurious fallback, so it's removed rather than rewritten against `Reduction`: the field, its computation, the `root_summary_agg_by_is_empty` tree walk that fed it, and the `live_serve.rs` early-return are all deleted. Tests that pinned the OLD behavior are inverted rather than dropped, since they cover exactly the case that changed: * `flag_on_ambiguous_shape_falls_back` -> `flag_on_global_merge_shape_is_served_merged_not_declined`: asserted `is_none()` (declined); now asserts the shape IS served as ONE merged series with cardinality ~6 across both sids, not ~3. * `ambiguous_global_merge_shape_is_flagged` -> `global_merge_shape_now_merges_instead_of_being_declined`: asserted the flag plus TWO unmerged series; now asserts ONE merged series. * The e2e `live_serve_ambiguous_hll_global_count_falls_back_correctly` -> `live_serve_hll_global_count_merges_across_sids`, and its assertion tightened from "any positive value" (which a legacy fallback also satisfied) to ">= 1.5", which distinguishes a real cross-sid merge (~2) from serving only one sid's registers (~1). Verified: `cargo check -p data_plane --all-targets` clean; `cargo test -p data_plane --lib` 956 passed / 0 failed. Refs ProjectASAP/ASAPController#163, #164, #165 Co-Authored-By: Claude Sonnet 5 --- .../asap_query_engine/l4_readout.rs | 116 +++++++++--------- .../asap_query_engine/live_serve.rs | 48 +++++--- .../asap_query_engine/summary_executor.rs | 4 +- ...e2e_controller_plans_and_backend_serves.rs | 37 +++--- 4 files changed, 113 insertions(+), 92 deletions(-) diff --git a/data_plane/src/query_engines/asap_query_engine/l4_readout.rs b/data_plane/src/query_engines/asap_query_engine/l4_readout.rs index 63f9f6d4..5801f7a8 100644 --- a/data_plane/src/query_engines/asap_query_engine/l4_readout.rs +++ b/data_plane/src/query_engines/asap_query_engine/l4_readout.rs @@ -26,22 +26,36 @@ pub type SeriesRows = Vec<(BTreeMap, Vec<(i64, f64)>)>; /// (`ExecOutcome::Value`) or `ExactAgg` (`ExecOutcome::State`) side — /// callers that only care about "did this answer the query, and is it /// safe to trust" don't need to know which. +/// NOTE — this used to carry an `ambiguous_merge_risk` flag, and +/// `live_serve.rs` used it to DECLINE to serve an ambiguous shape. +/// That gate is now removed, because the ambiguity it guarded against no +/// longer exists. +/// +/// It existed because an empty `by: Vec` was indistinguishable +/// between "no grouping concept applies" (the group split is correct) and +/// "an aggregation operator asked to reduce everything" (the groups +/// should have been merged) — exactly +/// [ASAPController#163](https://github.com/ProjectASAP/ASAPController/issues/163). +/// Unable to tell which, the safe move was to fall back to the legacy +/// path whenever an empty `by` produced >1 group. +/// +/// ASAPController#165 removed that ambiguity at the source by making the +/// reduction kind explicit (`Reduction::{PerEntity, Reduce(GroupKeys)}`), +/// and `summary_executor.rs::resolve_group_key` now acts on it directly. +/// Both branches are resolved correctly BEFORE reaching here: +/// +/// * `PerEntity` — the multi-group split is definitionally right (one row +/// per entity, never merged), so it was never a "risk" to begin with. +/// * `Reduce([])` — every candidate shares one group key, so the outcome +/// has exactly ONE group and the old `values.len() > 1` trigger cannot +/// fire at all. +/// +/// The flag would therefore be unconditionally `false` today; keeping it +/// would mean keeping a heuristic that can only ever misfire (declining +/// correct `PerEntity` answers) now that the real signal is available. pub struct L4ReadoutOutcome { pub series: SeriesRows, pub coverage: Option<(u64, u64)>, - /// `true` only for a sketch-family (`ExecOutcome::Value`) outcome - /// whose tree's root `SummaryAgg` had an empty `by` AND produced more - /// than one group. This is exactly the ambiguous shape - /// [ASAPController#163](https://github.com/ProjectASAP/ASAPController/issues/163) - /// describes — an empty `by` is indistinguishable between "no - /// grouping concept applies" (this group split is correct) and "an - /// aggregation operator asked to reduce everything" (these groups - /// should have been merged into one). Always `false` for `ExactAgg` - /// (`Sum`/`Increase` map only from genuine aggregation operators, so - /// their empty `by` is unambiguous — see the design doc's "Grouping - /// semantics"/"ExactAgg" sections) and for any sketch outcome with a - /// non-empty `by` or ≤1 resulting group (nothing to disagree about). - pub ambiguous_merge_risk: bool, } /// Lower `query`, execute it against `index` over `[t0_ms, t1_ms]`, and @@ -58,7 +72,6 @@ pub fn execute_l4_readout( accuracy: AccuracyTarget, ) -> Result { let node = lower_promql_to_l4node(query, accuracy)?; - let by_is_empty = root_summary_agg_by_is_empty(&node); let ctx = QueryExecutionContext { index, @@ -75,12 +88,7 @@ pub fn execute_l4_readout( fold_coverage(&mut coverage, value.coverage()); series.extend(summary_value_to_series(group_key, value)); } - let ambiguous_merge_risk = by_is_empty.unwrap_or(false) && values.len() > 1; - Ok(L4ReadoutOutcome { - series, - coverage, - ambiguous_merge_risk, - }) + Ok(L4ReadoutOutcome { series, coverage }) } Ok(ExecOutcome::State(groups)) => { let mut coverage: Option<(u64, u64)> = None; @@ -92,37 +100,12 @@ pub fn execute_l4_readout( }; series.push((group_key.clone(), vec![(t1_ms as i64, value)])); } - Ok(L4ReadoutOutcome { - series, - coverage, - ambiguous_merge_risk: false, - }) + Ok(L4ReadoutOutcome { series, coverage }) } Err(e) => Err(LoweringSkip::ExecuteFailed(format!("{e:?}"))), } } -/// Walk down to the tree's `SummaryAgg` node (through a `SummaryEstimate` -/// wrapper if present, and through the first child of a `SummaryMerge` — -/// its children agree on `(SummaryKind, SummaryParams)` by construction, -/// so their `by` agrees too) and report whether its `by` list is empty. -/// `None` for a bare `Logical` root (shouldn't happen here — -/// `lower_promql_to_l4node` already rejects that via `NotRealized` — kept -/// exhaustive and defensive rather than assumed unreachable). -fn root_summary_agg_by_is_empty(node: &L4Node) -> Option { - match &node.expr { - SummaryExpr::SummaryAgg { by, .. } => Some(by.is_empty()), - SummaryExpr::SummaryEstimate { sketch_input, .. } => { - root_summary_agg_by_is_empty(sketch_input) - } - SummaryExpr::SummaryMerge { children } => children - .first() - .and_then(|c| root_summary_agg_by_is_empty(c)), - SummaryExpr::Logical(_) => None, - _ => None, - } -} - /// `SummaryValue::Points`/`TopK` -> `ASAPTierResult.series`'s row shape. /// `TopK`'s ranked-list-per-timestamp shape is pivoted into one row per /// item (each row = the group's label map plus an `item` label, one point @@ -262,7 +245,7 @@ mod tests { } #[test] - fn unambiguous_sketch_query_is_not_flagged() { + fn bare_range_function_keeps_one_series_per_entity() { let idx = ddsketch_fixture(); let outcome = execute_l4_readout( &idx, @@ -273,34 +256,48 @@ mod tests { accuracy(), ) .expect("should execute"); - assert!( - !outcome.ambiguous_merge_risk, - "a single-series bare range function must not be flagged ambiguous" - ); assert_eq!(outcome.series.len(), 1); } #[test] - fn ambiguous_global_merge_shape_is_flagged() { + fn global_merge_shape_now_merges_instead_of_being_declined() { // The exact ASAPController#163 shape: two HLL sids, no explicit - // by(), an aggregation-operator query -- find_candidates can't - // tell whether these two groups should have been merged. + // by(), an aggregation-operator query. This test previously + // asserted `ambiguous_merge_risk == true` and TWO unmerged series + // -- i.e. it pinned the old workaround, where an empty `by` left + // `find_candidates` unable to tell "reduce everything" apart from + // "no grouping concept," so `live_serve` declined to serve the + // shape at all. + // + // With `Reduction` (ASAPController#165) that ambiguity is gone: + // `count(...)` is a genuine aggregation operator, so it lowers to + // `Reduce([])` and `resolve_group_key` gives every candidate the + // SAME group key -- the two sids MERGE into one answer, which is + // what the query actually asked for. No gate, no fallback. let idx = SketchStore::new(); register_hll(&idx, 1, "svc-a", &["a", "b", "c"]); register_hll(&idx, 2, "svc-b", &["d", "e", "f"]); let outcome = execute_l4_readout(&idx, "count(unique_users)", 1_000, 2_000, true, accuracy()) .expect("should execute"); - assert!( - outcome.ambiguous_merge_risk, - "two distinct-service HLL groups under a by-less count() must be flagged, got {:?}", + assert_eq!( + outcome.series.len(), + 1, + "a by-less count() is a full reduction -- both HLL sids must merge into ONE \ + series, not stay split (and not be declined), got {:?}", outcome.series ); - assert_eq!(outcome.series.len(), 2); + // Disjoint item sets {a,b,c} + {d,e,f} -> merged cardinality ~6. + let (_group, points) = &outcome.series[0]; + let card = points[0].1; + assert!( + (4.0..=8.0).contains(&card), + "merged cardinality {card} should be ~6 (both sids' disjoint items), not ~3" + ); } #[test] - fn exact_agg_outcome_is_never_flagged_ambiguous() { + fn exact_agg_outcome_reports_window_end_coverage() { let idx = SketchStore::new(); idx.register( crate::storage_engines::sketch_db::index::SketchInstanceMetadata { @@ -328,7 +325,6 @@ mod tests { ); let outcome = execute_l4_readout(&idx, "sum(bytes_total)", 1_000, 2_000, true, accuracy()) .expect("should execute"); - assert!(!outcome.ambiguous_merge_risk); // Window-end-only coverage: a single window (1_000, 2_000) is // keyed by its end (2_000) alone, so both bounds equal 2_000 -- // same semantics as `SummaryValue::coverage()`, reconfirmed for diff --git a/data_plane/src/query_engines/asap_query_engine/live_serve.rs b/data_plane/src/query_engines/asap_query_engine/live_serve.rs index f0c52397..26bf4444 100644 --- a/data_plane/src/query_engines/asap_query_engine/live_serve.rs +++ b/data_plane/src/query_engines/asap_query_engine/live_serve.rs @@ -35,11 +35,17 @@ pub fn summary_executor_live_enabled() -> bool { /// Try to serve `query` entirely from `SummaryExecutor`. Returns `None` /// whenever the caller should fall back to the legacy path exactly as -/// it does today (flag off, lowering/execution failed, or the -/// grouping-ambiguity gate tripped — see `L4ReadoutOutcome::ambiguous_merge_risk`'s -/// doc) — `None` here is indistinguishable from Phase 1's shadow-only -/// behavior. `Some(...)` means the new path answered and the caller -/// must NOT also call the legacy reducer for this candidate. +/// it does today (flag off, or lowering/execution failed) — `None` here +/// is indistinguishable from Phase 1's shadow-only behavior. `Some(...)` +/// means the new path answered and the caller must NOT also call the +/// legacy reducer for this candidate. +/// +/// This used to carry a third fallback reason: a grouping-ambiguity gate +/// that declined any empty-`by` shape producing >1 group +/// (ASAPController#163). That gate is gone — `Reduction` +/// (ASAPController#165) lets `summary_executor.rs` resolve both halves of +/// the ambiguity correctly on its own, so there is no longer a shape to +/// decline. See `L4ReadoutOutcome`'s doc for the full reasoning. pub fn try_serve_from_summary_executor( index: &SketchStore, query: &str, @@ -64,14 +70,6 @@ pub fn try_serve_from_summary_executor( } }; - if outcome.ambiguous_merge_risk { - tracing::debug!( - query, - "live: ambiguous global-merge shape (ASAPController#163), falling back to legacy path" - ); - return None; - } - tracing::debug!(query, "live: served from SummaryExecutor"); Some(ASAPTierResult { series: outcome.series, @@ -227,16 +225,34 @@ mod tests { } #[test] - fn flag_on_ambiguous_shape_falls_back() { + fn flag_on_global_merge_shape_is_served_merged_not_declined() { + // Previously `flag_on_ambiguous_shape_falls_back`, asserting + // `result.is_none()`: the grouping-ambiguity gate declined this + // shape because an empty `by` couldn't be told apart from "reduce + // everything" (ASAPController#163). With `Reduction` (#165) the + // executor resolves it -- `count(...)` lowers to `Reduce([])`, both + // sids share one group key, and the new path serves the correctly + // merged answer instead of falling back. let _guard = set_live_env("1"); let idx = SketchStore::new(); register_hll(&idx, 1, "svc-a", &["a", "b", "c"]); register_hll(&idx, 2, "svc-b", &["d", "e", "f"]); let result = try_serve_from_summary_executor(&idx, "count(unique_users)", 1_000, 2_000, true); + let result = result.expect( + "global-merge shape is no longer ambiguous -- it must be served, not declined", + ); + assert_eq!( + result.series.len(), + 1, + "a by-less count() must merge both sids into ONE series, got {:?}", + result.series + ); + // Disjoint item sets {a,b,c} + {d,e,f} -> merged cardinality ~6. + let card = result.series[0].1[0].1; assert!( - result.is_none(), - "ambiguous global-merge shape must fall back to legacy, not serve a possibly-wrong answer" + (4.0..=8.0).contains(&card), + "merged cardinality {card} should be ~6 (both sids), not ~3 (one sid)" ); } 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 59424f1f..bfe0bf10 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 @@ -2406,7 +2406,7 @@ mod tests { ); let child = scan_node("latency_ms", None); let tree = estimate_node( - kll_agg_node(child, vec![]), + kll_agg_node(child, Reduction::by(vec![])), SketchQuery::Quantile { q: 0.5 }, ); let exec = ctx(&idx); @@ -2421,7 +2421,7 @@ mod tests { &SummaryKind::Kll, &SummaryParams::Kll { k: 200 }, &ColumnRef::SampleValue, - &[], + &Reduction::by(vec![]), &scan_node("latency_ms", None), ) .expect("find_candidates should succeed"); diff --git a/data_plane/tests/e2e_controller_plans_and_backend_serves.rs b/data_plane/tests/e2e_controller_plans_and_backend_serves.rs index 2184467e..ac360475 100644 --- a/data_plane/tests/e2e_controller_plans_and_backend_serves.rs +++ b/data_plane/tests/e2e_controller_plans_and_backend_serves.rs @@ -2554,19 +2554,23 @@ async fn live_serve_actually_answers_ddsketch_quantile() { ); } -// ── Test — the live serving cutover falls back correctly on the known ────── -// ambiguous global-merge shape (ASAPController#163) +// ── Test — the live serving cutover MERGES the global-merge shape ───────── +// correctly, end to end (ASAPController#163/#165) // // `count(hll_metric)` with NO `by (...)` and MULTIPLE distinct-service HLL -// sids is exactly the ambiguous shape the design doc's "Grouping -// semantics" section describes: `SummaryAgg{by: []}` can't tell "no -// grouping concept" from "reduce everything." `live_serve.rs`'s -// `ambiguous_merge_risk` gate must decline to serve this from the new -// path even with the flag on, falling back to the legacy -// `evaluate_cardinality_global` special case (which already merges the -// registers correctly) -- so the end-to-end answer must still succeed. +// sids used to be the ambiguous shape the design doc's "Grouping +// semantics" section described: `SummaryAgg{by: []}` couldn't tell "no +// grouping concept" from "reduce everything," so `live_serve.rs`'s +// `ambiguous_merge_risk` gate DECLINED to serve it from the new path and +// fell back to the legacy `evaluate_cardinality_global` special case. +// +// `Reduction` (ASAPController#165) resolves that: `count(...)` is a +// genuine aggregation operator, so it lowers to `Reduce([])` and +// `resolve_group_key` gives both sids the same group key -- the new path +// merges them itself. The gate is gone; this now exercises the new +// path serving the shape directly, not a fallback. #[tokio::test(flavor = "multi_thread", worker_threads = 4)] -async fn live_serve_ambiguous_hll_global_count_falls_back_correctly() { +async fn live_serve_hll_global_count_merges_across_sids() { let _live = LiveServeEnvGuard::enable(); let stack = start_full_stack(19_595, 19_596).await; @@ -2640,8 +2644,7 @@ async fn live_serve_ambiguous_hll_global_count_falls_back_correctly() { assert_eq!( response["status"].as_str().unwrap_or("(missing)"), "success", - "the ambiguous global-merge shape must still succeed via legacy fallback \ - with live-serve on. Response:\n{}", + "the global-merge shape must succeed with live-serve on. Response:\n{}", serde_json::to_string_pretty(&response).unwrap_or_default() ); @@ -2649,8 +2652,14 @@ async fn live_serve_ambiguous_hll_global_count_falls_back_correctly() { .as_array() .and_then(|r| extract_first_scalar(&JsonValue::Array(r.clone()))) .expect("expected a scalar cardinality result"); + // Each service set exactly ONE distinct non-zero register, and the two + // are disjoint -- a correct cross-sid merge estimates ~2, whereas + // serving only one sid's state would estimate ~1. The assertion is + // loose (HLL at precision 10 is approximate) but still distinguishes + // "merged both" from "dropped one." assert!( - value.is_finite() && value > 0.0, - "expected a valid merged cardinality estimate, got {value}" + value.is_finite() && value >= 1.5, + "expected the MERGED cardinality across both services (~2), got {value} -- \ + a value near 1 means only one sid's registers were counted" ); }