From 817c4f5b6476ab7ad249c6e22d15445a13f9fd1e Mon Sep 17 00:00:00 2001 From: Deepyaman Datta Date: Thu, 8 Oct 2026 22:32:50 -0600 Subject: [PATCH 1/2] fix: route calc measures via deps to avoid fan-out A calc measure that reads a joined model's measure, e.g. `t.revenue / t["customers.n"]` on orders join_one customers, was routed by its own (root-owned) name onto the flattened-join path, so `customers.n` was counted over fanned-out order rows. Requesting the calc alone returned a silently wrong answer; requesting it alongside its dependencies went through source pre-aggregation and was correct. Route on the transitive dependencies of requested calc measures too. Assisted-by: Claude Opus 5.5 --- src/boring_semantic_layer/ops/_core.py | 37 +++++++++++++++++- .../tests/test_bi_traps.py | 39 +++++++++++++++++++ 2 files changed, 74 insertions(+), 2 deletions(-) diff --git a/src/boring_semantic_layer/ops/_core.py b/src/boring_semantic_layer/ops/_core.py index 34a3bbf7..fab444e0 100644 --- a/src/boring_semantic_layer/ops/_core.py +++ b/src/boring_semantic_layer/ops/_core.py @@ -1977,6 +1977,31 @@ 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]: + """Collect the measures that requested calc measures transitively read.""" + 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 + 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 +3306,18 @@ def collect_filters_to_join(node): for name, cardinality in join_tree_info.table_cardinalities.items() if cardinality == "root" } + 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 = any( - "." in name and name.split(".", 1)[0] not in root_names for name in self.aggs + "." in name and name.split(".", 1)[0] not in root_names + for name in (*self.aggs, *calc_deps) ) - merged_base_for_routing = _get_merged_fields(all_roots, "measures") requested_wrapper_base = any( "." not in name and name in merged_base_for_routing for name in self.aggs ) diff --git a/src/boring_semantic_layer/tests/test_bi_traps.py b/src/boring_semantic_layer/tests/test_bi_traps.py index 3c6f3739..a620798a 100644 --- a/src/boring_semantic_layer/tests/test_bi_traps.py +++ b/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]) From e486fc2a94fbfd2cce101b37465ee2d1f5ff6f9a Mon Sep 17 00:00:00 2001 From: Deepyaman Datta Date: Thu, 8 Oct 2026 22:34:07 -0600 Subject: [PATCH 2/2] fix: allow YAML calc measures over joined measures `from_config` attaches calculated_measures before joins, so an expression such as `_.revenue / _['customers.n']` could not be analyzed at definition time. Classification then demoted it to a base measure, which ran against the raw table at query time and raised DefinitionError. YAML calculated_measures are measure expressions by contract, so keep them as calc measures marked `unresolved` instead; the aggregation plan recovers their dependencies against the joined scope, and the router sends them through source pre-aggregation so they cannot fan out. Assisted-by: Claude Opus 5.5 --- src/boring_semantic_layer/ops/_core.py | 56 ++++++++++++++++++-- src/boring_semantic_layer/ops/_values.py | 13 ++++- src/boring_semantic_layer/tests/test_yaml.py | 56 ++++++++++++++++++++ 3 files changed, 120 insertions(+), 5 deletions(-) diff --git a/src/boring_semantic_layer/ops/_core.py b/src/boring_semantic_layer/ops/_core.py index fab444e0..7a30cc99 100644 --- a/src/boring_semantic_layer/ops/_core.py +++ b/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. @@ -1981,8 +2021,13 @@ def _requested_calc_dependencies( requested: Iterable[str], merged_base_measures: dict, merged_calc_measures: dict, -) -> set[str]: - """Collect the measures that requested calc measures transitively read.""" +) -> 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) @@ -1994,6 +2039,8 @@ def _requested_calc_dependencies( 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: @@ -3314,7 +3361,7 @@ def collect_filters_to_join(node): merged_base_for_routing, _get_merged_fields(all_roots, "calc_measures"), ) - requested_from_non_root = any( + 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) ) @@ -5245,6 +5292,7 @@ def qualify(name: str) -> str: depends_on=qualified_dependencies, prefer_known=qualified_preferred, metadata=calc.metadata, + unresolved=calc.unresolved, ) diff --git a/src/boring_semantic_layer/ops/_values.py b/src/boring_semantic_layer/ops/_values.py index 2054132b..1128c756 100644 --- a/src/boring_semantic_layer/ops/_values.py +++ b/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, + ) + ) diff --git a/src/boring_semantic_layer/tests/test_yaml.py b/src/boring_semantic_layer/tests/test_yaml.py index 984775f1..f5ec46d5 100644 --- a/src/boring_semantic_layer/tests/test_yaml.py +++ b/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 # ---------------------------------------------------------------------------