diff --git a/canonic/compiler/strategies/simple_additive.py b/canonic/compiler/strategies/simple_additive.py index b3b1415..20aae22 100644 --- a/canonic/compiler/strategies/simple_additive.py +++ b/canonic/compiler/strategies/simple_additive.py @@ -15,7 +15,7 @@ FinalityMetadata, ) from canonic.exc import FanoutUnsafe, GuardrailBlock, Unresolved, UnsupportedMeasure -from canonic.semantic.models import Additivity +from canonic.semantic.models import Additivity, Relationship if TYPE_CHECKING: from canonic.compiler.query import SemanticQuery @@ -53,6 +53,32 @@ _DEDUP_ALIAS = "_base" +def _cross_source_measure_unsafe(target: str, owner: str, join_edges: list[JoinEdge]) -> bool: + """True unless every join edge from ``owner`` to ``target`` is one_to_one. + + A flat single-SELECT (or the owner-grain ``DISTINCT ON`` dedup used for fanout) is + only guaranteed correct for a measure whose columns live on ``owner`` itself. A + non-``owner`` measure source reached through a one_to_many/many_to_many edge gets its + rows collapsed by the owner-grain dedup and silently undercounted; reached through a + many_to_one edge, its values get broadcast across every matching owner row and + silently overcounted (confirmed by direct reproduction against real data, both + directions). one_to_one edges duplicate nothing in either direction, so they're the + only relationship safe for combining a non-owner measure into this query today. + """ + if target == owner: + return False + by_alias = {edge.alias: edge for edge in join_edges} + cur = target + while cur != owner: + edge = by_alias.get(cur) + if edge is None: + return True + if edge.join.relationship is not Relationship.ONE_TO_ONE: + return True + cur = edge.from_alias + return False + + def _bindings_to_resolved( name_bindings: list[tuple[str, ResolverBinding]], sources_by_name: dict[str, SemanticSource], @@ -90,6 +116,7 @@ def _compile_simple_additive( alias_to_source = build_alias_tree(owner, sources_by_name) dimensions = _resolve_dimensions(query, sources_by_name, owner, alias_to_source) referenced = {alias for alias, _ in dimensions} + referenced |= {m.source for m in metrics} where_conditions, filter_sources = _bind_filters( query.filters, sources_by_name, owner, alias_to_source ) @@ -115,6 +142,22 @@ def _compile_simple_additive( f"measure {m.source}.{m.measure.name!r} uses an aggregate function " f"not supported at P0" ) + if _cross_source_measure_unsafe(m.source, owner, join_edges): + logger.warning( + "cross-source measure unsafe: %s.%s is reached from owner %s via a " + "non-one_to_one join; combining it in this query would silently " + "corrupt the aggregate", + m.source, + m.measure.name, + owner, + ) + raise FanoutUnsafe( + f"measure {m.source}.{m.measure.name!r} is reached from {owner!r} via a " + f"join that isn't one_to_one; combining it with the other requested " + f"metrics in a single query would either broadcast or collapse its rows " + f"and silently corrupt the aggregate; query it separately or at its " + f"native grain" + ) continue # Non-additive / semi-additive: safety floor — refuse corrupting aggregations. diff --git a/tests/compiler/test_simple_additive_multi_metric.py b/tests/compiler/test_simple_additive_multi_metric.py index 99dee4b..1e3741a 100644 --- a/tests/compiler/test_simple_additive_multi_metric.py +++ b/tests/compiler/test_simple_additive_multi_metric.py @@ -32,6 +32,7 @@ Severity, ) from canonic.contracts.resolver import ContractResolver +from canonic.exc import FanoutUnsafe, Unreachable from canonic.semantic.models import Column, Dimension, Join, Measure, Relationship, SemanticSource @@ -306,3 +307,162 @@ def test_fanout_dedup_path_conditional_aggregation( 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() + + +# --------------------------------------------------------------------------- +# Cross-source metrics: metrics bound to different (but joined) sources +# +# Regression for a bug where the join-planning target set (`referenced` in +# `_compile_simple_additive`) was built only from dimensions and filters, never from +# the source of any non-owner metric. A query combining e.g. `orders.revenue` and +# `order_items.units_sold` compiled to a SELECT that referenced `order_items` without +# ever joining it in, failing at execution time with a DuckDB BinderException. +# +# Fixing the missing join surfaced a second, more dangerous latent issue: once the +# join *is* planned, a one_to_many/many_to_many or many_to_one relationship between +# the two sources makes the existing owner-grain dedup / flat-SELECT emission produce +# a *wrong* (not merely uncompilable) aggregate for the non-owner metric — confirmed by +# direct reproduction. Only one_to_one is safe under the current single-flat-query +# design, so anything else must raise FanoutUnsafe instead of silently corrupting data. +# --------------------------------------------------------------------------- + + +@pytest.fixture +def orders() -> SemanticSource: + return SemanticSource( + name="orders", + connection="warehouse_duckdb", + table="fct_orders", + grain=["order_id"], + columns=[ + Column(name="order_id", type="string", nullable=False), + Column(name="amount", type="decimal", nullable=True), + ], + measures=[Measure(name="revenue", expr="sum(amount)", additivity="additive")], + dimensions=[], + joins=[ + Join( + to="order_details", + on="orders.order_id = order_details.order_id", + relationship=Relationship.ONE_TO_ONE, + ) + ], + ) + + +@pytest.fixture +def order_details() -> SemanticSource: + """A one_to_one companion to ``orders`` (e.g. one shipping record per order).""" + return SemanticSource( + name="order_details", + connection="warehouse_duckdb", + table="fct_order_details", + grain=["order_id"], + columns=[ + Column(name="order_id", type="string", nullable=False), + Column(name="shipping_cost", type="decimal", nullable=True), + ], + measures=[Measure(name="shipping", expr="sum(shipping_cost)", additivity="additive")], + dimensions=[], + ) + + +def test_cross_source_metrics_one_to_one_join_planned_and_correct( + orders: SemanticSource, order_details: SemanticSource +) -> None: + """Two metrics on two one_to_one-joined sources: the join is planned and the SQL is valid.""" + revenue_b = MetricBinding( + metric="revenue", canonical=CanonicalRef(source="orders", measure="revenue") + ) + shipping_b = MetricBinding( + metric="shipping", canonical=CanonicalRef(source="order_details", measure="shipping") + ) + resolver = ContractResolver(bindings=[revenue_b, shipping_b], guardrails=[]) + result = compile( + SemanticQuery(metrics=["revenue", "shipping"]), + resolver, + [order_details, orders], + connection_dialects={"warehouse_duckdb": "duckdb"}, + ) + _parse_ok(result.sql) + assert "JOIN" in result.sql.upper() + assert '"orders"' in result.sql + + con = duckdb.connect(":memory:") + con.execute("CREATE TABLE fct_orders (order_id VARCHAR, amount DECIMAL)") + con.execute("CREATE TABLE fct_order_details (order_id VARCHAR, shipping_cost DECIMAL)") + con.execute("INSERT INTO fct_orders VALUES ('1', 100), ('2', 50)") + con.execute("INSERT INTO fct_order_details VALUES ('1', 5), ('2', 3)") + row = con.execute(result.sql).fetchone() + assert row is not None + revenue, shipping = row + assert revenue == 150 + assert shipping == 8 + + +def test_cross_source_metrics_no_join_path_raises_unreachable( + orders: SemanticSource, transactions: SemanticSource +) -> None: + """Two metrics on sources with no declared join between them: compile-time error, not broken SQL.""" + revenue_b = MetricBinding( + metric="revenue", canonical=CanonicalRef(source="orders", measure="revenue") + ) + amount_b = MetricBinding( + metric="txn_amount", canonical=CanonicalRef(source="transactions", measure="amount") + ) + resolver = ContractResolver(bindings=[revenue_b, amount_b], guardrails=[]) + with pytest.raises(Unreachable): + compile(SemanticQuery(metrics=["revenue", "txn_amount"]), resolver, [orders, transactions]) + + +@pytest.mark.parametrize( + "relationship", + [Relationship.ONE_TO_MANY, Relationship.MANY_TO_ONE, Relationship.MANY_TO_MANY], +) +def test_cross_source_metrics_non_one_to_one_join_raises_fanout_unsafe( + orders: SemanticSource, relationship: Relationship +) -> None: + """A non-one_to_one join between two metric sources must fail loud, not corrupt data. + + A one_to_many join undercounts the many-side metric (owner-grain dedup collapses its + rows); a many_to_one join overcounts the one-side metric (its value is broadcast + across every matching owner row). Both were confirmed by direct reproduction against + real DuckDB data before this guard was added. + """ + order_items = SemanticSource( + name="order_items", + connection="warehouse_duckdb", + table="fct_order_items", + grain=["item_id"], + columns=[ + Column(name="item_id", type="string", nullable=False), + Column(name="order_id", type="string", nullable=False), + Column(name="quantity", type="decimal", nullable=True), + ], + measures=[Measure(name="units_sold", expr="sum(quantity)", additivity="additive")], + dimensions=[], + ) + orders_with_join = orders.model_copy( + update={ + "joins": [ + Join( + to="order_items", + on="orders.order_id = order_items.order_id", + relationship=relationship, + ) + ] + } + ) + revenue_b = MetricBinding( + metric="revenue", canonical=CanonicalRef(source="orders", measure="revenue") + ) + units_b = MetricBinding( + metric="units_sold", canonical=CanonicalRef(source="order_items", measure="units_sold") + ) + resolver = ContractResolver(bindings=[revenue_b, units_b], guardrails=[]) + with pytest.raises(FanoutUnsafe): + compile( + SemanticQuery(metrics=["revenue", "units_sold"]), + resolver, + [orders_with_join, order_items], + )