feat(query): count_over_time binds to FrequencyEstimate (CMS / CountSketch)

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-

control_plane/src/intent_algebra/lower.rs

@@ -485,6 +485,7 @@ fn agg_func_to_intents(func: &AggFunc) -> Vec<AggIntent> {
         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,

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"),

control_plane/src/query_parser/promql.rs

@@ -448,7 +448,25 @@ fn walk_call_to_op(call: &Call, ctx: &WalkCtx) -> anyhow::Result<AggFunc> {
         "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),

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]

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())),
         }
     }