diff --git a/control_plane/src/asap_tier_analysis.rs b/control_plane/src/asap_tier_analysis.rs index 46c02cde..57249ec6 100644 --- a/control_plane/src/asap_tier_analysis.rs +++ b/control_plane/src/asap_tier_analysis.rs @@ -831,12 +831,14 @@ mod tests { } #[test] - fn is_asap_tier_answerable_false_for_unsupported() { - // `count_over_time(...)` without an outer `count by (...)` lowers - // to `AggIntent::Count { accuracy: Exact }` — exact counts have - // no ASAP-tier sketch, so `capability_for` returns `None`. + fn is_asap_tier_answerable_true_for_count_over_time() { + // `count_over_time(...)` now lowers to + // `AggIntent::Frequency{Epsilon}` (per-series sample count + // over the window), which `capability_for` maps to + // `Capability::FrequencyEstimate(Any)` — answerable by any + // frequency-family sketch (CMS / CountSketch, heap-less). let a = analyze_promql_for_asap_tier("count_over_time(m[5m])"); - assert!(!a.is_asap_tier_answerable()); + assert!(a.is_asap_tier_answerable(), "{a:?}"); } #[test] @@ -849,29 +851,23 @@ mod tests { // ── Cardinality / count_over_time real-PromQL acceptance ──────────── - /// `count_over_time(...)` is real PromQL and lowers to - /// `AggIntent::Count{accuracy:Exact}` per `intent_algebra::lower`. - /// Exact-accuracy Count has no ASAP-tier binding, so the analyzer - /// surfaces this as `UnsupportedAggIntent("count")` — the routing - /// layer then sends it to archive, which is the right behavior - /// because `count_over_time` counts samples (not distinct values). + /// `count_over_time(metric[range])` is the PromQL per-series + /// sample-count idiom — exactly what a heap-less CMS / CountSketch + /// estimates. The parser maps it to `AggFunc::Frequency` which + /// lowers to `AggIntent::Frequency{Epsilon}`; `capability_for` + /// returns `Capability::FrequencyEstimate(Any)`. The warm engine + /// binds the query to any frequency-family policy registered for + /// the metric. #[test] - fn count_over_time_is_unsupported_at_exact_accuracy() { - // `count_over_time(metric[r])` without an outer `count by (...)` - // is the PromQL "count samples per window" idiom — exact at L3. - // The lowerer doesn't emit an AggIntent for it (no entry in - // `AggType`), so the analyzer surfaces the raw function name - // from the AST trace as the `UnsupportedAggIntent` label. + fn count_over_time_binds_to_frequency_estimate() { let a = analyze_promql_for_asap_tier("count_over_time(http_requests_total[5m])"); - match a.unsupported { - Some(UnsupportedReason::UnsupportedAggIntent(kind)) => { - assert!( - kind == "count" || kind == "count_over_time", - "unexpected intent kind: {kind}" - ); - } - other => panic!("expected UnsupportedAggIntent, got {other:?}"), - } + assert!(a.unsupported.is_none(), "{a:?}"); + assert_eq!(a.candidates.len(), 1, "{a:?}"); + assert_eq!( + a.candidates[0].required_capability, + Capability::FrequencyEstimate(SketchKindHandle::Any), + "{a:?}" + ); } /// `count by (...) (count_over_time(...))` is the PromQL distinct- diff --git a/control_plane/src/intent_algebra/lower.rs b/control_plane/src/intent_algebra/lower.rs index d8f99f6f..1e616e53 100644 --- a/control_plane/src/intent_algebra/lower.rs +++ b/control_plane/src/intent_algebra/lower.rs @@ -485,6 +485,7 @@ fn agg_func_to_intents(func: &AggFunc) -> Vec { AggFunc::Quantile(phi) => vec![default_quantile(*phi)], AggFunc::CountDistinct => vec![default_cardinality()], AggFunc::HeavyHitters { .. } => vec![default_frequency()], + AggFunc::Frequency => vec![default_frequency()], AggFunc::Count => vec![default_frequency()], AggFunc::Avg => vec![AggIntent::Quantile { q: 0.5, diff --git a/control_plane/src/intent_algebra/relational.rs b/control_plane/src/intent_algebra/relational.rs index 84d446fd..9059bedf 100644 --- a/control_plane/src/intent_algebra/relational.rs +++ b/control_plane/src/intent_algebra/relational.rs @@ -468,6 +468,14 @@ pub enum AggFunc { Quantile(f64), /// COUNT DISTINCT — maps to HLL. CountDistinct, + /// Per-series frequency estimation — maps to CMS / CountSketch. + /// PromQL surface: `count_over_time(metric[range])` (counts + /// samples per series in the window). Distinct from `Count` + /// because `count_over_time` is structurally per-series and + /// always sketchable, where `Count` carries the SQL `COUNT(*)` + /// exact-row-count case that the un-grouped lowering branch + /// pins to `AggIntent::Count{Exact}`. + Frequency, /// Top-K heavy hitters — maps to CountSketch. HeavyHitters { k: u64 }, /// PromQL `rate()` — per-second increase over a window. @@ -495,7 +503,10 @@ impl AggFunc { pub fn is_sketchable(&self) -> bool { matches!( self, - AggFunc::Quantile(_) | AggFunc::CountDistinct | AggFunc::HeavyHitters { .. } + AggFunc::Quantile(_) + | AggFunc::CountDistinct + | AggFunc::Frequency + | AggFunc::HeavyHitters { .. } ) } @@ -508,6 +519,7 @@ impl AggFunc { match self { AggFunc::Quantile(phi) => Some(default_quantile(*phi)), AggFunc::CountDistinct => Some(default_cardinality()), + AggFunc::Frequency => Some(default_frequency()), AggFunc::HeavyHitters { .. } => Some(default_frequency()), AggFunc::Count => Some(AggIntent::Count { accuracy: AccuracyTarget::Exact }), AggFunc::Sum => Some(AggIntent::Sum), @@ -683,6 +695,7 @@ impl std::fmt::Display for AggFunc { AggFunc::Variance { .. } => write!(f, "VARIANCE"), AggFunc::Quantile(p) => write!(f, "QUANTILE({p})"), AggFunc::CountDistinct => write!(f, "COUNT_DISTINCT"), + AggFunc::Frequency => write!(f, "FREQUENCY"), AggFunc::HeavyHitters { k } => write!(f, "HEAVY_HITTERS({k})"), AggFunc::Rate => write!(f, "rate"), AggFunc::Increase => write!(f, "increase"), diff --git a/control_plane/src/query_parser/promql.rs b/control_plane/src/query_parser/promql.rs index d29c5e02..efdb28fa 100644 --- a/control_plane/src/query_parser/promql.rs +++ b/control_plane/src/query_parser/promql.rs @@ -448,7 +448,25 @@ fn walk_call_to_op(call: &Call, ctx: &WalkCtx) -> anyhow::Result { "stddev_over_time" => Ok(AggFunc::StdDev { population: false }), "stdvar_over_time" => Ok(AggFunc::Variance { population: false }), "count_over_time" => { - Ok(if ctx.outer_count { AggFunc::CountDistinct } else { AggFunc::Count }) + // Three cases, in priority order: + // * Inside `count by (...) (count_over_time(...))` → + // `CountDistinct` (HLL distinct counting; the outer + // count of inner counts is cardinality). + // * Inside `topk(N, count_over_time(...))` → `Count` + // (the topk wrapper expects a count-shaped inner). + // * Otherwise → `Frequency` (per-series sample-count + // estimation; routes to CMS / CountSketch via the + // `AggFunc::Frequency → default_frequency()` lowering). + // This was previously `AggFunc::Count` which the + // un-grouped lowering branch pinned to + // `AggIntent::Count{Exact}` → unsupported by ASAP. + Ok(if ctx.outer_count { + AggFunc::CountDistinct + } else if ctx.topk.is_some() { + AggFunc::Count + } else { + AggFunc::Frequency + }) } "sum_over_time" | "last_over_time" | "present_over_time" | "absent_over_time" => Ok(AggFunc::Sum), 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 95a890e2..ae0bbfc2 100644 --- a/data_plane/src/query_engines/asap_query_engine/engine.rs +++ b/data_plane/src/query_engines/asap_query_engine/engine.rs @@ -6414,7 +6414,7 @@ mod analyzer_parity_tests { ctrl OK [metric=m gbk=[\"svc\"] cap=FrequencyTopk(Any) fn=topk args=[10.0] range_s=0] engine MISS(NoPattern) ─── q08: count_over_time(http_requests_total[5m]) - ctrl MISS(UnsupportedAggIntent(\"count\")) + ctrl OK [metric=http_requests_total gbk=[] cap=FrequencyEstimate(Any) fn=count_over_time args=[] range_s=300] engine OK pattern=only_temporal stats=[count] metric=http_requests_total fn=count_over_time agg_op= range_s=300 range_ms=Some(300000) spatial=\"\" grouping=[] ─── q09: count by (zone) (count_over_time(http_requests_total[5m])) ctrl OK [metric=http_requests_total gbk=[\"zone\"] cap=CardinalityApprox fn=count args=[] range_s=300 | metric=http_requests_total gbk=[\"zone\"] cap=CardinalityApprox fn=count args=[] range_s=300] diff --git a/data_plane/src/storage_engines/sketch_db/query/sketch_reducer.rs b/data_plane/src/storage_engines/sketch_db/query/sketch_reducer.rs index 5a3fa491..fad42d3c 100644 --- a/data_plane/src/storage_engines/sketch_db/query/sketch_reducer.rs +++ b/data_plane/src/storage_engines/sketch_db/query/sketch_reducer.rs @@ -230,9 +230,13 @@ impl<'a> SketchReducer<'a> { // `sum by (item) (rate(m[r]))` with epsilon accuracy. The // reducer answers these by decoding the CMS / CountSketch // matrix directly (no heap needed). `frequency` is the - // canonical name; `count_over_time` is accepted as an alias - // for back-compat with PromQL counter-style point queries. - "frequency" | "frequency_estimate" => Ok(QueryFamily::FrequencyEstimate), + // canonical name; `count_over_time` is the PromQL surface + // (per-series sample count over a range window — exactly + // what a CMS / CountSketch estimates without distinct-set + // tracking). + "frequency" | "frequency_estimate" | "count_over_time" => { + Ok(QueryFamily::FrequencyEstimate) + } other => Err(ASAPTierError::UnsupportedFunction(other.to_string())), } }