feat: make AggIntent inputs arity-agnostic, and lower multi-column COUNT(DISTINCT)

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

@@ -213,7 +213,7 @@ fn same_intent_except_accuracy(a: &AggIntent, b: &AggIntent) -> bool {
             AggIntent::Quantile { col: c2, q: q2, .. },
         ) => c1 == c2 && q1 == q2,
         (AggIntent::TopK { k: k1, .. }, AggIntent::TopK { k: k2, .. }) => k1 == k2,
-        (AggIntent::Cardinality { col: c1, .. }, AggIntent::Cardinality { col: c2, .. }) => {
+        (AggIntent::Cardinality { cols: c1, .. }, AggIntent::Cardinality { cols: c2, .. }) => {
             c1 == c2
         }
         _ => false,

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

@@ -722,6 +722,16 @@ pub const DEFAULT_DELTA: f64 = 0.01;
 pub fn summary_candidates(intent: &AggIntent) -> &'static [SketchAlgorithm] {
     match intent {
         AggIntent::Quantile { .. } => &[SketchAlgorithm::Kll, SketchAlgorithm::DDSketch],
+        // A distinct-tuple count hashes the whole tuple as one item
+        // (`SummaryInputExpr::Tuple`), which the distinct-count sketches take
+        // unchanged. UnivMon is dropped there: it estimates frequency moments
+        // over a single value stream, and `realize_value_frequency_summary_input`
+        // would feed it one column of the tuple.
+        AggIntent::Cardinality { cols, .. } if cols.len() > 1 => &[
+            SketchAlgorithm::Hll,
+            SketchAlgorithm::Theta,
+            SketchAlgorithm::Kmv,
+        ],
         AggIntent::Cardinality { .. } => &[
             SketchAlgorithm::Hll,
             SketchAlgorithm::Theta,
@@ -1641,6 +1651,9 @@ fn describe_implementation(intent: &AggIntent, implementation: &Implementation)
 pub(crate) fn describe_intent(intent: &AggIntent) -> String {
     match intent {
         AggIntent::Quantile { q, .. } => format!("quantile(q={q})"),
+        AggIntent::Cardinality { cols, .. } if cols.len() > 1 => {
+            format!("cardinality (distinct count over {} columns)", cols.len())
+        }
         AggIntent::Cardinality { .. } => "cardinality (distinct count)".to_string(),
         AggIntent::TopK { k, .. } => format!("top-{k} heavy-hitters"),
         AggIntent::Count { .. } => "count".to_string(),
@@ -2408,6 +2421,14 @@ fn realize_value_frequency_summary_input(
             "value frequency input needs a valid schema",
         );
     };
+    // One item per observation is a single value stream. `summary_candidates`
+    // already withholds UnivMon from a distinct-tuple count; refused here too
+    // so the invariant does not rest on that table alone.
+    if intent.input_cols().len() > 1 {
+        return PhysicalSummaryInputRuleResult::Unsupported(
+            "a value-frequency summary reads a single column",
+        );
+    }
     PhysicalSummaryInputRuleResult::Realized(PhysicalSummaryInput {
         child: Rc::clone(child),
         input: SummaryUpdate {
@@ -2444,7 +2465,11 @@ fn realize_physical_summary_input(
     }
     Ok(PhysicalSummaryInput {
         child: Rc::clone(child),
-        input: SummaryUpdate::column(summarised_column(intent, &child_schema)),
+        input: SummaryUpdate {
+            item: None,
+            weight: summarised_input(intent, &child_schema)?,
+            weight_domain: WeightDomain::UnknownOrSigned,
+        },
     })
 }
 
@@ -2890,24 +2915,62 @@ fn summary_col_index(out_schema: &Schema, by: &[usize], per_series: bool) -> usi
     }
 }
 
-/// The column fed into the summary: the intent's positional input column
-/// resolved to a name against the child schema, or the PromQL sample value.
+/// The column fed into a *single-column* summary: the intent's leading
+/// positional input resolved to a name against the child schema, or the PromQL
+/// sample value when it reads none. Callers are responsible for only reaching
+/// here with a one-column intent — [`summarised_input`] is the general form.
 fn summarised_column(intent: &AggIntent, child_schema: &Schema) -> ColumnRef {
     match intent
-        .input_col()
-        .and_then(|id| child_schema.columns.get(id))
+        .input_cols()
+        .first()
+        .and_then(|id| child_schema.columns.get(*id))
     {
-        Some(c) => match &c.table {
-            Some(t) => ColumnRef::Qualified {
-                table: t.clone(),
-                name: c.name.clone(),
-            },
-            None => ColumnRef::Named(c.name.clone()),
-        },
+        Some(c) => column_ref(c),
         None => ColumnRef::SampleValue,
     }
 }
 
+fn column_ref(column: &asap_types::pre_asap::Column) -> ColumnRef {
+    match &column.table {
+        Some(t) => ColumnRef::Qualified {
+            table: t.clone(),
+            name: column.name.clone(),
+        },
+        None => ColumnRef::Named(column.name.clone()),
+    }
+}
+
+/// What the summary consumes per input row. An intent that reads one column (or
+/// none) feeds that column; `COUNT(DISTINCT a, b)` feeds the whole tuple as one
+/// item, so the distinct-count sketch hashes `(a, b)` rather than `a` — the
+/// difference between tuple cardinality and single-column cardinality.
+///
+/// A tuple leg outside the child schema is an error rather than
+/// [`summarised_column`]'s sample-value fallback: a leg has no sample-value
+/// reading, and silently dropping one would under-count.
+fn summarised_input(
+    intent: &AggIntent,
+    child_schema: &Schema,
+) -> Result<SummaryInputExpr, ImplementError> {
+    let cols = intent.input_cols();
+    if cols.len() < 2 {
+        return Ok(SummaryInputExpr::Column(summarised_column(
+            intent,
+            child_schema,
+        )));
+    }
+    let legs = cols
+        .iter()
+        .map(|id| child_schema.columns.get(*id).map(column_ref))
+        .collect::<Option<Vec<_>>>()
+        .ok_or(ImplementError::PhysicalRealization(
+            "a tuple column is outside the input schema",
+        ))?;
+    Ok(SummaryInputExpr::Tuple(
+        legs.into_iter().map(SummaryInputExpr::Column).collect(),
+    ))
+}
+
 /// The `SummaryEstimate` readout for a summary-bound intent.
 fn readout(
     intent: &AggIntent,
@@ -5937,6 +6000,13 @@ mod tests {
             // approximate-capable, at an ε target → sketch
             (default_quantile(0.99), Sketch(K::Kll)),
             (default_cardinality(), Sketch(K::Hll)),
+            (
+                A::Cardinality {
+                    cols: vec![0, 1],
+                    accuracy: eps(0.01),
+                },
+                Sketch(K::Hll),
+            ),
             (
                 A::Count {
                     accuracy: eps(0.01),
@@ -5961,7 +6031,14 @@ mod tests {
             ),
             (
                 A::Cardinality {
-                    col: None,
+                    cols: vec![],
+                    accuracy: AccuracyTarget::Exact,
+                },
+                Pass,
+            ),
+            (
+                A::Cardinality {
+                    cols: vec![0, 1],
                     accuracy: AccuracyTarget::Exact,
                 },
                 Pass,
@@ -6531,7 +6608,7 @@ mod tests {
         let q = Rc::new(agg(
             vec![2],
             AggIntent::Cardinality {
-                col: None,
+                cols: vec![],
                 accuracy: AccuracyTarget::EpsilonDelta {
                     epsilon: 0.01,
                     delta: 0.01,
@@ -8352,7 +8429,7 @@ mod tests {
     #[test]
     fn explicit_empty_by_aggregate_realizes_summary_agg_with_reduce_reduction() {
         let intent = AggIntent::Cardinality {
-            col: None,
+            cols: vec![],
             accuracy: AccuracyTarget::Epsilon(0.01),
         };
         let q = agg(vec![], intent, metric_scan(&["job"]));
@@ -8408,39 +8485,45 @@ mod tests {
         assert!(matches!(leaf.expr, SummaryExpr::KeepPreAsap(_)));
     }
 
-    /// Issue #115: the summary is built over the intent's own input column.
-    /// Before `Cardinality`/`Quantile` carried `col`, `summarised_column` always
+    /// Issue #115: the summary is built over the intent's own input columns.
+    /// Before `Cardinality`/`Quantile` carried them, `summarised_input` always
     /// fell through to `ColumnRef::SampleValue`, so an HLL was built over the
-    /// wrong column for every SQL `COUNT(DISTINCT c)`.
+    /// wrong column for every SQL `COUNT(DISTINCT c)`. A distinct-tuple count
+    /// hashes the whole tuple as one item — feeding the sketch a single leg
+    /// would report single-column cardinality instead.
     #[test]
-    fn sketch_realizes_over_the_intents_input_column() {
+    fn sketch_realizes_over_the_intents_input_columns() {
         // `metric_scan(&["job"])` → columns [ts=0, value=1, job=2].
+        let column = |name: &str| SummaryInputExpr::Column(ColumnRef::Named(name.into()));
         let cases = [
-            (Some(2), ColumnRef::Named("job".into())),
-            (Some(1), ColumnRef::Named("value".into())),
+            (vec![2], column("job")),
+            (vec![1], column("value")),
             // PromQL convention: no column ⇒ the synthetic sample value.
-            (None, ColumnRef::SampleValue),
+            (vec![], SummaryInputExpr::Column(ColumnRef::SampleValue)),
+            (
+                vec![1, 2],
+                SummaryInputExpr::Tuple(vec![column("value"), column("job")]),
+            ),
         ];
-        for (col, want) in cases {
+        for (cols, want) in cases {
             let intent = AggIntent::Cardinality {
-                col,
+                cols: cols.clone(),
                 accuracy: AccuracyTarget::Epsilon(0.01),
             };
             let root = realize(&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:?}");
+            let bound = find_summary_input(&root)
+                .unwrap_or_else(|| panic!("expected a SummaryAgg for cols={cols:?}"));
+            assert_eq!(bound, want, "wrong summarised input for cols={cols:?}");
         }
     }
 
-    /// The single-column input of the first `SummaryAgg` in the tree.
-    fn find_summary_col(node: &SummaryNode) -> Option<ColumnRef> {
+    /// The update expression of the first `SummaryAgg` in the tree.
+    fn find_summary_input(node: &SummaryNode) -> Option<SummaryInputExpr> {
         match &node.expr {
-            SummaryExpr::SummaryAgg { input, .. } => match &input.weight {
-                SummaryInputExpr::Column(col) if input.item.is_none() => Some(col.clone()),
-                _ => None,
-            },
-            SummaryExpr::SummaryEstimate { summary_input, .. } => find_summary_col(summary_input),
+            SummaryExpr::SummaryAgg { input, .. } if input.item.is_none() => {
+                Some(input.weight.clone())
+            }
+            SummaryExpr::SummaryEstimate { summary_input, .. } => find_summary_input(summary_input),
             _ => None,
         }
     }

crates/frontend-metricsql/src/lib.rs

@@ -201,7 +201,7 @@ impl Lowerer {
             AggregateFunction::Min => AggIntent::Min { col: None },
             AggregateFunction::Max => AggIntent::Max { col: None },
             AggregateFunction::Count => AggIntent::Cardinality {
-                col: None,
+                cols: vec![],
                 accuracy: self.accuracy.clone(),
             },
             AggregateFunction::StdDev => AggIntent::StdDev {

crates/frontend-promql/src/promql.rs

@@ -1659,7 +1659,7 @@ fn inner_intent(f: &InnerFunc) -> AggIntent<ColumnRef> {
             accuracy: current_accuracy(),
         },
         InnerFunc::Cardinality => AggIntent::Cardinality {
-            col: None,
+            cols: vec![],
             accuracy: current_accuracy(),
         },
         InnerFunc::Quantile(q) => AggIntent::Quantile {

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

@@ -1561,7 +1561,8 @@ impl FunctionRewrite for ClickHouseBuiltinRewrite {
             // the ClickHouse name (`argMax`/`argMin`) directly (issue #232).
             RewriteKind::PassThrough => return Ok(Transformed::no(Expr::AggregateFunction(f))),
             // `f(args...)` -> `count(args...) DISTINCT` — `lower_agg_intent`
-            // already maps `count` + `DISTINCT` to `AggIntent::Cardinality`.
+            // already maps `count` + `DISTINCT` to `AggIntent::Cardinality`,
+            // at whatever arity the call carries.
             RewriteKind::CountDistinct => AggregateFunction::new_udf(
                 count_udaf(),
                 f.args,
@@ -1636,13 +1637,6 @@ fn lower_agg_intent(expr: &Expr) -> Result<AggIntent<ColumnRef>, LoweringError>
                     "DISTINCT {name}"
                 )));
             }
-            // Cardinality carries one column; dropping extra DISTINCT arguments
-            // would silently change tuple cardinality into single-column cardinality.
-            if matches!(semantic, AggSemantic::Count) && agg_fn.distinct && agg_fn.args.len() != 1 {
-                return Err(LoweringError::UnsupportedAggregate(
-                    "multi-column COUNT(DISTINCT)".into(),
-                ));
-            }
             // Value reducers (`reducer_col`) require a real column — `SUM(a*b)`
             // is rejected, not silently reduced over a probe column. Quantile
             // and CountDistinct reduce a column too, so they take the same path:
@@ -1672,9 +1666,22 @@ fn lower_agg_intent(expr: &Expr) -> Result<AggIntent<ColumnRef>, LoweringError>
                         right: expr_to_group_ref(right)?,
                     }
                 }
-                AggSemantic::Count if agg_fn.distinct => AggIntent::Cardinality {
-                    col: col(&agg_fn.args)?,
-                    accuracy: current_accuracy(),
+                // Every argument reaches the intent: `COUNT(DISTINCT a, b)`
+                // counts distinct *tuples*, which is a different quantity from
+                // the distinct count of either column.
+                AggSemantic::Count if agg_fn.distinct => match agg_fn.args.as_slice() {
+                    // DataFusion's planner rejects a bare `COUNT(DISTINCT)`
+                    // before lowering. Guarded anyway: an empty `cols` is the
+                    // PromQL sample-value convention, which SQL never has.
+                    [] => {
+                        return Err(LoweringError::UnsupportedAggregate(
+                            "COUNT(DISTINCT) without an argument".into(),
+                        ))
+                    }
+                    args => AggIntent::Cardinality {
+                        cols: args.iter().map(distinct_col).collect::<Result<_, _>>()?,
+                        accuracy: current_accuracy(),
+                    },
                 },
                 AggSemantic::Count => AggIntent::Count {
                     accuracy: current_accuracy(),
@@ -1714,7 +1721,7 @@ fn lower_agg_intent(expr: &Expr) -> Result<AggIntent<ColumnRef>, LoweringError>
                     accuracy: current_accuracy(),
                 },
                 AggSemantic::Cardinality => AggIntent::Cardinality {
-                    col: col(&agg_fn.args)?,
+                    cols: vec![reducer_col(&name, &agg_fn.args)?],
                     accuracy: current_accuracy(),
                 },
             })
@@ -2162,6 +2169,18 @@ fn reducer_col(name: &str, args: &[Expr]) -> Result<ColumnRef, LoweringError> {
     })
 }
 
+/// One argument of a `COUNT(DISTINCT ...)`. Resolved the way a grouping key is
+/// — what is being counted is an identity, and its qualifier has to survive a
+/// join (`a.k` vs `b.k`) — but reported as an aggregate restriction, since an
+/// aggregate call is what the user wrote.
+fn distinct_col(expr: &Expr) -> Result<ColumnRef, LoweringError> {
+    expr_to_group_ref(expr).map_err(|_| {
+        LoweringError::UnsupportedAggregate(
+            "COUNT(DISTINCT ...) over a non-column expression".into(),
+        )
+    })
+}
+
 fn expr_to_group_ref(expr: &Expr) -> Result<ColumnRef, LoweringError> {
     match expr {
         // Preserve the relation qualifier so a GROUP BY / PARTITION BY key over a