diff --git a/crates/asap-aware-mapping/src/replacement.rs b/crates/asap-aware-mapping/src/replacement.rs index fccc188d..eed755aa 100644 --- a/crates/asap-aware-mapping/src/replacement.rs +++ b/crates/asap-aware-mapping/src/replacement.rs @@ -2128,7 +2128,12 @@ fn realize_value_frequency_summary_input( _reduction: &Reduction, child: &Rc, ) -> 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` shape; `reduction` is what /// tells them apart now. diff --git a/crates/frontend-promql/src/promql.rs b/crates/frontend-promql/src/promql.rs index 6310f129..822987f2 100644 --- a/crates/frontend-promql/src/promql.rs +++ b/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, 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, outer: Outer) -> Result } }), 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, shift: TimeShift) } } -/// `count(v)` / `count by (…) (v)` — SQL `COUNT(DISTINCT col)`'s PromQL -/// counterpart, over the (always implicit) sample value. -fn cardinality() -> AggIntent { - AggIntent::Cardinality { - col: None, +/// Count vector elements regardless of their sample values. +fn count() -> AggIntent { + AggIntent::Count { accuracy: current_accuracy(), } } diff --git a/crates/frontend-promql/tests/count_planning.rs b/crates/frontend-promql/tests/count_planning.rs new file mode 100644 index 00000000..232f1edd --- /dev/null +++ b/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]) -> Vec { + 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::>(), + )] + } + } +} + +// 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}"); + } +} diff --git a/crates/frontend-promql/tests/promql_conformance.rs b/crates/frontend-promql/tests/promql_conformance.rs index 1d6830f5..b724cf3c 100644 --- a/crates/frontend-promql/tests/promql_conformance.rs +++ b/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,22 +718,17 @@ 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) ); @@ -741,12 +736,11 @@ fn count_maps_to_cardinality_and_inherits_accuracy() { 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 { .. }) )); } diff --git a/crates/frontend-promql/tests/promql_lowering.rs b/crates/frontend-promql/tests/promql_lowering.rs index 5cfa32cc..539c90b9 100644 --- a/crates/frontend-promql/tests/promql_lowering.rs +++ b/crates/frontend-promql/tests/promql_lowering.rs @@ -371,12 +371,9 @@ fn count_over_rate_keeps_both_levels() { measures, child, .. } = &qe else { - panic!("expected outer Aggregate{{Cardinality}}, got {qe:?}"); + panic!("expected outer Aggregate{{Count}}, got {qe:?}"); }; - assert!(matches!( - measures.as_slice(), - [AggIntent::Cardinality { .. }] - )); + assert!(matches!(measures.as_slice(), [AggIntent::Count { .. }])); assert!(matches!( child.as_ref(), QueryExpr::Aggregate { measures, .. } if matches!(measures.as_slice(), [AggIntent::Rate]) @@ -385,6 +382,32 @@ fn count_over_rate_keeps_both_levels() { // ── count / cardinality ─────────────────────────────────────────────────────── +// Both selector fast paths and recursive vector expressions count rows, not values. +#[test] +fn count_never_lowers_to_distinct_sample_values() { + for query in [ + "count(up)", + "count by (job) (up)", + "count without (instance) (up)", + "count(up + 1)", + "count(count_over_time(up[5m]))", + "count_over_time(up[5m])", + ] { + let tree = lower(query); + let intents = all_intents(&tree); + assert!( + intents.iter().any(|i| matches!(i, AggIntent::Count { .. })), + "{query}: {tree:?}" + ); + assert!( + !intents + .iter() + .any(|i| matches!(i, AggIntent::Cardinality { .. })), + "{query}: {tree:?}" + ); + } +} + #[test] fn count_over_time_is_count_intent() { let qe = lower("count_over_time(m[5m])"); @@ -399,9 +422,9 @@ fn count_over_time_is_count_intent() { } #[test] -fn outer_count_is_cardinality() { +fn outer_count_counts_series() { // `count by (symbol) (count_over_time(...))`: inner per-series sample count - // over the window (label-preserving), outer cross-series cardinality grouped + // over the window (label-preserving), outer cross-series row count grouped // on a positional `Aggregate.by`. Leaf = [ts, value, symbol] → symbol = col 2. let qe = lower("count by (symbol) (count_over_time(financial_last_trade_price[5m]))"); let QueryExpr::Aggregate { @@ -414,16 +437,13 @@ fn outer_count_is_cardinality() { panic!("expected outer Aggregate grouped by symbol, got {qe:?}"); }; assert_eq!(reduction, &Reduction::by(vec![2])); - assert!(matches!( - measures.as_slice(), - [AggIntent::Cardinality { .. }] - )); + assert!(matches!(measures.as_slice(), [AggIntent::Count { .. }])); // Inner: Aggregate{Count} over TimeRange (per-series count_over_time). let QueryExpr::Aggregate { measures, child, .. } = child.as_ref() else { - panic!("expected Aggregate (count_over_time) under the cardinality, got {child:?}"); + panic!("expected Aggregate (count_over_time) under the outer count, got {child:?}"); }; assert!(matches!(measures.as_slice(), [AggIntent::Count { .. }])); assert!(matches!(child.as_ref(), QueryExpr::TimeRange { .. })); diff --git a/crates/integration-tests/tests/aggregate.rs b/crates/integration-tests/tests/aggregate.rs index 1c9ffb8b..5395eb71 100644 --- a/crates/integration-tests/tests/aggregate.rs +++ b/crates/integration-tests/tests/aggregate.rs @@ -65,15 +65,14 @@ fn q06_sum_by_job() { ); } -// #7 — PromQL `count` is cross-series cardinality, not per-sample Count +// #7 — PromQL `count` counts vector elements regardless of sample values. #[test] -fn q07_count_is_cardinality() { +fn q07_count_is_row_count() { assert_eq!( lower("count(http_requests_total)"), agg( vec![], - AggIntent::Cardinality { - col: None, + AggIntent::Count { accuracy: AccuracyTarget::Exact }, scan("http_requests_total", &[]),