fix: calculated measures over a joined model's measures

src/boring_semantic_layer/ops/_core.py

@@ -9,7 +9,7 @@
 from typing import TYPE_CHECKING, Any
 
 import ibis
-from attrs import field, frozen
+from attrs import evolve, field, frozen
 from ibis.common.deferred import Deferred
 from ibis.expr import datatypes as dt
 from ibis.expr import operations as ibis_ops
@@ -863,6 +863,24 @@ def _classify_measure(
     except UnknownMeasureRefError:
         raise
     except Exception:
+        if expr_prefer_known is True:
+            # YAML ``calculated_measures`` are expressions over measures by
+            # contract, so never demote one to a raw-table aggregation.
+            # Evaluation fails here when it references a joined model's
+            # measure (``_['customers.n']``) that is not known until the
+            # join is added; it re-evaluates against the joined scope at
+            # query time.
+            return (
+                "calc",
+                CalcMeasure(
+                    expr=expr,
+                    description=description,
+                    requires_unnest=requires_unnest,
+                    prefer_known=prefer_known_set,
+                    metadata=metadata,
+                    unresolved=True,
+                ),
+            )
         # Could not evaluate against the analyzer scope (e.g. lambda
         # uses backend-specific methods MeasureScope didn't reflect).
         # Fall back to base classification — the lambda runs verbatim
@@ -1494,6 +1512,28 @@ def remember_local_name(local_name: str) -> None:
     # so the aggregation produces the columns the calc lambdas read.
     # Walk transitively so calc-of-calc chains pull all needed bases.
     if calc_specs:
+        merged_known = tuple(merged_base_measures) + tuple(merged_calc_measures)
+
+        def _bind_deferred(cm: CalcMeasure) -> CalcMeasure:
+            """Recover ``depends_on`` for a calc that could not be analyzed
+            at definition time (see ``_classify_measure``), now that the
+            merged scope knows every joined model's measures.
+            """
+            if not cm.unresolved:
+                return cm
+            kind, rebound = _classify_measure(
+                cm.expr, MeasureScope(_tbl=base_tbl, _known=merged_known)
+            )
+            if kind != "calc" or not rebound.depends_on:
+                return cm
+            return evolve(
+                cm,
+                depends_on=rebound.depends_on,
+                prefer_known=cm.prefer_known | rebound.depends_on,
+                unresolved=False,
+            )
+
+        calc_specs = {name: _bind_deferred(cm) for name, cm in calc_specs.items()}
 
         def _resolve_dep(ref: str) -> str | None:
             """Resolve a dependency name against base/calc measures.
@@ -1977,6 +2017,38 @@ def _resolve_short_name(
     return resolve_suffix(name, merged_base_measures, merged_calc_measures)
 
 
+def _requested_calc_dependencies(
+    requested: Iterable[str],
+    merged_base_measures: dict,
+    merged_calc_measures: dict,
+) -> set[str] | None:
+    """Collect the measures that requested calc measures transitively read.
+
+    Returns ``None`` when a reached calc is ``unresolved``: it could not be
+    analyzed at definition time (see ``_classify_measure``), so its
+    dependencies are unknown until the plan binds it.
+    """
+    deps: set[str] = set()
+    seen: set[str] = set()
+    worklist = list(requested)
+    while worklist:
+        resolved = _resolve_short_name(worklist.pop(), merged_base_measures, merged_calc_measures)
+        if resolved is None or resolved in seen:
+            continue
+        seen.add(resolved)
+        cm = merged_calc_measures.get(resolved)
+        if not isinstance(cm, CalcMeasure):
+            continue
+        if cm.unresolved:
+            return None
+        for dep in cm.depends_on:
+            dep_resolved = _resolve_short_name(dep, merged_base_measures, merged_calc_measures)
+            if dep_resolved is not None:
+                deps.add(dep_resolved)
+                worklist.append(dep_resolved)
+    return deps
+
+
 def _topological_calc_order(
     calc_specs: dict[str, CalcMeasure],
     base_tbl,
@@ -3281,10 +3353,18 @@ def collect_filters_to_join(node):
                 for name, cardinality in join_tree_info.table_cardinalities.items()
                 if cardinality == "root"
             }
-            requested_from_non_root = any(
-                "." in name and name.split(".", 1)[0] not in root_names for name in self.aggs
-            )
             merged_base_for_routing = _get_merged_fields(all_roots, "measures")
+            # A calc measure is only as fan-out-safe as the measures it
+            # reads, so route on its dependencies as well as its name.
+            calc_deps = _requested_calc_dependencies(
+                self.aggs,
+                merged_base_for_routing,
+                _get_merged_fields(all_roots, "calc_measures"),
+            )
+            requested_from_non_root = calc_deps is None or any(
+                "." in name and name.split(".", 1)[0] not in root_names
+                for name in (*self.aggs, *calc_deps)
+            )
             requested_wrapper_base = any(
                 "." not in name and name in merged_base_for_routing for name in self.aggs
             )
@@ -5212,6 +5292,7 @@ def qualify(name: str) -> str:
         depends_on=qualified_dependencies,
         prefer_known=qualified_preferred,
         metadata=calc.metadata,
+        unresolved=calc.unresolved,
     )
 
 

src/boring_semantic_layer/ops/_values.py

@@ -270,6 +270,8 @@ class CalcMeasure:
     actual base table. ``depends_on`` is captured at classification time
     so the planner can auto-include base-measure dependencies in
     aggregations even when the user did not request them explicitly.
+    ``unresolved`` marks a calc whose analysis had to wait until query time
+    (it reads a joined model's measure), so ``depends_on`` is not yet known.
     """
 
     expr: Any  # callable | Deferred
@@ -278,6 +280,7 @@ class CalcMeasure:
     depends_on: frozenset[str] = field(factory=frozenset, converter=frozenset)
     prefer_known: frozenset[str] = field(factory=frozenset, converter=frozenset)
     metadata: Mapping[str, Any] = field(factory=dict, eq=False, hash=False)
+    unresolved: bool = False
 
     def to_json(self) -> Mapping[str, Any]:
         base = {"description": self.description}
@@ -288,4 +291,12 @@ def to_json(self) -> Mapping[str, Any]:
         return base
 
     def __hash__(self) -> int:
-        return hash((self.description, self.requires_unnest, self.depends_on, self.prefer_known))
+        return hash(
+            (
+                self.description,
+                self.requires_unnest,
+                self.depends_on,
+                self.prefer_known,
+                self.unresolved,
+            )
+        )

src/boring_semantic_layer/tests/test_bi_traps.py

@@ -2304,3 +2304,42 @@ def test_group_by_nested_dim(self, models):
         assert elec["orders.total_amount"].iloc[0] == 400
         assert books["orders.order_count"].iloc[0] == 1
         assert books["orders.total_amount"].iloc[0] == 200
+
+
+# ---------------------------------------------------------------------------
+# TestCalcOverJoinedMeasure
+# ---------------------------------------------------------------------------
+class TestCalcOverJoinedMeasure:
+    """A calc measure reading a joined model's measure must not fan out.
+
+    orders (5 rows) join_one customers (3 rows); only customers 1 and 2
+    have orders, so ``customers.n`` is 2 over the join, not 5.
+    """
+
+    @pytest.fixture()
+    def orders(self):
+        con = ibis.duckdb.connect(":memory:")
+        orders_tbl = con.create_table(
+            "orders",
+            pd.DataFrame({"customer_id": [1, 1, 1, 2, 2], "amount": [10, 20, 30, 40, 50]}),
+        )
+        customers_tbl = con.create_table("customers", pd.DataFrame({"id": [1, 2, 3]}))
+        customers = to_semantic_table(customers_tbl, name="customers").with_measures(
+            n=lambda t: t.count()
+        )
+        return (
+            to_semantic_table(orders_tbl, name="orders")
+            .with_measures(revenue=lambda t: t.amount.sum())
+            .join_one(customers, on=lambda o, c: o.customer_id == c.id)
+        )
+
+    def test_calc_requested_alone(self, orders):
+        model = orders.with_measures(ratio=lambda t: t.revenue / t["customers.n"])
+        df = model.aggregate("ratio").execute()
+        assert df["ratio"].iloc[0] == pytest.approx(150 / 2)
+
+    def test_calc_matches_requesting_dependencies(self, orders):
+        model = orders.with_measures(ratio=lambda t: t.revenue / t["customers.n"])
+        alone = model.aggregate("ratio").execute()
+        with_deps = model.aggregate("ratio", "orders.revenue", "customers.n").execute()
+        assert alone["ratio"].iloc[0] == pytest.approx(with_deps["ratio"].iloc[0])

src/boring_semantic_layer/tests/test_yaml.py

@@ -1491,6 +1491,62 @@ def test_yaml_calculated_measures_prefer_same_named_measures(duckdb_conn):
     assert df["avg_required_per_cell"].tolist() == pytest.approx([6.0, 5.0])
 
 
+@pytest.mark.parametrize(
+    "expr",
+    [
+        "_.revenue / _['customers.n']",
+        "_['orders.revenue'] / _['customers.n']",
+    ],
+)
+def test_yaml_calculated_measures_over_joined_model(duckdb_conn, expr):
+    """A calculated measure may read a joined model's measure.
+
+    Only customers 1 and 2 have orders, so ``customers.n`` is 2 (not the
+    5 order rows), and the ratio is 150 / 2.
+    """
+    orders = duckdb_conn.create_table(
+        "calc_join_orders",
+        {"customer_id": [1, 1, 1, 2, 2], "amount": [10, 20, 30, 40, 50]},
+    )
+    customers = duckdb_conn.create_table("calc_join_customers", {"id": [1, 2, 3]})
+
+    config = {
+        "customers": {
+            "table": "calc_join_customers",
+            "dimensions": {"id": "_.id"},
+            "measures": {"n": "_.count()"},
+        },
+        "orders": {
+            "table": "calc_join_orders",
+            "dimensions": {"customer_id": "_.customer_id"},
+            "measures": {"revenue": "_.amount.sum()"},
+            "calculated_measures": {"revenue_per_customer": expr},
+            "joins": {
+                "customers": {
+                    "model": "customers",
+                    "type": "one",
+                    "left_on": "customer_id",
+                    "right_on": "id",
+                },
+            },
+        },
+    }
+
+    model = from_config(
+        config,
+        tables={"calc_join_orders": orders, "calc_join_customers": customers},
+    )["orders"]
+    df = model.query(measures=("revenue_per_customer",)).execute()
+    assert df["revenue_per_customer"].iloc[0] == pytest.approx(75.0)
+
+    df = (
+        model.query(dimensions=("customer_id",), measures=("orders.revenue_per_customer",))
+        .execute()
+        .sort_values("customer_id")
+    )
+    assert df["orders.revenue_per_customer"].tolist() == pytest.approx([60.0, 90.0])
+
+
 # ---------------------------------------------------------------------------
 # Issue #115: .all() for window aggregations in YAML
 # ---------------------------------------------------------------------------