From d0a86f5f1d576d091e2ac646cfc64907ace097f0 Mon Sep 17 00:00:00 2001 From: Mirko <2361009+mischuh@users.noreply.github.com> Date: Thu, 23 Jul 2026 08:40:39 +0200 Subject: [PATCH] fix(compiler): scope population_filter/guardrails per metric in multi-metric queries Compiling several single-kind metrics in one query (e.g. --metrics total_income,total_expenses,net_cashflow) folded every requested metric's population_filter and mandatory_filter guardrails into one shared WHERE clause covering the flat SELECT they all share. When two metrics carried different restrictions, the combined WHERE could never match any row, so every aggregate came back NULL. With more than one metric requested, each metric's own filter/guardrail conditions now scope only its own aggregate via conditional aggregation (CASE WHEN); single-metric queries are unaffected (still a plain shared WHERE, identical SQL to before). --- canonic/compiler/_helpers.py | 50 ++- .../compiler/strategies/simple_additive.py | 77 ++++- .../test_simple_additive_multi_metric.py | 308 ++++++++++++++++++ 3 files changed, 419 insertions(+), 16 deletions(-) create mode 100644 tests/compiler/test_simple_additive_multi_metric.py 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()