fix(compiler): dedup fanned-out additive leaves in composite metrics

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)
 

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,

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

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
 # ---------------------------------------------------------------------------