From 427a275a1618f6de17ca5ac5ea3ff976a50d5903 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 9 Jul 2026 10:35:09 -0600 Subject: [PATCH] fix(l3): thread the input column onto Quantile and Cardinality intents (#115) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `AggIntent::Quantile` and `AggIntent::Cardinality` carried no input column, so two aggregates over *different* columns lowered to byte-identical L3: approx_percentile_cont(v, 0.5) -> Quantile { q: 0.5, accuracy: Exact } approx_percentile_cont(w, 0.5) -> Quantile { q: 0.5, accuracy: Exact } // equal! Three consequences, all silent: 1. `plan::cse` dedupes on `AggIntent` equality, so a query computing `median(v)` and `median(w)` collapsed to one. 2. `plan::bind` builds the summary over `summarised_column(intent, …)`, which resolves `input_col()`. With no column it always fell through to `ColumnRef::SampleValue`, so every SQL `COUNT(DISTINCT c)` bound its HLL to the wrong column. 3. `approx_percentile_cont(v * 8, 0.95)` succeeded and dropped the expression, because the column slot that would have rejected it did not exist. Root cause was in `l2::lower`: `col` is already in scope and threaded onto Sum/Avg/Min/Max/StdDev/Variance, but was not passed to these two. Adds `col: Option` to both (`#[serde(default)]`, so pre-#115 payloads still deserialize as `None`), threads it through the converter, and includes them in `input_col()`. `None` keeps its meaning: the PromQL sample value. `TopK` deliberately keeps no `col` — it ranks by the aggregate output rather than a base column (#13 / #25). SQL-side, `approx_percentile_cont` / `COUNT(DISTINCT …)` / `approx_distinct` now resolve their argument with `reducer_col`, so an expression argument is rejected like `SUM(a*b)` instead of silently lowering to `col: None`. A SQL query has no sample value to fall back on. This makes `agg_col_ref` dead; it existed only to perform that fallback. Verified: 363 tests pass, `cargo clippy --all-targets` clean. Co-Authored-By: Claude Opus 4.8 (1M context) --- crates/e2e/tests/aggregate.rs | 2 + crates/e2e/tests/time_range.rs | 1 + .../tests/promql_conformance.rs | 2 + crates/frontend-sql/src/sql/mod.rs | 29 +++--- crates/frontend-sql/tests/sql_lowering.rs | 67 ++++++++++++++ crates/ir/src/intent_algebra/agg_intent.rs | 88 +++++++++++++++---- crates/l2/src/lower.rs | 2 + crates/plan/src/bind.rs | 36 +++++++- crates/plan/src/boundary.rs | 12 +-- crates/plan/src/cse.rs | 30 +++++++ 10 files changed, 230 insertions(+), 39 deletions(-) diff --git a/crates/e2e/tests/aggregate.rs b/crates/e2e/tests/aggregate.rs index 2be27dd1..d23b6804 100644 --- a/crates/e2e/tests/aggregate.rs +++ b/crates/e2e/tests/aggregate.rs @@ -71,6 +71,7 @@ fn q07_count_is_cardinality() { agg( vec![], AggIntent::Cardinality { + col: None, accuracy: AccuracyTarget::Exact }, scan("http_requests_total", &[]), @@ -170,6 +171,7 @@ fn q10_quantile_cross_series() { agg( vec![], AggIntent::Quantile { + col: None, q: 0.99, accuracy: AccuracyTarget::Exact }, diff --git a/crates/e2e/tests/time_range.rs b/crates/e2e/tests/time_range.rs index eae8e58c..f9dbe25f 100644 --- a/crates/e2e/tests/time_range.rs +++ b/crates/e2e/tests/time_range.rs @@ -150,6 +150,7 @@ fn q17_quantile_over_time() { range_agg( 300, AggIntent::Quantile { + col: None, q: 0.99, accuracy: AccuracyTarget::Exact }, diff --git a/crates/frontend-promql/tests/promql_conformance.rs b/crates/frontend-promql/tests/promql_conformance.rs index b5b77fb8..eb433b0d 100644 --- a/crates/frontend-promql/tests/promql_conformance.rs +++ b/crates/frontend-promql/tests/promql_conformance.rs @@ -635,6 +635,7 @@ fn count_maps_to_cardinality_and_inherits_accuracy() { has(&exact, |i| matches!( i, AggIntent::Cardinality { + col: None, accuracy: AccuracyTarget::Exact } )), @@ -647,6 +648,7 @@ fn count_maps_to_cardinality_and_inherits_accuracy() { has(&approx, |i| matches!( i, AggIntent::Cardinality { + col: None, accuracy: AccuracyTarget::Epsilon(e) } if (*e - 0.01).abs() < 1e-9 )), diff --git a/crates/frontend-sql/src/sql/mod.rs b/crates/frontend-sql/src/sql/mod.rs index 7775ee80..daf29757 100644 --- a/crates/frontend-sql/src/sql/mod.rs +++ b/crates/frontend-sql/src/sql/mod.rs @@ -437,9 +437,15 @@ fn lower_agg_item(expr: &Expr) -> Result { ))); } // Value reducers (`reducer_col`) require a real column — `SUM(a*b)` - // is rejected, not silently reduced over a probe column. + // is rejected, not silently reduced over a probe column. Quantile + // and CountDistinct reduce a column too, so they take the same path: + // at L3 their `col` is `Option` where `None` means "the + // PromQL sample value", which a SQL query never has. Taking an + // expression here would set `col: None` and silently drop it (#115). let (func, col) = match name.as_str() { - "count" if agg_fn.distinct => (AggFunc::CountDistinct, agg_col_ref(&agg_fn.args)), + "count" if agg_fn.distinct => { + (AggFunc::CountDistinct, reducer_col(&name, &agg_fn.args)?) + } "count" => (AggFunc::Count, ColumnRef::Wildcard), "sum" => (AggFunc::Sum, reducer_col(&name, &agg_fn.args)?), "min" => (AggFunc::Min, reducer_col(&name, &agg_fn.args)?), @@ -463,9 +469,9 @@ fn lower_agg_item(expr: &Expr) -> Result { ), "approx_percentile_cont" | "percentile_cont" => ( AggFunc::Quantile(extract_percentile_q(&agg_fn.args)?), - agg_col_ref(&agg_fn.args), + reducer_col(&name, &agg_fn.args)?, ), - "approx_distinct" => (AggFunc::CountDistinct, agg_col_ref(&agg_fn.args)), + "approx_distinct" => (AggFunc::CountDistinct, reducer_col(&name, &agg_fn.args)?), _ => return Err(LoweringError::UnsupportedAggregate(name)), }; Ok(AggItem { @@ -492,19 +498,10 @@ fn agg_col_name(args: &[Expr]) -> Option { args.first().and_then(col_name) } -/// The aggregated input column. `COUNT(*)` and non-column arguments yield -/// `Wildcard`; a bare/aliased/cast column yields its name. -fn agg_col_ref(args: &[Expr]) -> ColumnRef { - match agg_col_name(args) { - Some(name) => ColumnRef::Named(name), - None => ColumnRef::Wildcard, - } -} - /// The single input column of a value reducer (`SUM`/`MIN`/`MAX`/`AVG`/stddev/ -/// variance). Errors if the argument is not a column: L3 reduces a column, not -/// an arbitrary expression (`SUM(a*b)`), so silently picking a probe column -/// would compute the wrong result. +/// variance/quantile/count-distinct). Errors if the argument is not a column: +/// L3 reduces a column, not an arbitrary expression (`SUM(a*b)`), so silently +/// picking a probe column would compute the wrong result. fn reducer_col(name: &str, args: &[Expr]) -> Result { agg_col_name(args).map(ColumnRef::Named).ok_or_else(|| { LoweringError::UnsupportedAggregate(format!("{name} over a non-column expression")) diff --git a/crates/frontend-sql/tests/sql_lowering.rs b/crates/frontend-sql/tests/sql_lowering.rs index 7ac123fc..72b80803 100644 --- a/crates/frontend-sql/tests/sql_lowering.rs +++ b/crates/frontend-sql/tests/sql_lowering.rs @@ -718,3 +718,70 @@ async fn exists_subquery_in_predicate_is_rejected() { assert!(res.is_err(), "predicate subquery should be rejected: {q}"); } } + +// ── Issue #115: Quantile / Cardinality carry their input column ───────────── + +#[tokio::test] +async fn quantile_carries_its_input_column() { + // `metrics(ts=0, service=1, latency=2, bytes=3)`. Two quantiles over + // different columns must not compare equal — `plan::cse` dedupes on + // `AggIntent` equality, so a col-less intent would collapse them. + let qe = lower( + "SELECT approx_percentile_cont(latency, 0.5), \ + approx_percentile_cont(bytes, 0.5) FROM metrics", + ) + .await; + let (_, aggs) = find_aggregate(&qe).expect("expected an Aggregate"); + assert!( + matches!( + aggs.as_slice(), + [ + AggIntent::Quantile { col: Some(2), .. }, + AggIntent::Quantile { col: Some(3), .. } + ] + ), + "quantiles must bind their own column, got {aggs:?}" + ); + assert_ne!( + aggs[0], aggs[1], + "distinct-column quantiles must not compare equal" + ); +} + +#[tokio::test] +async fn count_distinct_carries_its_input_column() { + let qe = lower("SELECT COUNT(DISTINCT service), COUNT(DISTINCT bytes) FROM metrics").await; + let (_, aggs) = find_aggregate(&qe).expect("expected an Aggregate"); + assert!( + matches!( + aggs.as_slice(), + [ + AggIntent::Cardinality { col: Some(1), .. }, + AggIntent::Cardinality { col: Some(3), .. } + ] + ), + "cardinalities must bind their own column, got {aggs:?}" + ); + assert_ne!( + aggs[0], aggs[1], + "distinct-column cardinalities must not compare equal" + ); +} + +#[tokio::test] +async fn quantile_and_count_distinct_over_an_expression_are_rejected() { + // A SQL aggregate has no "sample value" to fall back on, so an expression + // argument would lower to `col: None` and silently drop the expression. + // Reject it, exactly as `SUM(a*b)` is rejected. + for q in [ + "SELECT approx_percentile_cont(bytes * 8, 0.95) FROM metrics", + "SELECT COUNT(DISTINCT bytes * 8) FROM metrics", + "SELECT approx_distinct(bytes * 8) FROM metrics", + ] { + let res = lower_sql(q, &catalog(), AccuracyTarget::Exact).await; + assert!( + res.is_err(), + "aggregate over an expression must be rejected: {q}" + ); + } +} diff --git a/crates/ir/src/intent_algebra/agg_intent.rs b/crates/ir/src/intent_algebra/agg_intent.rs index 2e6a9158..d955c7b0 100644 --- a/crates/ir/src/intent_algebra/agg_intent.rs +++ b/crates/ir/src/intent_algebra/agg_intent.rs @@ -26,10 +26,11 @@ use crate::types::AccuracyTarget; /// carries only `k` + the accuracy target. /// /// The single-column reducers (`Sum` / `Min` / `Max` / `Avg` / `StdDev` / -/// `Variance`) carry `col: Option` — the positional input column -/// they reduce. `None` is the PromQL convention "the time-series sample -/// value"; SQL `SUM(bytes), AVG(latency)` sets distinct `Some(id)`s so a -/// multi-aggregate node binds each reducer to the right column. +/// `Variance` / `Quantile` / `Cardinality`) carry `col: Option` — the +/// positional input column they reduce. `None` is the PromQL convention "the +/// time-series sample value"; SQL `SUM(bytes), AVG(latency)` sets distinct +/// `Some(id)`s so a multi-aggregate node binds each reducer to the right +/// column, and `plan::bind` knows which column to summarise over (issue #115). #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] #[serde(tag = "kind", rename_all = "snake_case")] pub enum AggIntent { @@ -66,17 +67,28 @@ pub enum AggIntent { col: Option, population: bool, }, + /// φ-quantile of `col`. SQL `approx_percentile_cont(col, φ)`; PromQL + /// `quantile(φ, …)` leaves `col` as `None` (the sample value). Quantile { + #[serde(default)] + col: Option, q: f64, accuracy: AccuracyTarget, }, /// Heavy-hitter top-k to the given accuracy. The group-by keys live on /// the enclosing `Aggregate.by`. + // + // Unlike `Quantile`/`Cardinality`, `TopK` ranks by the *aggregate output* + // rather than a base column, so it carries no `col` — see #13 / #25. TopK { k: usize, accuracy: AccuracyTarget, }, + /// Distinct-value count of `col`. SQL `COUNT(DISTINCT col)`; PromQL + /// `count_values` leaves `col` as `None` (the sample value). Cardinality { + #[serde(default)] + col: Option, accuracy: AccuracyTarget, }, @@ -344,14 +356,17 @@ impl AggIntent { /// The positional input column this intent reduces, if it carries one. /// `None` = the synthetic time-series sample value (PromQL) or an - /// argument-less aggregate (`Count` / `Cardinality` / `TopK`). Used by - /// schema derivation to resolve each reducer's input column. + /// argument-less aggregate (`Count` / `TopK`). Used by schema derivation to + /// resolve each reducer's input column, and by `plan::bind` to pick the + /// column a summary is built over. pub fn input_col(&self) -> Option { match self { AggIntent::Sum { col } | AggIntent::Min { col } | AggIntent::Max { col } | AggIntent::Avg { col } + | AggIntent::Quantile { col, .. } + | AggIntent::Cardinality { col, .. } | AggIntent::StdDev { col, .. } | AggIntent::Variance { col, .. } => *col, _ => None, @@ -549,7 +564,7 @@ pub fn agg_is_exact(op: &AggIntent) -> bool { pub fn agg_accuracy(op: &AggIntent) -> f64 { match op { AggIntent::Quantile { accuracy, .. } - | AggIntent::Cardinality { accuracy } + | AggIntent::Cardinality { accuracy, .. } | AggIntent::Count { accuracy } | AggIntent::TopK { accuracy, .. } => accuracy_target_to_f64(accuracy), _ => 0.0, @@ -564,16 +579,19 @@ fn accuracy_target_to_f64(t: &AccuracyTarget) -> f64 { } } -/// Default `Cardinality` intent — HLL standard error at precision p=14. +/// Default `Cardinality` intent over the sample value — HLL standard error at +/// precision p=14. pub fn default_cardinality() -> AggIntent { AggIntent::Cardinality { + col: None, accuracy: AccuracyTarget::Epsilon(1.04 / ((1u64 << 14) as f64).sqrt()), } } -/// Default `Quantile` intent at φ = `q`, `accuracy = ε 0.01`. +/// Default `Quantile` intent over the sample value at φ = `q`, `accuracy = ε 0.01`. pub fn default_quantile(q: f64) -> AggIntent { AggIntent::Quantile { + col: None, q, accuracy: AccuracyTarget::Epsilon(0.01), } @@ -602,6 +620,7 @@ mod tests { assert_eq!(AggIntent::Sum { col: None }.output_column(&v).name, "sum"); assert_eq!( AggIntent::Quantile { + col: None, q: 0.99, accuracy: AccuracyTarget::Epsilon(0.01) } @@ -683,12 +702,49 @@ mod tests { #[test] fn agg_intent_serde_roundtrip() { - let v = AggIntent::Quantile { - q: 0.99, - accuracy: AccuracyTarget::Epsilon(0.01), - }; - let json = serde_json::to_string(&v).unwrap(); - let back: AggIntent = serde_json::from_str(&json).unwrap(); - assert_eq!(v, back); + for v in [ + AggIntent::Quantile { + col: None, + q: 0.99, + accuracy: AccuracyTarget::Epsilon(0.01), + }, + AggIntent::Quantile { + col: Some(3), + q: 0.99, + accuracy: AccuracyTarget::Epsilon(0.01), + }, + AggIntent::Cardinality { + col: Some(2), + accuracy: AccuracyTarget::Exact, + }, + ] { + let json = serde_json::to_string(&v).unwrap(); + let back: AggIntent = serde_json::from_str(&json).unwrap(); + assert_eq!(v, back); + } + } + + /// `col` is `#[serde(default)]`, so L3 serialized before issue #115 — with + /// no `col` key — still deserializes, as the sample-value convention `None`. + #[test] + fn agg_intent_serde_reads_pre_115_payloads() { + let legacy = r#"{"kind":"quantile","q":0.99,"accuracy":"Exact"}"#; + assert_eq!( + serde_json::from_str::(legacy).unwrap(), + AggIntent::Quantile { + col: None, + q: 0.99, + accuracy: AccuracyTarget::Exact + } + ); + + let legacy = r#"{"kind":"cardinality","accuracy":"Exact"}"#; + assert_eq!( + serde_json::from_str::(legacy).unwrap(), + AggIntent::Cardinality { + col: None, + accuracy: AccuracyTarget::Exact + } + ); } } diff --git a/crates/l2/src/lower.rs b/crates/l2/src/lower.rs index 951f7c6d..225a48eb 100644 --- a/crates/l2/src/lower.rs +++ b/crates/l2/src/lower.rs @@ -671,10 +671,12 @@ fn agg_func_to_intent(func: &AggFunc, acc: &AccuracyTarget, col: Option AggIntent::Quantile { + col, q: *q, accuracy: acc.clone(), }, AggFunc::CountDistinct => AggIntent::Cardinality { + col, accuracy: acc.clone(), }, AggFunc::HeavyHitters { k } => AggIntent::TopK { diff --git a/crates/plan/src/bind.rs b/crates/plan/src/bind.rs index d44428d4..13ed9bc0 100644 --- a/crates/plan/src/bind.rs +++ b/crates/plan/src/bind.rs @@ -354,6 +354,40 @@ mod tests { assert!(matches!(leaf.expr, SummaryExpr::Logical(_))); } + /// Issue #115: the summary is built over the intent's own input column. + /// Before `Cardinality`/`Quantile` carried `col`, `summarised_column` always + /// fell through to `ColumnRef::SampleValue`, so an HLL was built over the + /// wrong column for every SQL `COUNT(DISTINCT c)`. + #[test] + fn sketch_binds_the_intents_input_column() { + // `metric_scan(&["job"])` → columns [ts=0, value=1, job=2]. + let cases = [ + (Some(2), ColumnRef::Named("job".into())), + (Some(1), ColumnRef::Named("value".into())), + // PromQL convention: no column ⇒ the synthetic sample value. + (None, ColumnRef::SampleValue), + ]; + for (col, want) in cases { + let intent = AggIntent::Cardinality { + col, + accuracy: AccuracyTarget::Epsilon(0.01), + }; + let root = bind(&agg(vec![0], intent, metric_scan(&["job"]))).unwrap(); + let bound = find_summary_col(&root) + .unwrap_or_else(|| panic!("expected a SummaryAgg for col={col:?}")); + assert_eq!(bound, want, "wrong summarised column for col={col:?}"); + } + } + + /// The `col` of the first `SummaryAgg` in the tree. + fn find_summary_col(node: &L4Node) -> Option { + match &node.expr { + SummaryExpr::SummaryAgg { col, .. } => Some(col.clone()), + SummaryExpr::SummaryEstimate { sketch_input, .. } => find_summary_col(sketch_input), + _ => None, + } + } + #[test] fn pass_through_intents_stay_logical() { // avg is exact but non-mergeable; histogram_quantile (classic @@ -362,7 +396,7 @@ mod tests { for intent in [ AggIntent::Avg { col: None }, AggIntent::HistogramQuantile { q: 0.99 }, - AggIntent::Quantile { q: 0.99, accuracy: AccuracyTarget::Exact }, + AggIntent::Quantile { col: None, q: 0.99, accuracy: AccuracyTarget::Exact }, ] { let q = agg(vec![2], intent.clone(), metric_scan(&["job"])); let root = bind(&q).unwrap(); diff --git a/crates/plan/src/boundary.rs b/crates/plan/src/boundary.rs index c1d9cd05..fb668979 100644 --- a/crates/plan/src/boundary.rs +++ b/crates/plan/src/boundary.rs @@ -80,7 +80,7 @@ pub fn realize(intent: &AggIntent) -> Realization { match intent { // ── Approximate-capable intents — the AccuracyTarget decides ──────── AggIntent::Quantile { accuracy, .. } - | AggIntent::Cardinality { accuracy } + | AggIntent::Cardinality { accuracy, .. } | AggIntent::Count { accuracy } | AggIntent::TopK { accuracy, .. } => match accuracy { AccuracyTarget::Exact => exact_realization(intent), @@ -298,8 +298,8 @@ mod tests { (A::Count { accuracy: eps(0.01) }, Sketch(K::Cms)), (A::TopK { k: 10, accuracy: eps(0.01) }, Sketch(K::CmsWithHeap)), // the same intents at Exact → exact realization - (A::Quantile { q: 0.5, accuracy: AccuracyTarget::Exact }, Pass), - (A::Cardinality { accuracy: AccuracyTarget::Exact }, Pass), + (A::Quantile { col: None, q: 0.5, accuracy: AccuracyTarget::Exact }, Pass), + (A::Cardinality { col: None, accuracy: AccuracyTarget::Exact }, Pass), (A::Count { accuracy: AccuracyTarget::Exact }, Acc(K::Count)), (A::TopK { k: 10, accuracy: AccuracyTarget::Exact }, Pass), // exact mergeable accumulators @@ -368,7 +368,7 @@ mod tests { #[test] fn accuracy_target_drives_the_boundary() { // Same intent, three targets → three different decisions. - let exact = AggIntent::Quantile { q: 0.99, accuracy: AccuracyTarget::Exact }; + let exact = AggIntent::Quantile { col: None, q: 0.99, accuracy: AccuracyTarget::Exact }; assert_eq!(realize(&exact), Realization::PassThrough); let approx = default_quantile(0.99); // ε = 0.01 @@ -380,7 +380,7 @@ mod tests { } ); - let looser = AggIntent::Quantile { q: 0.99, accuracy: eps(0.05) }; + let looser = AggIntent::Quantile { col: None, q: 0.99, accuracy: eps(0.05) }; assert_eq!( realize(&looser), Realization::Sketch { @@ -465,7 +465,7 @@ mod tests { #[test] fn degenerate_epsilon_saturates_to_tightest_params() { - let intent = AggIntent::Quantile { q: 0.99, accuracy: eps(0.0) }; + let intent = AggIntent::Quantile { col: None, q: 0.99, accuracy: eps(0.0) }; assert_eq!( realize(&intent), Realization::Sketch { diff --git a/crates/plan/src/cse.rs b/crates/plan/src/cse.rs index f3e3d8ba..73bf1c38 100644 --- a/crates/plan/src/cse.rs +++ b/crates/plan/src/cse.rs @@ -157,6 +157,7 @@ mod tests { let q = QueryExpr::Aggregate { by: vec![1].into(), aggs: vec![AggIntent::Quantile { + col: None, q: 0.99, accuracy: AccuracyTarget::Epsilon(0.01), }], @@ -169,11 +170,40 @@ mod tests { assert_eq!(out.roots[0].1, q); } + /// Issue #115: CSE dedupes on `AggIntent` equality. Before `Quantile` + /// carried its input column, `median(a)` and `median(b)` compared equal, so + /// two aggregates over *different* columns collapsed into one — a wrong + /// answer, not just a missed optimisation. + #[test] + fn quantiles_over_different_columns_do_not_dedupe() { + let mk = |col: usize| QueryExpr::Aggregate { + by: vec![1].into(), + aggs: vec![AggIntent::Quantile { + col: Some(col), + q: 0.5, + accuracy: AccuracyTarget::Epsilon(0.01), + }], + output_names: vec![], + having: None, + child: Box::new(windowed_scan()), + }; + let (a, b) = (mk(2), mk(3)); + assert_ne!(a, b, "distinct-column quantiles must not compare equal"); + + // The shared scan is still hoisted; the Aggregate roots stay distinct. + let out = dedupe_subtrees(vec![(QueryId::new("q1"), a), (QueryId::new("q2"), b)]); + assert_ne!( + out.roots[0].1, out.roots[1].1, + "aggregates over different columns must not collapse" + ); + } + #[test] fn dedupe_subtrees_basic() { let mk = |q: f64| QueryExpr::Aggregate { by: vec![1].into(), aggs: vec![AggIntent::Quantile { + col: None, q, accuracy: AccuracyTarget::Epsilon(0.01), }],