diff --git a/crates/asap-aware-mapping/src/accuracy_reconciliation.rs b/crates/asap-aware-mapping/src/accuracy_reconciliation.rs index 571e00b2..8ed47576 100644 --- a/crates/asap-aware-mapping/src/accuracy_reconciliation.rs +++ b/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, diff --git a/crates/asap-aware-mapping/src/replacement.rs b/crates/asap-aware-mapping/src/replacement.rs index 9777cf2e..80c03776 100644 --- a/crates/asap-aware-mapping/src/replacement.rs +++ b/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 { + 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::>>() + .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 { + /// The update expression of the first `SummaryAgg` in the tree. + fn find_summary_input(node: &SummaryNode) -> Option { 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, } } diff --git a/crates/frontend-metricsql/src/lib.rs b/crates/frontend-metricsql/src/lib.rs index adc68f4e..4d07f64d 100644 --- a/crates/frontend-metricsql/src/lib.rs +++ b/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 { diff --git a/crates/frontend-promql/src/promql.rs b/crates/frontend-promql/src/promql.rs index 424666aa..1a9bd754 100644 --- a/crates/frontend-promql/src/promql.rs +++ b/crates/frontend-promql/src/promql.rs @@ -1659,7 +1659,7 @@ fn inner_intent(f: &InnerFunc) -> AggIntent { accuracy: current_accuracy(), }, InnerFunc::Cardinality => AggIntent::Cardinality { - col: None, + cols: vec![], accuracy: current_accuracy(), }, InnerFunc::Quantile(q) => AggIntent::Quantile { diff --git a/crates/frontend-sql/src/sql/mod.rs b/crates/frontend-sql/src/sql/mod.rs index 0142dd03..84ca7441 100644 --- a/crates/frontend-sql/src/sql/mod.rs +++ b/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, 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, 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::>()?, + accuracy: current_accuracy(), + }, }, AggSemantic::Count => AggIntent::Count { accuracy: current_accuracy(), @@ -1714,7 +1721,7 @@ fn lower_agg_intent(expr: &Expr) -> Result, 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 { }) } +/// 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 { + 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 { match expr { // Preserve the relation qualifier so a GROUP BY / PARTITION BY key over a diff --git a/crates/frontend-sql/tests/data_quality_check/synthetic_packet_trace.rs b/crates/frontend-sql/tests/data_quality_check/synthetic_packet_trace.rs index 596b6564..273597e7 100644 --- a/crates/frontend-sql/tests/data_quality_check/synthetic_packet_trace.rs +++ b/crates/frontend-sql/tests/data_quality_check/synthetic_packet_trace.rs @@ -186,7 +186,7 @@ struct Tally { } #[tokio::test] -async fn corpus_lowering_rejects_only_unsupported_tuple_counts() { +async fn corpus_lowering_covers_every_query() { let cat = catalog(); let mut t = Tally::default(); for q in queries() { @@ -195,11 +195,6 @@ async fn corpus_lowering_rejects_only_unsupported_tuple_counts() { Ok(_) => t.lowered += 1, // DataFusion surfaces parse/plan failures as `DataFusion(_)`. Err(LoweringError::DataFusion(_)) => t.unparseable += 1, - Err(LoweringError::UnsupportedAggregate(reason)) - if reason == "multi-column COUNT(DISTINCT)" => - { - t.rejected += 1 - } Err(error) => panic!("unexpected lowering failure for {q}: {error}"), } } @@ -214,8 +209,10 @@ async fn corpus_lowering_rejects_only_unsupported_tuple_counts() { "some DQC queries failed to parse/plan: {t:?}" ); - assert_eq!(t.rejected, 9, "expected nine unsupported tuple counts"); - assert_eq!(t.lowered, 61, "SQL lowering coverage changed: {t:?}"); + // The nine tuple counts lower since `COUNT(DISTINCT a, b, ...)` became + // a multi-column `AggIntent::Cardinality`. + assert_eq!(t.rejected, 0, "no query is rejected: {t:?}"); + assert_eq!(t.lowered, 70, "SQL lowering coverage changed: {t:?}"); } impl Tally { @@ -247,18 +244,19 @@ async fn distinct_source_ips_is_cardinality() { } #[tokio::test] -async fn multi_arg_count_distinct_flow_is_rejected() { - // A single-column Cardinality intent cannot represent a distinct 5-tuple. - let error = lower_sql( +async fn multi_arg_count_distinct_flow_counts_the_whole_tuple() { + // Every leg of the distinct 5-tuple reaches the intent, in argument order — + // the whole flow identity is counted, not its first column. + let qe = lower( "SELECT srcport, COUNT(DISTINCT srcip, dstip, srcport, dstport, proto) AS n \ FROM packets GROUP BY srcport ORDER BY n DESC", - &catalog(), - AccuracyTarget::Exact, ) - .await - .unwrap_err(); - assert!(matches!(error, LoweringError::UnsupportedAggregate(reason) - if reason == "multi-column COUNT(DISTINCT)")); + .await; + let (_, measures) = first_aggregate(&qe).expect("expected an Aggregate"); + let [AggIntent::Cardinality { cols, .. }] = measures.as_slice() else { + panic!("expected one Cardinality measure, got {measures:?}"); + }; + assert_eq!(cols, &[0, 1, 2, 3, 4]); } #[tokio::test] diff --git a/crates/frontend-sql/tests/data_quality_check/tpch_deequ.rs b/crates/frontend-sql/tests/data_quality_check/tpch_deequ.rs index ab370ce3..679fbc95 100644 --- a/crates/frontend-sql/tests/data_quality_check/tpch_deequ.rs +++ b/crates/frontend-sql/tests/data_quality_check/tpch_deequ.rs @@ -76,7 +76,7 @@ fn queries() -> Vec<(&'static str, &'static str)> { queries } -// Pin rejected IDs and error reasons so coverage swaps cannot pass the ratchet. +// Pin the rejected IDs (none) so coverage swaps cannot pass the ratchet. #[tokio::test] async fn lowers_the_warehouse_ingestion_check_set() { let cat = catalog(); @@ -92,10 +92,9 @@ async fn lowers_the_warehouse_ingestion_check_set() { } } // P4d and P4l are Pearson correlation checks; they lower since `corr` - // became `AggIntent::PearsonCorr`. - assert_eq!( - rejected, - vec![("P2b", "multi-column COUNT(DISTINCT)".into())] - ); - assert_eq!(lowered, 49); + // became `AggIntent::PearsonCorr`. P2b is the `(l_orderkey, l_linenumber)` + // primary-key check; it lowers since `COUNT(DISTINCT a, b)` became + // a multi-column `AggIntent::Cardinality`. + assert_eq!(rejected, Vec::<(&str, String)>::new()); + assert_eq!(lowered, 50); } diff --git a/crates/frontend-sql/tests/sql_lowering.rs b/crates/frontend-sql/tests/sql_lowering.rs index 0dc17b2c..5482e6f6 100644 --- a/crates/frontend-sql/tests/sql_lowering.rs +++ b/crates/frontend-sql/tests/sql_lowering.rs @@ -5,7 +5,7 @@ //! the shared `resolve_root` produces the positional, resolved canonical //! tree (the same resolver the PromQL path uses). -use asap_frontend_sql::{lower_sql, lower_sql_dialect, SqlCatalog}; +use asap_frontend_sql::{lower_sql, lower_sql_dialect, SqlCatalog, SqlError as LoweringError}; use asap_types::pre_asap::schema::{Column, DataType, Schema}; use asap_types::pre_asap::{ AggIntent, CompareOpKind, GroupKeys, JoinKind, QueryExpr, Reduction, ScalarValue, Source, @@ -175,7 +175,7 @@ fn reducer_input_names(qe: &QueryExpr) -> (Vec, bool) { let schema = child.output_schema().expect("child schema"); let names = measures .iter() - .filter_map(|a| a.input_col()) + .flat_map(|a| a.input_cols()) .map(|id| schema.columns[id].name.clone()) .collect(); (names, matches!(**child, QueryExpr::Project { .. })) @@ -1169,9 +1169,9 @@ async fn count_distinct_carries_its_input_column() { matches!( measures.as_slice(), [ - AggIntent::Cardinality { col: Some(1), .. }, - AggIntent::Cardinality { col: Some(3), .. } - ] + AggIntent::Cardinality { cols: c1, .. }, + AggIntent::Cardinality { cols: c2, .. } + ] if c1 == &[1] && c2 == &[3] ), "cardinalities must bind their own column, got {measures:?}" ); @@ -1195,8 +1195,8 @@ async fn quantile_and_count_distinct_over_an_expression_bind_the_derived_column( let qe = lower(q).await; let (_, measures) = find_aggregate(&qe).expect("expected an Aggregate"); assert!( - measures[0].input_col().is_some(), - "{q} must bind a column, never `col: None`, got {measures:?}" + !measures[0].input_cols().is_empty(), + "{q} must bind a column, never the implicit input, got {measures:?}" ); let (names, materialized) = reducer_input_names(&qe); assert!(materialized, "{q} expected a materializing Project"); @@ -1366,7 +1366,7 @@ async fn a_shared_expression_is_materialized_once() { 1, "the two reducers should share one derived column" ); - assert_eq!(measures[0].input_col(), measures[1].input_col()); + assert_eq!(measures[0].input_cols(), measures[1].input_cols()); } // ── Issue #118: multi-level grouping expands into one Aggregate per level ─── @@ -2465,9 +2465,10 @@ async fn corr_result_is_nullable_float() { assert!(schema.columns[0].nullable); } -// A multi-column DISTINCT must not silently count only the first column. +// A multi-column DISTINCT counts tuples; one column stays the single-column +// intent, so neither form can be mistaken for the other downstream. #[tokio::test] -async fn composite_distinct_is_rejected() { +async fn composite_distinct_counts_tuples() { let cat = SqlCatalog::new().with_table( "t", Schema::new(vec![ @@ -2475,22 +2476,82 @@ async fn composite_distinct_is_rejected() { Column::new("b", DataType::Int64, false), ]), ); - let error = lower_sql( + let composite = lower_sql( "SELECT COUNT(DISTINCT a, b) FROM t", &cat, AccuracyTarget::Exact, ) .await - .unwrap_err(); + .unwrap(); + let QueryExpr::Aggregate { measures, .. } = + find_aggregate_node(&composite).expect("expected an Aggregate") + else { + unreachable!() + }; assert!( - error.to_string().contains("multi-column COUNT(DISTINCT)"), - "{error}" + matches!(measures.as_slice(), [AggIntent::Cardinality { cols, .. }] if cols == &[0, 1]), + "{measures:?}" ); - lower_sql( + + let single = lower_sql( "SELECT COUNT(DISTINCT a) FROM t", &cat, AccuracyTarget::Exact, ) .await .unwrap(); + let QueryExpr::Aggregate { measures, .. } = + find_aggregate_node(&single).expect("expected an Aggregate") + else { + unreachable!() + }; + assert!( + matches!(measures.as_slice(), [AggIntent::Cardinality { cols, .. }] if cols == &[0]), + "{measures:?}" + ); +} + +// An expression argument has no column identity to hash, so it is rejected +// rather than silently reduced over a probe column. +#[tokio::test] +async fn composite_distinct_rejects_expression_arguments() { + let cat = SqlCatalog::new().with_table( + "t", + Schema::new(vec![ + Column::new("a", DataType::Int64, false), + Column::new("b", DataType::Int64, false), + ]), + ); + let error = lower_sql( + "SELECT COUNT(DISTINCT a, b + 1) FROM t", + &cat, + AccuracyTarget::Exact, + ) + .await + .unwrap_err(); + assert!( + matches!(&error, LoweringError::UnsupportedAggregate(reason) + if reason.contains("non-column expression")), + "{error}" + ); +} + +// DISTINCT inputs survive projections introduced by sibling aggregates. +#[tokio::test] +async fn distinct_with_derived_sibling() { + let catalog = SqlCatalog::new().with_table( + "t", + Schema::new(vec![ + Column::new("a", DataType::Int64, false), + Column::new("b", DataType::Int64, false), + ]), + ); + for sql in [ + "SELECT count(DISTINCT a), sum(b + 1) FROM t", + "SELECT count(DISTINCT a, b), sum(b + 1) FROM t", + "SELECT count(DISTINCT a, b), corr(a,b) FROM t", + ] { + let result = lower_sql(sql, &catalog, AccuracyTarget::Exact).await; + assert!(result.is_ok(), "{sql}: {result:?}"); + } } diff --git a/crates/sql-function-catalog/src/lib.rs b/crates/sql-function-catalog/src/lib.rs index df2c0926..bf7dd143 100644 --- a/crates/sql-function-catalog/src/lib.rs +++ b/crates/sql-function-catalog/src/lib.rs @@ -22,7 +22,8 @@ //! resolves (`sum`, `avg`, `approx_percentile_cont`, ...). [`lookup_native`] //! maps one to the [`AggSemantic`] `lower_agg_intent` builds an `AggIntent` //! from. The DISTINCT-modifier rule ("`COUNT DISTINCT` alone maps, to -//! `Cardinality`; reject DISTINCT elsewhere") and the "reducer argument +//! `Cardinality`, over however many columns the call names; reject DISTINCT +//! elsewhere") and the "reducer argument //! must be a bare column" rule are call-site logic, not per-function data, //! and stay in `asap-frontend-sql`. //! - [`CLICKHOUSE_BUILTINS`] -- ClickHouse-only *aggregate* names DataFusion @@ -84,8 +85,9 @@ pub enum Arity { pub enum AggSemantic { /// `COUNT(*)` / `COUNT(x)` -- ignores its argument (always a row count). /// `COUNT(DISTINCT x)` is the one exception the call site special-cases - /// into `Cardinality` instead; that combination is not its own catalog - /// entry (it is the same name, `count`, with a modifier). + /// into `Cardinality` instead (over every argument, so `COUNT(DISTINCT a, b)` + /// counts distinct pairs); that combination is not its own catalog entry + /// (it is the same name, `count`, with a modifier). Count, Sum, Min, diff --git a/crates/types/src/pre_asap/agg_intent.rs b/crates/types/src/pre_asap/agg_intent.rs index 6b020c74..e8789645 100644 --- a/crates/types/src/pre_asap/agg_intent.rs +++ b/crates/types/src/pre_asap/agg_intent.rs @@ -26,7 +26,7 @@ use crate::types::AccuracyTarget; /// carries only `k` + the accuracy target. /// /// The single-column reducers (`Sum` / `Min` / `Max` / `Avg` / `StdDev` / -/// `Variance` / `Quantile` / `Cardinality`) carry `col: Option` — the input +/// `Variance` / `Quantile`) carry `col: Option` — the input /// column they reduce, generic over the column-reference state the same way /// [`QueryExpr`](super::query_expr::QueryExpr) is: positional `ColumnId` once /// bound (the default, and every existing use of the bare `AggIntent` name), @@ -35,9 +35,13 @@ use crate::types::AccuracyTarget; /// `None` is the PromQL convention "the time-series sample value"; SQL /// `SUM(bytes), AVG(latency)` sets distinct `Some(_)`s so a multi-aggregate /// node binds each reducer to the right column, and `plan::bind` knows which -/// column to summarise over (issue #115). +/// column to summarise over (issue #115). `Cardinality` and `PearsonCorr` read +/// more than one column, so they carry their own lists; [`input_cols`] is the +/// arity-agnostic accessor every consumer goes through. +/// +/// [`input_cols`]: AggIntent::input_cols #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] -#[serde(tag = "kind", rename_all = "snake_case")] +#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)] #[serde(bound(serialize = "C: Serialize", deserialize = "C: Deserialize<'de>"))] pub enum AggIntent { // ── Data-model-agnostic ────────────────────────────────────────────── @@ -98,11 +102,14 @@ pub enum AggIntent { k: usize, accuracy: AccuracyTarget, }, - /// Distinct-value count of `col`. SQL `COUNT(DISTINCT col)`; PromQL - /// `count_values` leaves `col` as `None` (the sample value). + /// Distinct count over `cols`, in argument order. One column is SQL + /// `COUNT(DISTINCT col)`; several count distinct *tuples* + /// (`COUNT(DISTINCT a, b)`), which is not the distinct count of any one of + /// them. Empty is the PromQL convention "the sample value" — `count_values` + /// and `distinct_over_time` leave it so. Cardinality { - #[serde(default)] - col: Option, + #[serde(default = "Vec::new")] + cols: Vec, accuracy: AccuracyTarget, }, /// L2 norm of the frequency vector of distinct input values. @@ -475,32 +482,29 @@ impl AggIntent { } impl AggIntent { - /// All explicit value-column dependencies, in argument order. - /// An empty list retains the existing implicit sample/row-count convention. + /// Every value-column dependency, in argument order. The only accessor: + /// an intent's arity is its own business, so no consumer can ask for "the" + /// input column of an aggregate that reads two (`PearsonCorr`, a + /// distinct-tuple `Cardinality`) and silently receive one leg of it. + /// + /// An empty list is the implicit input — the PromQL sample value, or an + /// argument-less aggregate (`Count` / `TopK`). Schema derivation resolves + /// each reducer's input through this, and `plan::bind` picks what a summary + /// is built over from it. pub fn input_cols(&self) -> Vec { - match self { - Self::PearsonCorr { left, right } => vec![left.clone(), right.clone()], - _ => self.input_col().into_iter().collect(), - } - } - - /// The explicit input of a single-column reducer. `PearsonCorr`, - /// argument-less aggregates, and implicit PromQL sample inputs return `None`. - /// Use `input_cols` for dependency tracking; this accessor is for consumers - /// that have already selected a single-column implementation. - 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::FrequencyL2 { col, .. } | AggIntent::FrequencyEntropy { col, .. } | AggIntent::StdDev { col, .. } - | AggIntent::Variance { col, .. } => col.clone(), - _ => None, + | AggIntent::Variance { col, .. } => col.clone().into_iter().collect(), + AggIntent::Cardinality { cols, .. } => cols.clone(), + AggIntent::PearsonCorr { left, right } => vec![left.clone(), right.clone()], + _ => vec![], } } } @@ -700,7 +704,7 @@ fn accuracy_target_to_f64(t: &AccuracyTarget) -> f64 { /// precision p=14. pub fn default_cardinality() -> AggIntent { AggIntent::Cardinality { - col: None, + cols: vec![], accuracy: AccuracyTarget::Epsilon(1.04 / ((1u64 << 14) as f64).sqrt()), } } @@ -728,7 +732,6 @@ mod tests { fn pearson_corr_contract() { let intent = AggIntent::PearsonCorr { left: 2, right: 5 }; assert_eq!(intent.input_cols(), vec![2, 5]); - assert_eq!(intent.input_col(), None); assert!(agg_is_exact(&intent)); assert!(!agg_is_mergeable(&intent)); let output = intent.output_column(&c("x", DataType::Float64)); @@ -739,6 +742,24 @@ mod tests { assert_eq!(serde_json::from_value::(value).unwrap(), intent); } + // A distinct count over a tuple exposes every leg, and reports the same + // output shape as the one-column form — the count of distinct tuples. + #[test] + fn distinct_tuple_cardinality_contract() { + let intent = AggIntent::Cardinality { + cols: vec![2, 5], + accuracy: AccuracyTarget::Epsilon(0.01), + }; + assert_eq!(intent.input_cols(), vec![2, 5]); + assert!(!agg_is_exact(&intent)); + // HLL/Theta/KMV states combine, unlike `Avg`'s finalized value. + assert!(agg_is_mergeable(&intent)); + assert_eq!(agg_accuracy(&intent), 0.01); + let output = intent.output_column(&c("x", DataType::Float64)); + assert_eq!(output.name, "cardinality"); + assert_eq!(output.dtype, DataType::Int64); + } + #[test] fn output_column_names_are_intent_keyed() { let v = c("value", DataType::Float64); @@ -829,20 +850,19 @@ mod tests { } #[test] - fn input_col_tracks_only_reducers() { - assert_eq!(AggIntent::Sum { col: Some(3) }.input_col(), Some(3)); - assert_eq!( - AggIntent::::Avg { col: None }.input_col(), - None, - "None = PromQL sample value" - ); - assert_eq!( - AggIntent::::Count { - accuracy: AccuracyTarget::Exact - } - .input_col(), - None + fn input_cols_tracks_only_reducers() { + assert_eq!(AggIntent::Sum { col: Some(3) }.input_cols(), vec![3]); + assert!( + AggIntent::::Avg { col: None } + .input_cols() + .is_empty(), + "empty = PromQL sample value" ); + assert!(AggIntent::::Count { + accuracy: AccuracyTarget::Exact + } + .input_cols() + .is_empty()); } #[test] @@ -859,7 +879,11 @@ mod tests { accuracy: AccuracyTarget::Epsilon(0.01), }, AggIntent::Cardinality { - col: Some(2), + cols: vec![2], + accuracy: AccuracyTarget::Exact, + }, + AggIntent::Cardinality { + cols: vec![2, 3], accuracy: AccuracyTarget::Exact, }, AggIntent::TopK { @@ -873,6 +897,21 @@ mod tests { } } + // Legacy explicit inputs must fail instead of becoming implicit sample inputs. + #[test] + fn cardinality_rejects_legacy_col_payloads() { + for payload in [ + r#"{"kind":"cardinality","col":2,"accuracy":"Exact"}"#, + r#"{"kind":"cardinality","col":null,"accuracy":"Exact"}"#, + r#"{"kind":"cardinality","col":2,"cols":[1],"accuracy":"Exact"}"#, + ] { + assert!( + serde_json::from_str::(payload).is_err(), + "{payload}" + ); + } + } + /// `col` is `#[serde(default)]`, so a tree serialized before issue #115 — /// with no `col` key — still deserializes, as the sample-value convention `None`. #[test] @@ -900,7 +939,7 @@ mod tests { assert_eq!( serde_json::from_str::(legacy).unwrap(), AggIntent::Cardinality { - col: None, + cols: vec![], accuracy: AccuracyTarget::Exact } ); diff --git a/crates/types/src/pre_asap/query_expr.rs b/crates/types/src/pre_asap/query_expr.rs index 3763a553..7a7d9887 100644 --- a/crates/types/src/pre_asap/query_expr.rs +++ b/crates/types/src/pre_asap/query_expr.rs @@ -1553,9 +1553,13 @@ pub fn aggregate_output_schema( out_cols.push(cnt); continue; } + // Only the output *type* is read from here, so the leading column is + // enough for the multi-column intents: `Cardinality` and `PearsonCorr` + // both have a fixed output type that ignores it. let in_col = intent - .input_col() - .and_then(|id| in_schema.columns.get(id)) + .input_cols() + .first() + .and_then(|id| in_schema.columns.get(*id)) .unwrap_or(&probe); let mut out = intent.output_column(in_col); if let Some((arg, _)) = intent @@ -1625,9 +1629,13 @@ fn without_output_schema( .cloned() .unwrap_or_else(|| Column::new("value", DataType::Float64, false)); for (i, intent) in measures.iter().enumerate() { + // Only the output *type* is read from here, so the leading column is + // enough for the multi-column intents: `Cardinality` and `PearsonCorr` + // both have a fixed output type that ignores it. let in_col = intent - .input_col() - .and_then(|id| in_schema.columns.get(id)) + .input_cols() + .first() + .and_then(|id| in_schema.columns.get(*id)) .unwrap_or(&probe); let mut out = intent.output_column(in_col); if let Some((arg, _)) = intent diff --git a/crates/types/src/pre_asap/resolve.rs b/crates/types/src/pre_asap/resolve.rs index 5d1e722c..b7782ea7 100644 --- a/crates/types/src/pre_asap/resolve.rs +++ b/crates/types/src/pre_asap/resolve.rs @@ -547,8 +547,11 @@ fn resolve_agg_intent( k: *k, accuracy: accuracy.clone(), }, - AggIntent::Cardinality { col: c, accuracy } => AggIntent::Cardinality { - col: col(c)?, + AggIntent::Cardinality { cols, accuracy } => AggIntent::Cardinality { + cols: cols + .iter() + .map(|c| resolve_column_ref(c, schema)) + .collect::>()?, accuracy: accuracy.clone(), }, AggIntent::FrequencyL2 { col: c, accuracy } => AggIntent::FrequencyL2 { @@ -645,6 +648,38 @@ mod tests { assert!(resolve_agg_intent(&missing, &schema).is_err()); } + // Every leg resolves independently, qualifiers included; one unknown leg + // fails rather than silently shortening the tuple. + #[test] + fn resolve_distinct_tuple_columns() { + use crate::pre_asap::{Column, DataType}; + use crate::types::AccuracyTarget; + let schema = Schema::new(vec![ + Column::new("k", DataType::Int64, true).with_table("a"), + Column::new("k", DataType::Int64, true).with_table("b"), + ]); + let qualified = |table: &str| ColumnRef::Qualified { + table: table.into(), + name: "k".into(), + }; + let intent = AggIntent::Cardinality { + cols: vec![qualified("b"), qualified("a")], + accuracy: AccuracyTarget::Exact, + }; + assert_eq!( + resolve_agg_intent(&intent, &schema).unwrap(), + AggIntent::Cardinality { + cols: vec![1, 0], + accuracy: AccuracyTarget::Exact, + } + ); + let missing = AggIntent::Cardinality { + cols: vec![qualified("a"), ColumnRef::Named("missing".into())], + accuracy: AccuracyTarget::Exact, + }; + assert!(resolve_agg_intent(&missing, &schema).is_err()); + } + /// `resolve_root` over a `BinaryOp { , PromqlScalarBridge, vector_match }` /// (issue #220): the bridged scalar operand resolves through the same /// generic walk as every other node (its `Literal` child has no diff --git a/docs/develop_docs/pre-asap-ir.md b/docs/develop_docs/pre-asap-ir.md index 8f7b8c87..15bfc979 100644 --- a/docs/develop_docs/pre-asap-ir.md +++ b/docs/develop_docs/pre-asap-ir.md @@ -106,7 +106,7 @@ native-histogram accessors — this list is representative, not exhaustive: ```text Count, Sum(col), Min(col), Max(col), Avg(col), StdDev(col), Variance(col), -Quantile(col, q), TopK(k), Cardinality(col), PearsonCorr(left, right) // data-model-agnostic +Quantile(col, q), TopK(k), Cardinality(cols), PearsonCorr(left, right) // data-model-agnostic Rate, Increase // counter derivatives Changes, Delta, IDelta, Deriv, Resets, PredictLinear(seconds), DoubleExpSmoothing(sf, tf) // range-vector functions @@ -115,16 +115,38 @@ HistogramStdVar, HistogramFraction(lo, hi), HistogramQuantile(q) // native-hist Math(func) // element-wise transform ``` -`PearsonCorr { left, right }` is the only measure with two value inputs. Both +`PearsonCorr { left, right }` has two value inputs. Both references resolve to positional column IDs, and `input_cols()` exposes both -dependencies — `input_col()` returns `None`, so single-column consumers cannot -pick up half of the pair. SQL lowering projects both arguments, preserving +dependencies. SQL lowering projects both arguments, preserving expressions, casts, and qualified join columns. The result is nullable `Float64`, with pairwise null handling owned by the executing engine. It remains exact: finalized correlation coefficients cannot be combined as scalar rollups, and no sketch or maintained correlation accumulator is selected. Physical costing accepts it as a hash aggregate with provider-supplied accumulator size. +`Cardinality { cols, accuracy }` carries a list, not one column. One entry is +SQL `COUNT(DISTINCT col)`; several count distinct *tuples* +(`COUNT(DISTINCT a, b)`), which is not the distinct count of any one of them. +Empty is the PromQL convention "the sample value" (`count_values`, +`distinct_over_time`). Serialized intents reject unknown fields: legacy +`Cardinality` payloads containing `col` must be migrated to `cols` before loading +(`col: n` becomes `cols: [n]`, and `col: null` becomes `cols: []`). Omitting both +fields still selects the implicit sample input. + +`input_cols()` is the only column accessor on `AggIntent` — an intent's arity is +its own business, so no consumer can ask for "the" input column of an aggregate +that reads two and silently receive one leg of it. That mattered concretely: +before `Cardinality` took a list, SQL lowering dropped every argument after the +first, reporting single-column cardinality as tuple cardinality. + +Realization is the single-column one with a wider item: a tuple becomes a +`SummaryInputExpr::Tuple`, which the distinct-count sketches (HLL, Theta, KMV) +hash as one value. UnivMon is withheld from a tuple — it estimates frequency +moments over a single value stream. At `AccuracyTarget::Exact` the node stays a +logical pass-through at any arity. Each SQL argument must be a bare column, +qualifier preserved so a tuple over a join resolves to the correct side; an +expression argument is rejected rather than reduced over a probe column. + SQL `corr` currently rejects `DISTINCT`, aggregate `FILTER`, aggregate `ORDER BY`, explicit null treatment, and window usage (`OVER`). Further two-input statistics (`covar`, the `regr_*` family) would each add their own variant, following the @@ -365,9 +387,9 @@ topk(3, up) ### Dedup -δ — row-level deduplication (SQL `SELECT DISTINCT`). Distinct from `AggIntent::Cardinality` -(`COUNT(DISTINCT col)`), which collapses to a single number — `Dedup` still returns -multiple rows. +δ — row-level deduplication (SQL `SELECT DISTINCT`). Distinct from +`AggIntent::Cardinality` (`COUNT(DISTINCT col)`, `COUNT(DISTINCT a, b)`), which +collapses to a single number — `Dedup` still returns multiple rows. ```sql SELECT DISTINCT srcip, dstip FROM packets