From dbb00e054ee1c373526960959273d2e4841e5fa1 Mon Sep 17 00:00:00 2001 From: Hussain Sultan Date: Mon, 6 Jul 2026 19:31:47 -0400 Subject: [PATCH] fix: fan-out-safe t.all() totals and loud, ownership-based filter pushdown MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fixes three silent-wrong-answer defects in the join_many pre-aggregation path, found during a soundness evaluation of v0.3.15: 1. t.all(...) totals were built by re-running agg specs on the fanned-out joined table, inflating denominators (percent_of_total summed to 0.5 with 2 line items per order). The pre-agg path now captures each (filtered) source table plus its original measure expressions and passes a lazy fan-out-safe totals builder through _apply_calc_specs → apply_calc_measures(totals_base_builder=...): each table aggregates at zero grain — where fan-out cannot occur — and the one-row results are cross-joined. Correct for sum/count, mean, and count-distinct totals. 2. Filters were silently dropped or mis-pushed. Root cause: _Resolver .__getitem__ ignored the dimension map, so t["orders.status"] raised and the entire filtered-join build collapsed to tbl=None inside a broad except, discarding the filter. Bracket access on _Resolver / _AggResolver now falls back to declared dims/measures (materialized columns still win, preserving post-agg stale-dimension semantics), and pre-agg filter pushdown is ownership-based: each filter resolves against every source table's dims (bare and table-prefixed); it is pushed to a raw table only when exactly one table owns it, ambiguous filters apply through the filtered joined table's key bridge, and a filter handled by neither path raises ValueError instead of being silently ignored. The deferred-join path's filter application and _find_deferrable_joins' reference detection get the same treatment. 3. find_join_in_tree treated the wrapper SemanticTableOp created by SemanticJoin.with_measures()/with_dimensions() as a leaf, so a .filter() between the wrapper and .aggregate() silently skipped pre-aggregation and returned fanned-out sums. The wrapper's _source_join is now followed. Regression coverage in test_preagg_filters_and_totals.py: totals under uneven 1:N fan (sum and mean), totals with filters, prefixed / bare- ambiguous / single-owner filter pushdown, unresolvable-filter error, and prefixed filters on the join_one path. Co-Authored-By: Claude Fable 5 --- src/boring_semantic_layer/calc_compiler.py | 34 ++- src/boring_semantic_layer/convert.py | 25 +- src/boring_semantic_layer/ops.py | 247 +++++++++++++----- .../tests/test_preagg_filters_and_totals.py | 211 +++++++++++++++ 4 files changed, 447 insertions(+), 70 deletions(-) create mode 100644 src/boring_semantic_layer/tests/test_preagg_filters_and_totals.py diff --git a/src/boring_semantic_layer/calc_compiler.py b/src/boring_semantic_layer/calc_compiler.py index 76739f40..4ddbbb6c 100644 --- a/src/boring_semantic_layer/calc_compiler.py +++ b/src/boring_semantic_layer/calc_compiler.py @@ -474,6 +474,7 @@ def apply_calc_measures( real_totals_tbl=None, agg_specs: dict[str, Any] | None = None, totals_prefix: str = TOTALS_PREFIX, + totals_base_builder: Any | None = None, ): """Re-run each calc-measure lambda against the real aggregated table. @@ -553,7 +554,11 @@ def apply_calc_measures( if real_with_totals is None: if real_totals_tbl is None: real_totals_tbl = _build_totals_from_agg_specs( - base_tbl, agg_specs, calc_lambdas, known_measures + base_tbl, + agg_specs, + calc_lambdas, + known_measures, + base_totals_builder=totals_base_builder, ) if real_totals_tbl is not None: real_with_totals = _join_totals( @@ -847,6 +852,7 @@ def _build_totals_from_agg_specs( calc_lambdas: dict[str, Any], known_measures: frozenset[str], classifications: dict[str, CalcExprAnalysis] | None = None, + base_totals_builder: Any | None = None, ): """Build a no-group-by totals table when callers passed ``agg_specs``. @@ -855,15 +861,25 @@ def _build_totals_from_agg_specs( see correctly-recomputed dependencies. Returns ``None`` when there is no way to construct totals (no ``agg_specs`` supplied or the specs fail to evaluate against the base). + + ``base_totals_builder``, when given, replaces the agg-spec re-run as + the source of base totals. The pre-agg join path uses it because + re-running agg specs on a fanned-out join inflates the totals; the + builder aggregates each source table at zero grain instead. """ - if not agg_specs: - return None - try: - totals_aggs = {n: f(base_tbl) for n, f in agg_specs.items()} - except Exception as exc: - logger.debug("totals aggregation failed to evaluate: %s", exc) - return None - real_totals = base_tbl.aggregate(**totals_aggs) + if base_totals_builder is not None: + real_totals = base_totals_builder() + if real_totals is None: + return None + else: + if not agg_specs: + return None + try: + totals_aggs = {n: f(base_tbl) for n, f in agg_specs.items()} + except Exception as exc: + logger.debug("totals aggregation failed to evaluate: %s", exc) + return None + real_totals = base_tbl.aggregate(**totals_aggs) if classifications is None: classifications = classify_calc_lambdas(calc_lambdas, base_tbl, known_measures) diff --git a/src/boring_semantic_layer/convert.py b/src/boring_semantic_layer/convert.py index 1d4e2c34..d9383760 100644 --- a/src/boring_semantic_layer/convert.py +++ b/src/boring_semantic_layer/convert.py @@ -105,7 +105,17 @@ def __getattr__(self, name: str): return getattr(self._t, name) def __getitem__(self, name: str): - return getattr(self._t, name) + # Materialized columns win (e.g. prefixed columns already present + # on a post-aggregation result, where the dimension lambda would + # be stale). Fall back to declared dimensions so bracket access + # also resolves prefixed names like t["orders.status"] on tables + # that don't carry them as literal columns yet. + try: + return getattr(self._t, name) + except AttributeError: + if name in self._dims: + return self._dims[name](self._t).name(name) + raise @frozen @@ -128,7 +138,18 @@ def __getattr__(self, key: str): return getattr(self._t, key) def __getitem__(self, key: str): - return getattr(self._t, key) + # Materialized columns win (a stale dim/measure lambda must not + # shadow a column that already exists on the result); fall back + # to declared dims/measures for prefixed names like + # t["orders.total"] that aren't literal columns yet. + try: + return getattr(self._t, key) + except AttributeError: + if key in self._dims: + return self._dims[key](self._t) + if key in self._meas: + return self._meas[key](self._t) + raise # ============================================================================ diff --git a/src/boring_semantic_layer/ops.py b/src/boring_semantic_layer/ops.py index bb9fda16..8d333b9f 100644 --- a/src/boring_semantic_layer/ops.py +++ b/src/boring_semantic_layer/ops.py @@ -2171,6 +2171,23 @@ class _DeferrableJoin: deferred_dims: tuple # Prefixed dimension names to add post-agg +def _table_filter_resolver(raw_tbl, table_op, table_name): + """Build a filter resolver scoped to one source table. + + Exposes the table's declared dimensions under both their bare and + table-prefixed names (raw columns resolve via the fallback), so + ownership checks resolve ``t["orders.status"]`` only against + ``orders``. + """ + from .convert import _Resolver + + dims = dict(_get_field_dict(table_op, "dimensions")) + if table_name: + for dname, dim in list(dims.items()): + dims[f"{table_name}.{dname}"] = dim + return _Resolver(raw_tbl, dims) + + def _find_deferrable_joins( join_op, group_by_keys: tuple[str, ...], @@ -2195,12 +2212,14 @@ def _filter_references_table(table_name, table_op): if not filters: return False raw_tbl = _to_untagged(table_op) + resolver = _table_filter_resolver(raw_tbl, table_op, table_name) for pred in filters: pred_fn = _unwrap(pred) try: - # If the predicate can be resolved against this table, - # it references its columns → can't defer - pred_fn(raw_tbl) + # If the predicate resolves against this table's columns or + # dimensions (bare or table-prefixed), it references the + # table → can't defer + _resolve_expr(pred_fn, resolver) return True except Exception: pass @@ -2555,8 +2574,12 @@ def find_join_in_tree(node): if isinstance(node, SemanticJoinOp): return node if isinstance(node, SemanticTableOp): - # Leaf operations - no source to traverse - return None + # Wrapper tables from SemanticJoin.with_measures()/ + # with_dimensions() carry the join in _source_join; plain + # leaf tables carry None. Following it here keeps queries + # with filters between the wrapper and the aggregate on + # the fan-out-safe pre-aggregation path. + return node._source_join if node.source is not None: return find_join_in_tree(node.source) return None @@ -2748,6 +2771,7 @@ def _to_untagged_with_preagg( group_by_cols = list(self.keys) filters = filters or [] + filter_fns = [_unwrap(pred) for pred in filters] # --- 1. Try to build the full joined table (for scope / dim bridge) --- # Pre-agg needs all tables for dimension bridges — no pruning here. @@ -2758,19 +2782,70 @@ def _to_untagged_with_preagg( [k for k in self.keys if k in merged_dimensions], merged_dimensions, ) - # Apply collected filters to the full joined table so that - # dimension bridges only include rows surviving the filter. - if filters: - from .convert import _Resolver - - for pred in filters: - pred_fn = _unwrap(pred) - resolver = _Resolver(tbl, merged_dimensions) - pred_expr = _resolve_expr(pred_fn, resolver) - tbl = tbl.filter(pred_expr) except Exception: tbl = None # chasm / column collision – work without full join + # Apply collected filters to the full joined table so that + # dimension bridges only include rows surviving the filter. A + # filter that fails to resolve here may still be pushed to its + # owning source table below; anything handled by neither path + # raises instead of silently dropping the filter. + filters_on_tbl: set[int] = set() + if tbl is not None and filter_fns: + from .convert import _Resolver + + for i, pred_fn in enumerate(filter_fns): + try: + resolver = _Resolver(tbl, merged_dimensions) + filtered = tbl.filter(_resolve_expr(pred_fn, resolver)) + except Exception: + continue + tbl = filtered + filters_on_tbl.add(i) + + # --- 1b. Determine which source table(s) each filter belongs to --- + # Ownership resolution uses each table's own dimensions (bare and + # table-prefixed) so ``t["orders.status"]`` resolves only against + # ``orders``. A filter owned by exactly one table is pushed to that + # table's raw table; a filter resolving against several tables is + # ambiguous and is only applied through the filtered joined table. + raw_tables: dict = {} + filter_owners: list[frozenset] = [] + if filter_fns: + for tname, top in join_tree_info.table_ops.items(): + try: + raw_tables[tname] = _to_untagged(top) + except Exception: + continue + for pred_fn in filter_fns: + owners = set() + for tname, top in join_tree_info.table_ops.items(): + raw = raw_tables.get(tname) + if raw is None: + continue + try: + _resolve_expr(pred_fn, _table_filter_resolver(raw, top, tname)) + owners.add(tname) + except Exception: + pass + filter_owners.append(frozenset(owners)) + for i, owners in enumerate(filter_owners): + if i in filters_on_tbl or len(owners) == 1: + continue + if len(owners) > 1: + raise ValueError( + f"Filter #{i} is ambiguous: it resolves against multiple " + f"joined tables ({', '.join(sorted(owners))}) and could " + "not be applied to the full joined table. Qualify the " + 'field with a table prefix (e.g. t["orders.status"]).' + ) + raise ValueError( + f"Filter #{i} does not resolve against the joined table or " + "any single source table; it would be silently ignored. " + "Check the dimension/column name, or qualify it with a " + 'table prefix (e.g. t["orders.status"]).' + ) + # --- 2. Build aggregation plan --- if tbl is not None: scope = MeasureScope( @@ -2817,12 +2892,17 @@ def _to_untagged_with_preagg( # Track COUNT DISTINCT measures deferred past pre-aggregation. # Value: (table_name, short_name, raw_tbl, measure_fn) _deferred_count_distincts: dict[str, tuple] = {} + # Fan-out-safe totals sources for t.all(...): per table, the + # filtered raw table plus the original (undecomposed) measure + # expressions, aggregated at zero grain on first need. + _totals_sources: dict = {} for table_name, measures in partitioned.items(): if table_name is None: # Unprefixed – aggregate on the full join if available if tbl is not None: agg_exprs = {n: f(tbl) for n, f in measures.items()} + _totals_sources[None] = (tbl, dict(agg_exprs)) if group_by_cols: r = tbl.group_by([tbl[c] for c in group_by_cols]).aggregate(**agg_exprs) else: @@ -2836,21 +2916,25 @@ def _to_untagged_with_preagg( raw_tbl = _to_untagged(table_op) - # Push applicable filters to per-table raw tables - if filters: - all_pushed = True - for pred in filters: - pred_fn = _unwrap(pred) - try: - pred_expr = pred_fn(raw_tbl) + # Push filters owned by this table onto its raw table. Filters + # handled elsewhere (applied to the full joined table, or owned + # by another table) reach this table via a join-key bridge. + if filter_fns: + needs_bridge = False + for i, pred_fn in enumerate(filter_fns): + if filter_owners[i] == frozenset({table_name}): + pred_expr = _resolve_expr( + pred_fn, + _table_filter_resolver(raw_tbl, table_op, table_name), + ) raw_tbl = raw_tbl.filter(pred_expr) - except Exception: - all_pushed = False # filter references columns not on this table + else: + needs_bridge = True - # If some filters couldn't be pushed (cross-table filters), - # restrict via join keys from the filtered full joined table - # or from filtered raw tables that own the filter columns. - if not all_pushed: + # Filters not pushed here (cross-table, ambiguous, or owned + # by another table) restrict via join keys from the filtered + # full joined table, or from the owning table's raw table. + if needs_bridge: jk = join_tree_info.table_join_keys.get(table_name, set()) if tbl is not None: shared = sorted(jk & set(raw_tbl.columns) & set(tbl.columns)) @@ -2859,32 +2943,34 @@ def _to_untagged_with_preagg( preds = [raw_tbl[c] == key_bridge[c] for c in shared] raw_tbl = raw_tbl.inner_join(key_bridge, preds).select(raw_tbl) else: - # Chasm fallback: build key bridge from other raw tables - for other_name, other_op in join_tree_info.table_ops.items(): - if other_name == table_name: + # Chasm fallback: restrict via each owning table's keys + for i, pred_fn in enumerate(filter_fns): + owners = filter_owners[i] + if table_name in owners or len(owners) != 1: + continue + (owner_name,) = owners + owner_op = join_tree_info.table_ops.get(owner_name) + owner_raw = raw_tables.get(owner_name) + if owner_op is None or owner_raw is None: continue - other_raw = _to_untagged(other_op) - can_filter = False - for pred in filters: - pred_fn = _unwrap(pred) - try: - pred_expr = pred_fn(other_raw) - other_raw = other_raw.filter(pred_expr) - can_filter = True - except Exception: - pass - if can_filter: - other_jk = join_tree_info.table_join_keys.get(other_name, set()) - shared = sorted( - jk & other_jk & set(raw_tbl.columns) & set(other_raw.columns) + owner_raw = owner_raw.filter( + _resolve_expr( + pred_fn, + _table_filter_resolver( + owner_raw, owner_op, owner_name + ), ) - if shared: - key_bridge = other_raw.select( - [other_raw[c] for c in shared] - ).distinct() - preds = [raw_tbl[c] == key_bridge[c] for c in shared] - raw_tbl = raw_tbl.inner_join(key_bridge, preds).select(raw_tbl) - break + ) + owner_jk = join_tree_info.table_join_keys.get(owner_name, set()) + shared = sorted( + jk & owner_jk & set(raw_tbl.columns) & set(owner_raw.columns) + ) + if shared: + key_bridge = owner_raw.select( + [owner_raw[c] for c in shared] + ).distinct() + preds = [raw_tbl[c] == key_bridge[c] for c in shared] + raw_tbl = raw_tbl.inner_join(key_bridge, preds).select(raw_tbl) table_measures = _get_field_dict(table_op, "measures") table_dims = _get_field_dict(table_op, "dimensions") @@ -2892,10 +2978,14 @@ def _to_untagged_with_preagg( # Build agg expressions on the raw table agg_exprs: dict = {} + _tot_exprs: dict = {} for mname, _mfn in measures.items(): short = mname.split(".", 1)[1] if "." in mname else mname if short in table_measures: expr = table_measures[short](raw_tbl) + # Original expression on the filtered raw table: at zero + # grain this is fan-out-safe, so it powers t.all(...) totals + _tot_exprs[mname] = expr # Decompose MEAN into SUM + COUNT for correct re-aggregation if _is_mean_expr(expr): mean_op = expr.op() @@ -2914,6 +3004,9 @@ def _to_untagged_with_preagg( _reagg_ops[mname] = _reagg_op_for_expr(expr) agg_exprs[mname] = expr + if _tot_exprs: + _totals_sources[table_name] = (raw_tbl, _tot_exprs) + if not agg_exprs: continue @@ -3100,7 +3193,29 @@ def _to_untagged_with_preagg( # --- 6. Apply calc_specs --- if plan.calc_specs: - result = self._apply_calc_specs(result, plan, tbl) + + def _fanout_safe_totals(): + """Zero-grain totals from per-table raw aggregates. + + Aggregating each (filtered) source table without grouping + cannot fan out, so ``t.all(...)`` denominators stay correct + under join_many. + """ + parts = [ + src.aggregate(**texprs) + for src, texprs in _totals_sources.values() + if texprs + ] + if not parts: + return None + total = parts[0] + for p in parts[1:]: + total = total.cross_join(p) + return total + + result = self._apply_calc_specs( + result, plan, tbl, totals_builder=_fanout_safe_totals + ) # --- 7. Select requested columns --- available = frozenset(result.columns) @@ -3191,18 +3306,26 @@ def strip_deferred(node): merged_dimensions, ) - # Apply filters + # Apply filters — loudly. Filters referencing deferred tables were + # already excluded from deferral by _find_deferrable_joins, so a + # filter that fails to resolve here is a genuine error, not a + # cross-table predicate awaiting another mechanism. if filters: from .convert import _Resolver - for pred in filters: + for i, pred in enumerate(filters): pred_fn = _unwrap(pred) try: resolver = _Resolver(core_tbl, merged_dimensions) pred_expr = _resolve_expr(pred_fn, resolver) - core_tbl = core_tbl.filter(pred_expr) - except Exception: - pass + except Exception as exc: + raise ValueError( + f"Filter #{i} does not resolve against the aggregation " + "source table; it would be silently ignored. Check the " + "dimension/column name, or qualify it with a table " + 'prefix (e.g. t["orders.status"]).' + ) from exc + core_tbl = core_tbl.filter(pred_expr) # Build aggregation expressions scope = MeasureScope( @@ -3465,7 +3588,7 @@ def _bridge_one_preagg(pt): return result @staticmethod - def _apply_calc_specs(result, plan, tbl): + def _apply_calc_specs(result, plan, tbl, totals_builder=None): """Apply calculated measure specs to the pre-aggregated result. Each calc spec is a :class:`CalcMeasure` whose lambda is @@ -3475,6 +3598,11 @@ def _apply_calc_specs(result, plan, tbl): result so non-sum measures (mean/quantile/…) get correct overall values; ``apply_calc_measures`` builds the totals lazily on first use. + + ``totals_builder`` overrides how the base totals table is built. + The pre-agg path passes a fan-out-safe builder that aggregates + each source table at zero grain — re-running agg specs on the + fanned-out join would inflate ``t.all(...)`` denominators. """ if not plan.calc_specs: return result @@ -3486,6 +3614,7 @@ def _apply_calc_specs(result, plan, tbl): dict(plan.calc_specs), known, agg_specs=dict(plan.agg_specs), + totals_base_builder=totals_builder, ) diff --git a/src/boring_semantic_layer/tests/test_preagg_filters_and_totals.py b/src/boring_semantic_layer/tests/test_preagg_filters_and_totals.py new file mode 100644 index 00000000..d123bcfe --- /dev/null +++ b/src/boring_semantic_layer/tests/test_preagg_filters_and_totals.py @@ -0,0 +1,211 @@ +"""Regression tests: filter ownership and fan-out-safe totals in the pre-agg path. + +Covers two silent-wrong-answer defects found in the July 2026 soundness +evaluation: + +1. ``t.all(...)`` totals were computed by re-running agg specs on the + fanned-out join under ``join_many``, inflating denominators (e.g. + percent-of-total summed to 0.5 with 2 line items per order). +2. Filters written with table-prefixed names (``t["orders.status"]``) + resolved against no raw table and were silently dropped; bare names + shared by several tables (``t.status``) were pushed to every table, + zeroing out unrelated measures. +""" + +import ibis +import pandas as pd +import pytest + +from boring_semantic_layer import to_semantic_table + + +@pytest.fixture +def con(): + return ibis.duckdb.connect(":memory:") + + +@pytest.fixture +def orders_items(con): + """Orders (sum 300) with an uneven 1:N fan into line items (3/1/2).""" + orders = con.create_table( + "orders", + pd.DataFrame( + { + "order_id": [1, 2, 3], + "customer_id": [10, 10, 20], + "status": ["open", "closed", "open"], # order status + "amount": [100, 120, 80], + } + ), + ) + items = con.create_table( + "items", + pd.DataFrame( + { + "item_id": [1, 2, 3, 4, 5, 6], + "order_id": [1, 1, 1, 2, 3, 3], + "status": ["ok"] * 6, # item QC status — same name, other meaning + "qty": [1, 2, 1, 3, 1, 1], + } + ), + ) + o_st = ( + to_semantic_table(orders, name="orders") + .with_dimensions( + customer_id=lambda t: t.customer_id, + status=lambda t: t.status, + ) + .with_measures( + total_amount=lambda t: t.amount.sum(), + avg_amount=lambda t: t.amount.mean(), + ) + ) + i_st = to_semantic_table(items, name="items").with_measures( + item_count=lambda t: t.count(), + ) + return o_st, i_st + + +def _joined(orders_items): + o_st, i_st = orders_items + return o_st.join_many(i_st, lambda o, i: o.order_id == i.order_id) + + +class TestFanoutSafeTotals: + def test_percent_of_total_sums_to_one(self, orders_items): + joined = _joined(orders_items).with_measures( + pot=lambda t: t["orders.total_amount"] / t.all(t["orders.total_amount"]), + ) + df = ( + joined.group_by("orders.customer_id") + .aggregate("orders.total_amount", "pot") + .order_by("orders.customer_id") + .execute() + ) + assert df["orders.total_amount"].tolist() == [220, 80] + # Denominator must be 300 (orders grain), not 580 (fanned-out rows) + assert df["pot"].tolist() == pytest.approx([220 / 300, 80 / 300]) + assert df["pot"].sum() == pytest.approx(1.0) + + def test_all_of_mean_uses_source_grain(self, orders_items): + joined = _joined(orders_items).with_measures( + mean_ratio=lambda t: t["orders.avg_amount"] / t.all(t["orders.avg_amount"]), + ) + df = ( + joined.group_by("orders.customer_id") + .aggregate("orders.avg_amount", "mean_ratio") + .order_by("orders.customer_id") + .execute() + ) + assert df["orders.avg_amount"].tolist() == pytest.approx([110.0, 80.0]) + # Grand mean over orders is 100; the fanned-out join would give 580/6 + assert df["mean_ratio"].tolist() == pytest.approx([110 / 100, 80 / 100]) + + def test_totals_respect_filters(self, orders_items): + joined = _joined(orders_items).with_measures( + pot=lambda t: t["orders.total_amount"] / t.all(t["orders.total_amount"]), + ) + df = ( + joined.filter(lambda t: t["orders.status"] == "open") + .group_by("orders.customer_id") + .aggregate("orders.total_amount", "pot") + .order_by("orders.customer_id") + .execute() + ) + # Open orders: #1 (cust 10, 100) and #3 (cust 20, 80) → total 180 + assert df["orders.total_amount"].tolist() == [100, 80] + assert df["pot"].tolist() == pytest.approx([100 / 180, 80 / 180]) + assert df["pot"].sum() == pytest.approx(1.0) + + +class TestPreaggFilterOwnership: + def test_prefixed_filter_is_applied(self, orders_items): + df = ( + _joined(orders_items) + .filter(lambda t: t["orders.status"] == "open") + .aggregate("items.item_count", "orders.total_amount") + .execute() + ) + assert df["orders.total_amount"].iloc[0] == 180 + # Items of orders 1 and 3 → 3 + 2 + assert df["items.item_count"].iloc[0] == 5 + + def test_prefixed_filter_with_group_by(self, orders_items): + df = ( + _joined(orders_items) + .filter(lambda t: t["orders.status"] == "open") + .group_by("orders.customer_id") + .aggregate("items.item_count", "orders.total_amount") + .order_by("orders.customer_id") + .execute() + ) + assert df["orders.total_amount"].tolist() == [100, 80] + assert df["items.item_count"].tolist() == [3, 2] + + def test_bare_ambiguous_filter_follows_join_semantics(self, orders_items): + # `status` exists on both tables. Bare access must follow the joined + # table's semantics (left table wins → orders.status), not get pushed + # into every table that happens to have the column. + df = ( + _joined(orders_items) + .filter(lambda t: t.status == "open") + .aggregate("items.item_count", "orders.total_amount") + .execute() + ) + assert df["orders.total_amount"].iloc[0] == 180 + assert df["items.item_count"].iloc[0] == 5 + + def test_unowned_filter_pushes_to_owner_only(self, orders_items): + # `qty` exists only on items — must not disturb orders' measures + df = ( + _joined(orders_items) + .filter(lambda t: t.qty >= 2) + .aggregate("items.item_count") + .execute() + ) + assert df["items.item_count"].iloc[0] == 2 + + def test_unresolvable_filter_raises(self, orders_items): + expr = ( + _joined(orders_items) + .filter(lambda t: t["orders.no_such_field"] == 1) + .aggregate("items.item_count") + ) + with pytest.raises(ValueError, match="does not resolve"): + expr.execute() + + +class TestPrefixedFilterJoinOnePath: + def test_prefixed_filter_on_join_one_dimension_table(self, con): + orders = con.create_table( + "orders2", + pd.DataFrame( + { + "order_id": [1, 2, 3], + "customer_id": [10, 10, 20], + "amount": [100, 120, 80], + } + ), + ) + customers = con.create_table( + "customers2", + pd.DataFrame({"customer_id": [10, 20], "region": ["west", "east"]}), + ) + o_st = ( + to_semantic_table(orders, name="orders") + .with_dimensions(customer_id=lambda t: t.customer_id) + .with_measures(total_amount=lambda t: t.amount.sum()) + ) + c_st = to_semantic_table(customers, name="customers").with_dimensions( + customer_id=lambda t: t.customer_id, + region=lambda t: t.region, + ) + df = ( + o_st.join_one(c_st, lambda o, c: o.customer_id == c.customer_id) + .filter(lambda t: t["customers.region"] == "west") + .group_by("customers.region") + .aggregate("orders.total_amount") + .execute() + ) + assert len(df) == 1 + assert df["orders.total_amount"].iloc[0] == 220