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..5801f7a8 --- /dev/null +++ b/data_plane/src/query_engines/asap_query_engine/l4_readout.rs @@ -0,0 +1,334 @@ +//! 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. +/// 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)>, +} + +/// 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 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)); + } + Ok(L4ReadoutOutcome { series, coverage }) + } + 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 }) + } + Err(e) => Err(LoweringSkip::ExecuteFailed(format!("{e:?}"))), + } +} + +/// `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 bare_range_function_keeps_one_series_per_entity() { + 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_eq!(outcome.series.len(), 1); + } + + #[test] + 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. 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_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 + ); + // 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_reports_window_end_coverage() { + 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"); + // 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..26bf4444 --- /dev/null +++ b/data_plane/src/query_engines/asap_query_engine/live_serve.rs @@ -0,0 +1,267 @@ +//! 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, 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, + 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; + } + }; + + 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_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!( + (4.0..=8.0).contains(&card), + "merged cardinality {card} should be ~6 (both sids), not ~3 (one sid)" + ); + } + + #[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..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 @@ -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, Reduction::by(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, + &Reduction::by(vec![]), + &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..ac360475 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,225 @@ 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 MERGES the global-merge shape ───────── +// correctly, end to end (ASAPController#163/#165) +// +// `count(hll_metric)` with NO `by (...)` and MULTIPLE distinct-service HLL +// 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_hll_global_count_merges_across_sids() { + 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 global-merge shape must succeed 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"); + // 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 >= 1.5, + "expected the MERGED cardinality across both services (~2), got {value} -- \ + a value near 1 means only one sid's registers were counted" + ); +}