diff --git a/canonic/compiler/_helpers.py b/canonic/compiler/_helpers.py index 21fecff..d4e295c 100644 --- a/canonic/compiler/_helpers.py +++ b/canonic/compiler/_helpers.py @@ -303,6 +303,80 @@ def _build_simple( return select +_DEDUP_ALIAS = "_base" + + +def _build_deduped( + owner: str, + metrics: list[_ResolvedMetric], + dimensions: list[tuple[str, Dimension]], + where_conditions: list[exp.Expression], + join_edges: list[JoinEdge], + sources_by_name: dict[str, SemanticSource], + metric_conditions: list[list[exp.Expression]] | None = None, + measure_aliases: list[str] | 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. + + ``measure_aliases``, when given, overrides the outer projection's alias for each + metric (by index) instead of the default ``m.measure.name`` — used by composite.py's + ratio/weighted_avg leaves, which alias their single measure to ``"n"``/``"d"``. + """ + owner_source = sources_by_name[owner] + grain_cols = [exp.column(g, table=owner) for g in owner_source.grain] + + # Inner: DISTINCT ON (grain) projecting dimensions + each measure's input columns. + dim_names = _dimension_output_names(dimensions) + inner = exp.Select() + inner_projections: list[exp.Expression] = [] + measure_inputs: dict[str, exp.Expression] = {} + for (src, dim), name in zip(dimensions, dim_names, strict=True): + inner_projections.append(_alias(_dimension_expr(src, dim), name)) + 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) + inner = _from_and_joins(inner, owner, join_edges, sources_by_name) + if where_conditions: + inner = inner.where(exp.and_(*where_conditions)) + + # Outer: aggregate the deduped rows. + outer = exp.Select() + projections: list[exp.Expression] = [] + group_exprs: list[exp.Expression] = [] + aliases = measure_aliases or [m.measure.name for m in metrics] + for name in dim_names: + dim_col = exp.column(name, table=_DEDUP_ALIAS) + projections.append(_alias(dim_col, name)) + group_exprs.append(dim_col) + 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, aliases[i])) + outer = outer.select(*projections).from_(_alias(inner.subquery(), _DEDUP_ALIAS)) + if group_exprs: + outer = outer.group_by(*group_exprs) + return outer + + def _find_measure(source: SemanticSource, name: str) -> Measure | None: return next((m for m in source.measures if m.name == name), None) diff --git a/canonic/compiler/strategies/composite.py b/canonic/compiler/strategies/composite.py index c14a2c1..ddd4378 100644 --- a/canonic/compiler/strategies/composite.py +++ b/canonic/compiler/strategies/composite.py @@ -28,6 +28,7 @@ _FANOUT, _alias, _bind_filters, + _build_deduped, _build_finality_union, _dimension_expr, _dimension_output_names, @@ -196,6 +197,22 @@ def _plan_leaf( result_flag=finality_rule.result_flag or "per_row", ) used = {r.source for r in finality_rule.realizations} + elif fanout and add is Additivity.ADDITIVE: + # Same fanout-safe dedup simple_additive.py applies to a bare additive metric + # (SPEC §4 step 4, S3 AC1): a one_to_many/many_to_many join to reach the + # requested dimension multiplies this leaf's rows, so a flat SUM/COUNT here + # would inflate the numerator/denominator exactly like an un-deduped `revenue` + # query would. + leaf_select = _build_deduped( + source_name, + [_ResolvedMetric(name=component.metric, source=source_name, measure=measure_obj)], + dimensions, + where_conditions, + join_edges, + sources_by_name, + measure_aliases=[measure_alias], + ) + used = {source_name} | {e.join.to for e in join_edges} else: leaf_select = _build_leaf_select( source_name, diff --git a/canonic/compiler/strategies/simple_additive.py b/canonic/compiler/strategies/simple_additive.py index 9488356..e7f8b18 100644 --- a/canonic/compiler/strategies/simple_additive.py +++ b/canonic/compiler/strategies/simple_additive.py @@ -23,36 +23,29 @@ from canonic.contracts.models import FinalityRule from canonic.contracts.resolver import Binding as ResolverBinding from canonic.contracts.resolver import ContractResolver - from canonic.semantic.models import Dimension, SemanticSource + from canonic.semantic.models import SemanticSource from canonic.compiler._helpers import ( _FANOUT, _TIME_TYPES, _alias, _bind_filters, + _build_deduped, _build_finality_union, _build_simple, - _dimension_expr, _dimension_output_names, _enforce_guardrails, _find_measure, _find_time_dim_name, _freshness, - _from_and_joins, - _guard_aggregate, - _input_columns, _parse, _population_filter_conditions, - _qualify_to, - _requalify_all, _resolve_dimensions, _ResolvedMetric, ) logger = logging.getLogger(__name__) -_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. @@ -509,71 +502,6 @@ def _build_multi_source( return ast, fired, used_sources -def _build_deduped( - owner: str, - metrics: list[_ResolvedMetric], - dimensions: list[tuple[str, Dimension]], - 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] - - # Inner: DISTINCT ON (grain) projecting dimensions + each measure's input columns. - dim_names = _dimension_output_names(dimensions) - inner = exp.Select() - inner_projections: list[exp.Expression] = [] - measure_inputs: dict[str, exp.Expression] = {} - for (src, dim), name in zip(dimensions, dim_names, strict=True): - inner_projections.append(_alias(_dimension_expr(src, dim), name)) - 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) - inner = _from_and_joins(inner, owner, join_edges, sources_by_name) - if where_conditions: - inner = inner.where(exp.and_(*where_conditions)) - - # Outer: aggregate the deduped rows. - outer = exp.Select() - projections: list[exp.Expression] = [] - group_exprs: list[exp.Expression] = [] - for name in dim_names: - dim_col = exp.column(name, table=_DEDUP_ALIAS) - projections.append(_alias(dim_col, name)) - group_exprs.append(dim_col) - 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) - return outer - - def _parse_datetime_literal(lit: str) -> datetime | None: """Try to parse an ISO date or datetime literal; return None if unparseable.""" from datetime import UTC diff --git a/tests/compiler/test_composable.py b/tests/compiler/test_composable.py index 6e06c58..42c8f5c 100644 --- a/tests/compiler/test_composable.py +++ b/tests/compiler/test_composable.py @@ -11,6 +11,7 @@ import re +import duckdb import pytest import sqlglot @@ -61,7 +62,12 @@ def damages() -> SemanticSource: to="vehicles", on="damages.vehicle_id = vehicles.vehicle_id", relationship=Relationship.MANY_TO_ONE, - ) + ), + Join( + to="repair_line_items", + on="damages.damage_id = repair_line_items.damage_id", + relationship=Relationship.ONE_TO_MANY, + ), ], ) @@ -83,8 +89,27 @@ def vehicles() -> SemanticSource: @pytest.fixture -def sources(damages: SemanticSource, vehicles: SemanticSource) -> list[SemanticSource]: - return [damages, vehicles] +def repair_line_items() -> SemanticSource: + """Child table joined one_to_many from damages — fans out the damage grain.""" + return SemanticSource( + name="repair_line_items", + connection="warehouse_pg", + table="fct_repair_line_items", + grain=["line_item_id"], + columns=[ + Column(name="line_item_id", type="string", nullable=False), + Column(name="damage_id", type="string", nullable=False), + Column(name="part_name", type="string", nullable=False), + ], + dimensions=[Dimension(name="part_name", column="part_name")], + ) + + +@pytest.fixture +def sources( + damages: SemanticSource, vehicles: SemanticSource, repair_line_items: SemanticSource +) -> list[SemanticSource]: + return [damages, vehicles, repair_line_items] @pytest.fixture @@ -228,6 +253,69 @@ def test_ac1_sql_structure_numerator_denominator( assert re.search(r"count\(.+damage_id.+\)", result.sql, re.IGNORECASE) +# --------------------------------------------------------------------------- +# Fanout regression — additive leaves must dedup like simple_additive.py does +# (a one_to_many join to reach the requested dimension must not inflate the +# numerator/denominator sums; see the jaffle_shop bug report). +# --------------------------------------------------------------------------- + + +def test_fanout_composite_ratio_leaves_use_distinct_on_dedup( + resolver: ContractResolver, sources: list[SemanticSource] +) -> None: + """Grouping by a one_to_many-reached dimension: both leaves dedup before aggregating.""" + result = compile( + SemanticQuery(metrics=["avg_repair_costs"], dimensions=["part_name"]), + resolver, + sources, + ) + _parse_ok(result.sql) + sql_upper = result.sql.upper() + assert sql_upper.count("DISTINCT ON") == 2 + assert '"damages"."damage_id"' in result.sql + + +def test_fanout_composite_ratio_numeric_correctness( + resolver: ContractResolver, sources: list[SemanticSource] +) -> None: + """Executed regression: a multi-line-item damage must not inflate the ratio. + + damage 'd1' costs 100 and has two repair line items (both 'bumper'), 'd2' costs + 40 and has one. Correct: total_repair_cost=140, damage_count=2 -> ratio=70. A + flat (un-deduped) join fans d1's row out twice, so the buggy numerator sums + 100+100+40=240 and the buggy denominator counts 3 -> ratio=80 — wrong on both + sides, and not just wrong but not even reduced to the correct value. + """ + result = compile( + SemanticQuery(metrics=["avg_repair_costs"], dimensions=["part_name"]), + resolver, + sources, + connection_dialects={"warehouse_pg": "duckdb"}, + ) + con = duckdb.connect(":memory:") + con.execute( + "CREATE TABLE fct_damages (damage_id VARCHAR, vehicle_id VARCHAR, " + "repair_cost DECIMAL, reported_month VARCHAR, warranty_claim INT)" + ) + con.execute( + "CREATE TABLE fct_repair_line_items (line_item_id VARCHAR, damage_id VARCHAR, " + "part_name VARCHAR)" + ) + con.execute( + "INSERT INTO fct_damages VALUES " + "('d1', 'v1', 100, '2024-01', 0), ('d2', 'v1', 40, '2024-01', 0)" + ) + con.execute( + "INSERT INTO fct_repair_line_items VALUES " + "('l1', 'd1', 'bumper'), ('l2', 'd1', 'bumper'), ('l3', 'd2', 'bumper')" + ) + rows = {r[0]: r[1] for r in con.execute(result.sql).fetchall()} + # Correct: total_repair_cost=140, damage_count=2 -> 70. The fanout bug would + # sum repair_cost once per line item (100*2 + 40 = 240) and inflate the count + # (3), landing on 80 instead of 70. + assert rows["bumper"] == pytest.approx(70.0) + + # --------------------------------------------------------------------------- # AC2 — zero denominator behaviour # ---------------------------------------------------------------------------