diff --git a/canonic/compiler/_helpers.py b/canonic/compiler/_helpers.py index 7c100f1..21fecff 100644 --- a/canonic/compiler/_helpers.py +++ b/canonic/compiler/_helpers.py @@ -213,6 +213,39 @@ def _measure_expr(source: str, measure: Measure) -> exp.Expression: return _qualify_to(parsed, source) +def _guard_aggregate(agg: exp.Expression, condition: exp.Expression) -> exp.Expression: + """Scope one metric's own aggregate to rows matching ``condition`` via conditional + aggregation (``SUM(CASE WHEN condition THEN x END)``), rather than ANDing the + condition into a WHERE shared with sibling metrics in the same flat SELECT — which + would drop rows those sibling metrics still need (GH multi-metric population_filter + collision). + """ + inner = agg.this + if isinstance(inner, exp.Distinct): + guarded = [exp.Case(ifs=[exp.If(this=condition.copy(), true=e)]) for e in inner.expressions] + agg.set("this", exp.Distinct(expressions=guarded)) + elif isinstance(inner, exp.Star): + agg.set("this", exp.Case(ifs=[exp.If(this=condition, true=exp.Literal.number(1))])) + else: + agg.set("this", exp.Case(ifs=[exp.If(this=condition, true=inner)])) + return agg + + +def _requalify_all(expr: exp.Expression, new_alias: str) -> exp.Expression: + """Replace every qualified column's table with ``new_alias``, regardless of its + current table. Used to re-point a per-metric guard condition at a dedup subquery's + bare-named inner projection (``_DEDUP_ALIAS``), where the original source alias no + longer applies. + """ + + def _transform(node: exp.Expression) -> exp.Expression: + if isinstance(node, exp.Column) and node.table: + return exp.column(node.name, table=new_alias) + return node + + return expr.transform(_transform) + + def _from_and_joins( select: exp.Select, owner: str, @@ -239,8 +272,15 @@ def _build_simple( where_conditions: list[exp.Expression], join_edges: list[JoinEdge], sources_by_name: dict[str, SemanticSource], + metric_conditions: list[list[exp.Expression]] | None = None, ) -> exp.Select: - """Single-SELECT emission for the no-fanout case (SPEC §4 step 7).""" + """Single-SELECT emission for the no-fanout case (SPEC §4 step 7). + + ``metric_conditions``, when given, holds one condition list per metric (its own + population_filter/guardrails) applied via conditional aggregation on that metric's + own projection — used when multiple metrics share this SELECT and a shared WHERE + would incorrectly apply one metric's restriction to every other metric's aggregate. + """ select = exp.Select() projections: list[exp.Expression] = [] group_exprs: list[exp.Expression] = [] @@ -248,8 +288,12 @@ def _build_simple( expr = _dimension_expr(src, dim) projections.append(_alias(expr, name)) group_exprs.append(expr) - for m in metrics: - projections.append(_alias(_measure_expr(m.source, m.measure), m.measure.name)) + for i, m in enumerate(metrics): + expr = _measure_expr(m.source, m.measure) + conditions = metric_conditions[i] if metric_conditions else [] + if conditions: + expr = _guard_aggregate(expr, cast("exp.Expression", exp.and_(*conditions))) + projections.append(_alias(expr, m.measure.name)) select = select.select(*projections) select = _from_and_joins(select, owner, join_edges, sources_by_name) if where_conditions: diff --git a/canonic/compiler/strategies/simple_additive.py b/canonic/compiler/strategies/simple_additive.py index e6ea444..b3b1415 100644 --- a/canonic/compiler/strategies/simple_additive.py +++ b/canonic/compiler/strategies/simple_additive.py @@ -38,10 +38,12 @@ _find_time_dim_name, _freshness, _from_and_joins, + _guard_aggregate, _input_columns, _parse, _population_filter_conditions, _qualify_to, + _requalify_all, _resolve_dimensions, _ResolvedMetric, ) @@ -154,18 +156,42 @@ def _compile_simple_additive( logger.debug("stage 5b: restrict_source enforcement") _enforce_restrict_source(query, metrics, resolver, finality_rule, sources_by_name) - # population_filter — defines the population the metric is about (§4.5); before guardrails. - for _, b in raw_bindings: + # population_filter — defines the population each metric is about (§4.5); before guardrails. + # With exactly one metric, its filter/guardrails fold into the single shared WHERE + # (unchanged, current SQL shape). With multiple metrics sharing this one flat SELECT, + # each metric's own filter/guardrails must scope only its own aggregate — ANDing them + # into one shared WHERE would apply metric A's restriction to metric B's aggregate too, + # and vice versa, silently collapsing all metrics to NULL when the restrictions differ. + logger.debug("stage 6: enforcing guardrails") + metric_conditions: list[list[exp.Expression]] | None = None + if len(metrics) == 1: + _, b = raw_bindings[0] where_conditions += _population_filter_conditions( b.binding.canonical.population_filter, sources_by_name, owner, alias_to_source ) - - # Stage 6 — enforce guardrails: AND mandatory filters into WHERE. - logger.debug("stage 6: enforcing guardrails") - guard_conditions, fired = _enforce_guardrails(metrics, resolver, query.context, sources_by_name) + guard_conditions, fired = _enforce_guardrails( + metrics, resolver, query.context, sources_by_name + ) + where_conditions += guard_conditions + else: + metric_conditions = [] + fired = [] + fired_seen: set[str] = set() + for m, (_, b) in zip(metrics, raw_bindings, strict=True): + conds = _population_filter_conditions( + b.binding.canonical.population_filter, sources_by_name, owner, alias_to_source + ) + guard_conds, m_fired = _enforce_guardrails( + [m], resolver, query.context, sources_by_name + ) + conds += guard_conds + for g in m_fired: + if g.id not in fired_seen: + fired_seen.add(g.id) + fired.append(g) + metric_conditions.append(conds) if fired: logger.info("guardrails applied: count=%d ids=%s", len(fired), [g.id for g in fired]) - where_conditions += guard_conditions # Stage 7 — emit SQL through the dialect adapter. logger.debug("stage 7: emitting SQL via dialect %s", dialect) @@ -195,11 +221,23 @@ def _compile_simple_additive( ) elif fanout: ast = _build_deduped( - owner, metrics, dimensions, where_conditions, join_edges, sources_by_name + owner, + metrics, + dimensions, + where_conditions, + join_edges, + sources_by_name, + metric_conditions, ) else: ast = _build_simple( - owner, metrics, dimensions, where_conditions, join_edges, sources_by_name + owner, + metrics, + dimensions, + where_conditions, + join_edges, + sources_by_name, + metric_conditions, ) sql = adapter.emit(ast, limit=query.limit) @@ -231,12 +269,17 @@ def _build_deduped( where_conditions: list[exp.Expression], join_edges: list[JoinEdge], sources_by_name: dict[str, SemanticSource], + metric_conditions: list[list[exp.Expression]] | None = None, ) -> exp.Select: """Fanout-safe emission: dedup the measure grain in an inner ``DISTINCT ON`` subquery. A one→many / many→many join multiplies the measure source's rows; aggregating directly would inflate an additive sum. The inner query keeps one row per grain (Postgres ``DISTINCT ON``); the outer query aggregates over it (SPEC §4 step 4, S3 AC1). + + ``metric_conditions``, when given, holds one condition list per metric applied via + conditional aggregation in the outer query (see ``_build_simple`` for why) — any + column such a condition references must also be projected by the inner subquery. """ owner_source = sources_by_name[owner] grain_cols = [exp.column(g, table=owner) for g in owner_source.grain] @@ -251,6 +294,11 @@ def _build_deduped( for m in metrics: for input_col in _input_columns(m.measure): measure_inputs.setdefault(input_col, exp.column(input_col, table=m.source)) + for conds in metric_conditions or []: + for cond in conds: + for col in cond.find_all(exp.Column): + if col.table: + measure_inputs.setdefault(col.name, exp.column(col.name, table=col.table)) for col_name in sorted(measure_inputs): inner_projections.append(_alias(measure_inputs[col_name], col_name)) inner = inner.select(*inner_projections).distinct(*grain_cols) @@ -266,10 +314,13 @@ def _build_deduped( dim_col = exp.column(name, table=_DEDUP_ALIAS) projections.append(_alias(dim_col, name)) group_exprs.append(dim_col) - for m in metrics: - projections.append( - _alias(_qualify_to(_parse(m.measure.expr), _DEDUP_ALIAS), m.measure.name) - ) + for i, m in enumerate(metrics): + expr = _qualify_to(_parse(m.measure.expr), _DEDUP_ALIAS) + conditions = metric_conditions[i] if metric_conditions else [] + if conditions: + requalified = [_requalify_all(c, _DEDUP_ALIAS) for c in conditions] + expr = _guard_aggregate(expr, cast("exp.Expression", exp.and_(*requalified))) + projections.append(_alias(expr, m.measure.name)) outer = outer.select(*projections).from_(_alias(inner.subquery(), _DEDUP_ALIAS)) if group_exprs: outer = outer.group_by(*group_exprs) diff --git a/tests/compiler/test_simple_additive_multi_metric.py b/tests/compiler/test_simple_additive_multi_metric.py new file mode 100644 index 0000000..99dee4b --- /dev/null +++ b/tests/compiler/test_simple_additive_multi_metric.py @@ -0,0 +1,308 @@ +"""Multi-metric simple/additive compile path — per-metric filter isolation. + +Regression for a bug where compiling several SINGLE-kind metrics together (e.g. +``--metrics total_income,total_expenses,net_cashflow``) returned ``None`` for every +metric, while querying each one individually worked fine. Root cause: every requested +metric's ``population_filter`` and per-measure ``mandatory_filter`` guardrails were +folded into ONE shared WHERE clause covering the single flat SELECT that all metrics +share as sibling projections. When two metrics carry different (e.g. mutually +exclusive) restrictions, the combined WHERE can never match any row, so every +aggregate in that one result row comes back SQL NULL. + +Fix: with more than one requested metric, each metric's own population_filter/ +guardrail conditions scope only that metric's own aggregate via conditional +aggregation (``SUM(CASE WHEN THEN END)``), never a shared WHERE. +A single-metric query is emitted exactly as before (still a plain shared WHERE) — +see the "no-op" and "single-metric shape frozen" tests below. +""" + +from __future__ import annotations + +import duckdb +import pytest +import sqlglot + +from canonic.compiler import SemanticQuery, compile +from canonic.contracts.models import ( + AppliesTo, + CanonicalRef, + Guardrail, + GuardrailKind, + MetricBinding, + Severity, +) +from canonic.contracts.resolver import ContractResolver +from canonic.semantic.models import Column, Dimension, Join, Measure, Relationship, SemanticSource + + +def _parse_ok(sql: str) -> None: + sqlglot.parse_one(sql, dialect="postgres") + + +@pytest.fixture +def transactions() -> SemanticSource: + """Fact table with a type column so population_filter can split income/expense.""" + return SemanticSource( + name="transactions", + connection="warehouse_duckdb", + table="fct_transactions", + grain=["txn_id"], + columns=[ + Column(name="txn_id", type="string", nullable=False), + Column(name="amount", type="decimal", nullable=True), + Column(name="type", type="string", nullable=False), + ], + measures=[Measure(name="amount", expr="sum(amount)", additivity="additive")], + dimensions=[], + ) + + +@pytest.fixture +def total_income_binding() -> MetricBinding: + return MetricBinding( + metric="total_income", + canonical=CanonicalRef( + source="transactions", measure="amount", population_filter="type = 'income'" + ), + ) + + +@pytest.fixture +def total_expenses_binding() -> MetricBinding: + return MetricBinding( + metric="total_expenses", + canonical=CanonicalRef( + source="transactions", measure="amount", population_filter="type = 'expense'" + ), + ) + + +@pytest.fixture +def net_cashflow_binding() -> MetricBinding: + return MetricBinding( + metric="net_cashflow", + canonical=CanonicalRef(source="transactions", measure="amount"), + ) + + +@pytest.fixture +def resolver_multi( + total_income_binding: MetricBinding, + total_expenses_binding: MetricBinding, + net_cashflow_binding: MetricBinding, +) -> ContractResolver: + return ContractResolver( + bindings=[total_income_binding, total_expenses_binding, net_cashflow_binding], + guardrails=[], + ) + + +# --------------------------------------------------------------------------- +# Direct regression: two conflicting population_filters no longer collide +# --------------------------------------------------------------------------- + + +def test_two_conflicting_population_filters_dont_collide_structurally( + resolver_multi: ContractResolver, transactions: SemanticSource +) -> None: + """No shared WHERE; each metric's filter is scoped via its own CASE WHEN.""" + result = compile( + SemanticQuery(metrics=["total_income", "total_expenses"]), resolver_multi, [transactions] + ) + _parse_ok(result.sql) + sql_upper = result.sql.upper() + assert "WHERE" not in sql_upper + assert sql_upper.count("CASE WHEN") == 2 + assert "'income'" in result.sql + assert "'expense'" in result.sql + + +def test_population_filter_mixed_with_metric_without_filter( + resolver_multi: ContractResolver, transactions: SemanticSource +) -> None: + """Three metrics, only two carry a filter: exactly 2 CASE WHEN, third stays plain.""" + result = compile( + SemanticQuery(metrics=["total_income", "total_expenses", "net_cashflow"]), + resolver_multi, + [transactions], + ) + _parse_ok(result.sql) + assert result.sql.upper().count("CASE WHEN") == 2 + assert result.sql.upper().count("SUM(") == 3 + + +def test_multi_metric_execution_no_longer_returns_none( + resolver_multi: ContractResolver, transactions: SemanticSource +) -> None: + """The falsifying test: execute the compiled SQL against real data — no NULLs.""" + result = compile( + SemanticQuery(metrics=["total_income", "total_expenses", "net_cashflow"]), + resolver_multi, + [transactions], + connection_dialects={"warehouse_duckdb": "duckdb"}, + ) + con = duckdb.connect(":memory:") + con.execute("CREATE TABLE fct_transactions (txn_id VARCHAR, amount DECIMAL, type VARCHAR)") + con.execute( + "INSERT INTO fct_transactions VALUES " + "('1', 100, 'income'), ('2', 50, 'income'), ('3', -30, 'expense'), ('4', -20, 'expense')" + ) + row = con.execute(result.sql).fetchone() + assert row is not None + total_income, total_expenses, net_cashflow = row + assert None not in row + assert total_income == 150 + assert total_expenses == -50 + assert net_cashflow == 100 + + +# --------------------------------------------------------------------------- +# Guardrail mandatory_filter suffers (and is fixed for) the same collision +# --------------------------------------------------------------------------- + + +def test_guardrail_mandatory_filter_scoped_per_metric() -> None: + """Two measures, each with a mutually exclusive mandatory_filter guardrail. + + Guardrails key on (source, measure), so each metric needs its own measure to + keep the two guardrails independent. + """ + expense_source = SemanticSource( + name="transactions", + connection="warehouse_duckdb", + table="fct_transactions", + grain=["txn_id"], + columns=[ + Column(name="txn_id", type="string", nullable=False), + Column(name="amount", type="decimal", nullable=True), + Column(name="type", type="string", nullable=False), + ], + measures=[ + Measure(name="income_amount", expr="sum(amount)", additivity="additive"), + Measure(name="expense_amount", expr="sum(amount)", additivity="additive"), + ], + dimensions=[], + ) + income_binding = MetricBinding( + metric="total_income", + canonical=CanonicalRef(source="transactions", measure="income_amount"), + ) + expense_binding = MetricBinding( + metric="total_expenses", + canonical=CanonicalRef(source="transactions", measure="expense_amount"), + ) + income_guardrail = Guardrail( + id="income-only", + applies_to=AppliesTo(source="transactions", measure="income_amount"), + kind=GuardrailKind.MANDATORY_FILTER, + filter="type = 'income'", + severity=Severity.ERROR, + rationale="income_amount must only ever see income rows.", + ) + expense_guardrail = Guardrail( + id="expense-only", + applies_to=AppliesTo(source="transactions", measure="expense_amount"), + kind=GuardrailKind.MANDATORY_FILTER, + filter="type = 'expense'", + severity=Severity.ERROR, + rationale="expense_amount must only ever see expense rows.", + ) + resolver = ContractResolver( + bindings=[income_binding, expense_binding], + guardrails=[income_guardrail, expense_guardrail], + ) + result = compile( + SemanticQuery(metrics=["total_income", "total_expenses"]), resolver, [expense_source] + ) + _parse_ok(result.sql) + sql_upper = result.sql.upper() + assert "WHERE" not in sql_upper + assert sql_upper.count("CASE WHEN") == 2 + assert {g.id for g in result.guardrails_fired} == {"income-only", "expense-only"} + + +# --------------------------------------------------------------------------- +# No-op regression guard: no filters/guardrails at all → unchanged SQL shape +# --------------------------------------------------------------------------- + + +def test_no_filters_multi_metric_unchanged_shape(transactions: SemanticSource) -> None: + """Two plain metrics with no population_filter/guardrails: no CASE WHEN, no WHERE.""" + b1 = MetricBinding( + metric="metric_a", canonical=CanonicalRef(source="transactions", measure="amount") + ) + b2 = MetricBinding( + metric="metric_b", canonical=CanonicalRef(source="transactions", measure="amount") + ) + resolver = ContractResolver(bindings=[b1, b2], guardrails=[]) + result = compile(SemanticQuery(metrics=["metric_a", "metric_b"]), resolver, [transactions]) + _parse_ok(result.sql) + sql_upper = result.sql.upper() + assert "CASE WHEN" not in sql_upper + assert "WHERE" not in sql_upper + assert sql_upper.count("SUM(") == 2 + + +# --------------------------------------------------------------------------- +# Fanout / dedup path also gets per-metric conditional aggregation +# --------------------------------------------------------------------------- + + +@pytest.fixture +def transactions_with_fanout() -> SemanticSource: + """Same transactions source, joined one_to_many to a child table to force dedup.""" + return SemanticSource( + name="transactions", + connection="warehouse_duckdb", + table="fct_transactions", + grain=["txn_id"], + columns=[ + Column(name="txn_id", type="string", nullable=False), + Column(name="amount", type="decimal", nullable=True), + Column(name="type", type="string", nullable=False), + ], + measures=[Measure(name="amount", expr="sum(amount)", additivity="additive")], + dimensions=[], + joins=[ + Join( + to="transaction_tags", + on="transactions.txn_id = transaction_tags.txn_id", + relationship=Relationship.ONE_TO_MANY, + ) + ], + ) + + +@pytest.fixture +def transaction_tags() -> SemanticSource: + return SemanticSource( + name="transaction_tags", + connection="warehouse_duckdb", + table="fct_transaction_tags", + grain=["tag_id"], + columns=[ + Column(name="tag_id", type="string", nullable=False), + Column(name="txn_id", type="string", nullable=False), + Column(name="tag", type="string", nullable=False), + ], + dimensions=[Dimension(name="tag", column="tag")], + ) + + +def test_fanout_dedup_path_conditional_aggregation( + resolver_multi: ContractResolver, + transactions_with_fanout: SemanticSource, + transaction_tags: SemanticSource, +) -> None: + """A one_to_many join forces _build_deduped; per-metric filters still isolate correctly.""" + result = compile( + SemanticQuery(metrics=["total_income", "total_expenses"], dimensions=["tag"]), + resolver_multi, + [transactions_with_fanout, transaction_tags], + ) + _parse_ok(result.sql) + sql_upper = result.sql.upper() + assert "DISTINCT ON" in sql_upper + assert sql_upper.count("CASE WHEN") == 2 + # The filter's source column must be projected by the inner dedup subquery. + assert '"type"' in result.sql.lower() or "type" in result.sql.lower()