fix(l3): thread the input column onto Quantile and Cardinality intents (#115)

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

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

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

crates/frontend-sql/src/sql/mod.rs

@@ -437,9 +437,15 @@ fn lower_agg_item(expr: &Expr) -> Result<AggItem, LoweringError> {
                 )));
             }
             // 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<ColumnId>` 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<AggItem, LoweringError> {
                 ),
                 "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<String> {
     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<ColumnRef, LoweringError> {
     agg_col_name(args).map(ColumnRef::Named).ok_or_else(|| {
         LoweringError::UnsupportedAggregate(format!("{name} over a non-column expression"))

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}"
+        );
+    }
+}

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<ColumnId>` — 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<ColumnId>` — 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<ColumnId>,
         population: bool,
     },
+    /// φ-quantile of `col`. SQL `approx_percentile_cont(col, φ)`; PromQL
+    /// `quantile(φ, …)` leaves `col` as `None` (the sample value).
     Quantile {
+        #[serde(default)]
+        col: Option<ColumnId>,
         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<ColumnId>,
         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<ColumnId> {
         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::<AggIntent>(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::<AggIntent>(legacy).unwrap(),
+            AggIntent::Cardinality {
+                col: None,
+                accuracy: AccuracyTarget::Exact
+            }
+        );
     }
 }

crates/l2/src/lower.rs

@@ -671,10 +671,12 @@ fn agg_func_to_intent(func: &AggFunc, acc: &AccuracyTarget, col: Option<ColumnId
             population: *population,
         },
         AggFunc::Quantile(q) => AggIntent::Quantile {
+            col,
             q: *q,
             accuracy: acc.clone(),
         },
         AggFunc::CountDistinct => AggIntent::Cardinality {
+            col,
             accuracy: acc.clone(),
         },
         AggFunc::HeavyHitters { k } => AggIntent::TopK {

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<ColumnRef> {
+        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();