fix(promql): count observations with unit sketch weights

crates/asap-aware-mapping/src/replacement.rs

@@ -2128,7 +2128,12 @@ fn realize_value_frequency_summary_input(
     _reduction: &Reduction,
     child: &Rc<QueryExpr>,
 ) -> PhysicalSummaryInputRuleResult {
-    if !matches!(family, SummaryFamilyType::Sketch(kind, _) if kind.algorithm() == &SketchAlgorithm::UnivMon)
+    // Frequency counts hash sample values as items but add one per observation.
+    // Using the sample as a weight would turn counts into sums and admit signed CMS updates.
+    if !matches!(family, SummaryFamilyType::Sketch(kind, _)
+        if kind.algorithm() == &SketchAlgorithm::UnivMon
+            || (matches!(intent, AggIntent::Count { .. })
+                && matches!(kind.algorithm(), SketchAlgorithm::Cms | SketchAlgorithm::CountSketch)))
     {
         return PhysicalSummaryInputRuleResult::NotApplicable;
     }
@@ -7979,7 +7984,7 @@ mod tests {
     }
 
     /// Issue #163, case 2: an aggregation operator explicitly invoked with
-    /// no `by(...)` (e.g. `count(hll_metric)`) realizes to `SummaryAgg {
+    /// no grouping keys realizes to `SummaryAgg {
     /// reduction: Reduce(vec![]), .. }` — byte-identical `by: []` to the
     /// previous test at the old `Vec<ColumnId>` shape; `reduction` is what
     /// tells them apart now.

crates/frontend-promql/src/promql.rs

@@ -45,7 +45,7 @@
 //! | `info(v, [selector])` | `PromqlInfoEnrich{selector}` — label-enrichment join against the info metric(s); join keys resolved during post-ASAP binding (issue #84) |
 //! | `group` / `offset` / `@` / `info` | **rejected** — distinct semantics with no intent-algebra representation yet (`info` label-join → #84) |
 //! | `OUTER by (dims) (…)` | `Aggregate.reduction = Reduce(by = dims)` (generic `topk by`/`bottomk` grouping → `Sort.partition_by`) |
-//! | `count by (d) (…)` | `Aggregate{[Cardinality], …}` |
+//! | `count by (d) (…)` | `Aggregate{[Count], …}` |
 //! | `group(v)` / `count_values("l", v)` | `Aggregate{[Group]}` (constant 1) / `Aggregate{[CountValues{l}]}` (group-by-value + count, new label `l`) — issue #49 |
 //! | `limitk(k, v)` / `limit_ratio(r, v)` | `PromqlSeriesSample{LimitK(k) \| LimitRatio(r)}` — series-sampling selection, whole series kept unchanged (issue #86) |
 //! | `topk(k, count_over_time(…))` / `topk(k, sum_over_time(…))` | `Aggregate{[TopK{k}]}` (heavy-hitter intent) over the explicit inner `Aggregate{[Count/Sum]}` |
@@ -590,14 +590,7 @@ fn build_over_subtree(outer: Outer, keys: Vec<ColumnRef>, child: Unresolved) ->
         // `walk_aggregate` always passes a real aggregator; `None` can't occur.
         Outer::None => child,
         Outer::Plain(intent) => outer_aggregate(keys, outer_intent(&intent), child),
-        Outer::Count => outer_aggregate(
-            keys,
-            AggIntent::Cardinality {
-                col: None,
-                accuracy: current_accuracy(),
-            },
-            child,
-        ),
+        Outer::Count => outer_aggregate(keys, count(), child),
         Outer::CountValues { label } => {
             outer_aggregate(keys, AggIntent::CountValues { label }, child)
         }
@@ -1448,11 +1441,11 @@ fn build(inner: Inner, keys: Vec<ColumnRef>, outer: Outer) -> Result<Unresolved>
             }
         }),
         Outer::Count => Ok(match &inner.func {
-            None => windowed_aggregate(inner, keys, cardinality()),
+            None => windowed_aggregate(inner, keys, count()),
             Some(f) => {
                 let inner_i = inner_intent(f);
                 let inner_agg = windowed_aggregate(inner, vec![], inner_i);
-                outer_aggregate(keys, cardinality(), inner_agg)
+                outer_aggregate(keys, count(), inner_agg)
             }
         }),
         Outer::CountValues { label } => Ok(match &inner.func {
@@ -1648,11 +1641,9 @@ fn filtered_source(metric: String, matchers: Vec<Unresolved>, shift: TimeShift)
     }
 }
 
-/// `count(v)` / `count by (…) (v)` — SQL `COUNT(DISTINCT col)`'s PromQL
-/// counterpart, over the (always implicit) sample value.
-fn cardinality() -> AggIntent<ColumnRef> {
-    AggIntent::Cardinality {
-        col: None,
+/// Count vector elements regardless of their sample values.
+fn count() -> AggIntent<ColumnRef> {
+    AggIntent::Count {
         accuracy: current_accuracy(),
     }
 }

crates/frontend-promql/tests/count_planning.rs

@@ -0,0 +1,235 @@
+//! Query text through summary selection: counts use observations, never value weights.
+use std::rc::Rc;
+
+use asap_aware_mapping::{Replacement, ReplacementStrategy, SketchAlgorithmStrategy, TargetSubDAG};
+use asap_frontend_promql::lower_promql;
+use asap_types::post_asap::{
+    compile_executable_dag, ExactKind, ExecutableOperatorPayload, NonNegativeWeightProof,
+    SketchAlgorithm, SummaryExpr, SummaryFamilyType, SummaryInputExpr, WeightDomain,
+};
+use asap_types::types::AccuracyTarget;
+
+// Exact series and temporal counts must select a count accumulator, not distinct or sum.
+#[test]
+fn exact_counts_select_count_accumulators() {
+    for query in ["count(up)", "count by(job)(up)", "count_over_time(up[5m])"] {
+        let root = Rc::new(lower_promql(query, AccuracyTarget::Exact).unwrap());
+        let candidates =
+            SketchAlgorithmStrategy::default_cost_model().replacements(&TargetSubDAG::new(&root));
+        assert!(
+            candidates.iter().any(|candidate| {
+                matches!(&candidate.replacement, Replacement::Summary(node)
+                if matches!(&node.expr, SummaryExpr::SummaryAgg {
+                    family: SummaryFamilyType::ExactAggregate(ExactKind::Count, _), .. }))
+            }),
+            "{query}: {candidates:?}"
+        );
+    }
+}
+
+// CMS/CountSketch count updates must stay +1 even for zero or negative samples.
+#[test]
+fn frequency_count_candidates_use_unit_weights() {
+    for query in ["count_over_time(up[5m])", "count(up)"] {
+        let root = Rc::new(lower_promql(query, AccuracyTarget::Epsilon(0.02)).unwrap());
+        let candidates =
+            SketchAlgorithmStrategy::default_cost_model().replacements(&TargetSubDAG::new(&root));
+        let mut algorithms = Vec::new();
+        for candidate in &candidates {
+            let Replacement::Summary(node) = &candidate.replacement else {
+                continue;
+            };
+            let SummaryExpr::SummaryEstimate { summary_input, .. } = &node.expr else {
+                continue;
+            };
+            let SummaryExpr::SummaryAgg {
+                family: SummaryFamilyType::Sketch(kind, _),
+                input,
+                ..
+            } = &summary_input.expr
+            else {
+                continue;
+            };
+            assert!(
+                !matches!(kind.algorithm(), SketchAlgorithm::Hll),
+                "{query}: {node:?}"
+            );
+            if !matches!(
+                kind.algorithm(),
+                SketchAlgorithm::Cms | SketchAlgorithm::CountSketch
+            ) {
+                continue;
+            }
+            let dag = compile_executable_dag(node).unwrap();
+            assert!(
+                dag.nodes.iter().any(|node| matches!(
+                    &node.payload,
+                    ExecutableOperatorPayload::SummaryAgg { input: actual, .. } if actual == input
+                )),
+                "executable DAG must preserve the count update contract"
+            );
+            algorithms.push(kind.algorithm().clone());
+            assert!(input.item.is_some(), "frequency keys must be explicit");
+            assert_eq!(
+                input.weight,
+                SummaryInputExpr::Constant(1.0),
+                "{query}: {node:?}"
+            );
+            assert_eq!(
+                input.weight_domain,
+                WeightDomain::NonNegative {
+                    proof: NonNegativeWeightProof::UnitCount
+                }
+            );
+        }
+        assert!(
+            algorithms.contains(&SketchAlgorithm::Cms),
+            "{query}: {candidates:?}"
+        );
+        assert!(
+            algorithms.contains(&SketchAlgorithm::CountSketch),
+            "{query}: {candidates:?}"
+        );
+    }
+}
+
+// Fixtures are already selected at one instant or within one five-minute window.
+// This narrow test oracle interprets the emitted aggregate, not Prometheus ingestion,
+// staleness, or scrape scheduling. Unsupported plan shapes fail explicitly.
+fn aggregate_fixture(query: &str, series: &[Vec<f64>]) -> Vec<f64> {
+    use asap_types::pre_asap::{AggIntent, QueryExpr, Reduction};
+    let root = lower_promql(query, AccuracyTarget::Exact).unwrap();
+    let QueryExpr::Aggregate {
+        reduction,
+        measures,
+        child,
+        ..
+    } = &root
+    else {
+        panic!("expected aggregate: {root:?}");
+    };
+    match child.as_ref() {
+        QueryExpr::Scan { .. } => assert!(series.iter().all(|samples| samples.len() == 1)),
+        QueryExpr::TimeRange { range, child } => {
+            assert_eq!(range.as_secs(), 300);
+            assert!(matches!(child.as_ref(), QueryExpr::Scan { .. }));
+        }
+        other => panic!("unsupported fixture input: {other:?}"),
+    }
+    let aggregate = |values: &[f64]| match measures.as_slice() {
+        [AggIntent::Count { .. }] => values.len() as f64,
+        [AggIntent::Cardinality { .. }] => {
+            let mut distinct = values.to_vec();
+            distinct.sort_by(f64::total_cmp);
+            distinct.dedup();
+            distinct.len() as f64
+        }
+        [AggIntent::Sum { .. }] => values.iter().sum(),
+        other => panic!("unsupported fixture aggregate: {other:?}"),
+    };
+    match reduction {
+        Reduction::PerEntity => series.iter().map(|values| aggregate(values)).collect(),
+        Reduction::Reduce(_) => {
+            assert_eq!(reduction, &Reduction::by(vec![]));
+            vec![aggregate(
+                &series.iter().flatten().copied().collect::<Vec<_>>(),
+            )]
+        }
+    }
+}
+
+// Three targets remain three whether healthy, unhealthy, or carrying signed values.
+#[test]
+fn count_up_is_three_independent_of_target_health() {
+    for values in [[1.0, 1.0, 1.0], [1.0, 1.0, 0.0], [-1.0, -1.0, 0.0]] {
+        let series: Vec<_> = values.into_iter().map(|value| vec![value]).collect();
+        assert_eq!(
+            aggregate_fixture("count(up)", &series),
+            vec![3.0],
+            "{values:?}"
+        );
+    }
+}
+
+// Ten scrapes per series count as ten, including all-zero and all-negative series.
+#[test]
+fn count_over_time_counts_scrapes_not_sample_values() {
+    // up{instance="a"} and up{instance="b"}, evaluated in the same window.
+    assert_eq!(
+        aggregate_fixture("count_over_time(up[5m])", &[vec![1.0; 10], vec![0.0; 10]]),
+        vec![10.0, 10.0]
+    );
+    // The http_reqs series is a separate metric and therefore a separate query.
+    assert_eq!(
+        aggregate_fixture(
+            "count_over_time(http_reqs{code=\"200\"}[5m])",
+            &[vec![3.0; 10]]
+        ),
+        vec![10.0]
+    );
+    assert_eq!(
+        aggregate_fixture(
+            "count_over_time(temperature[5m])",
+            &[
+                vec![-3.0; 10],
+                vec![-2.0, 0.0, 2.0, -2.0, 0.0, 2.0, -2.0, 0.0, 2.0, -2.0]
+            ]
+        ),
+        vec![10.0, 10.0]
+    );
+}
+
+// Execute the emitted CMS update-weight expression on the reported values.
+// This checks the planner's numerical update contract, not a sketch-library runtime.
+#[test]
+fn cms_count_updates_total_ten_for_zero_positive_and_negative_samples() {
+    use asap_types::pre_asap::ColumnRef;
+    let root =
+        Rc::new(lower_promql("count_over_time(up[5m])", AccuracyTarget::Epsilon(0.02)).unwrap());
+    let candidates =
+        SketchAlgorithmStrategy::default_cost_model().replacements(&TargetSubDAG::new(&root));
+    let dag = candidates
+        .iter()
+        .find_map(|candidate| {
+            let Replacement::Summary(node) = &candidate.replacement else {
+                return None;
+            };
+            let dag = compile_executable_dag(node).unwrap();
+            dag.nodes
+                .iter()
+                .any(|node| {
+                    matches!(&node.payload,
+            ExecutableOperatorPayload::SummaryAgg { family: SummaryFamilyType::Sketch(kind, _), .. }
+                if kind.algorithm() == &SketchAlgorithm::Cms)
+                })
+                .then_some(dag)
+        })
+        .expect("CMS count candidate");
+    let update = dag
+        .nodes
+        .iter()
+        .find_map(|node| match &node.payload {
+            ExecutableOperatorPayload::SummaryAgg { input, .. } => Some(input),
+            _ => None,
+        })
+        .unwrap();
+    for value in [1.0, 0.0, 3.0, -3.0] {
+        let total: f64 = [value; 10]
+            .into_iter()
+            .map(|sample| {
+                let weight = match &update.weight {
+                    SummaryInputExpr::Constant(value) => *value,
+                    SummaryInputExpr::Column(ColumnRef::SampleValue) => sample,
+                    SummaryInputExpr::Column(ColumnRef::Named(name)) if name == "value" => sample,
+                    other => panic!("unsupported update weight: {other:?}"),
+                };
+                assert!(
+                    weight >= 0.0,
+                    "CMS must not receive a negative update for {sample}"
+                );
+                weight
+            })
+            .sum();
+        assert_eq!(total, 10.0, "ten samples of {value}");
+    }
+}

crates/frontend-promql/tests/promql_conformance.rs

@@ -328,10 +328,10 @@ fn sum_by_groups_via_positional_aggregate() {
 }
 
 #[test]
-fn count_is_cardinality() {
+fn count_is_row_count() {
     assert!(has(&ok("count(up)"), |i| matches!(
         i,
-        AggIntent::Cardinality { .. }
+        AggIntent::Count { .. }
     )));
 }
 
@@ -718,35 +718,29 @@ fn double_unary_negation_nests() {
 }
 
 #[test]
-fn count_maps_to_cardinality_and_inherits_accuracy() {
-    // SEMANTICS (review #2): PromQL `count by (...)` counts distinct series → the
-    // `Cardinality` intent. The workload's AccuracyTarget threads onto it:
-    // `Exact` stays exact (no silent HLL substitution); an approximate target is
-    // carried through for post-ASAP binding to honor. This pins the
-    // intentional count→Cardinality mapping and its accuracy gating.
+fn count_maps_to_count_and_inherits_accuracy() {
+    // Counts preserve the workload accuracy target without counting distinct values.
     let exact = lower_promql("count by (job) (up)", AccuracyTarget::Exact).unwrap();
     assert!(
         has(&exact, |i| matches!(
             i,
-            AggIntent::Cardinality {
-                col: None,
+            AggIntent::Count {
                 accuracy: AccuracyTarget::Exact
             }
         )),
-        "count→Cardinality must stay Exact under AccuracyTarget::Exact, got {:?}",
+        "Count must stay Exact under AccuracyTarget::Exact, got {:?}",
         intents(&exact)
     );
 
     let approx = lower_promql("count by (job) (up)", AccuracyTarget::Epsilon(0.01)).unwrap();
     assert!(
         has(&approx, |i| matches!(
             i,
-            AggIntent::Cardinality {
-                col: None,
+            AggIntent::Count {
                 accuracy: AccuracyTarget::Epsilon(e)
             } if (*e - 0.01).abs() < 1e-9
         )),
-        "count→Cardinality must carry the approximate target, got {:?}",
+        "Count must carry the approximate target, got {:?}",
         intents(&approx)
     );
 }
@@ -2072,7 +2066,7 @@ fn limitk_by_carries_the_grouping_and_composes_in_a_set_op() {
     // the PromqlSeriesSample must be preserved under the set op (it must lower, not reject).
     assert!(has(
         &ok("count(limitk(2, http_requests) and http_requests)"),
-        |i| matches!(i, AggIntent::Cardinality { .. })
+        |i| matches!(i, AggIntent::Count { .. })
     ));
 }