From cdcb007bb37cb9fe823e35005a5c957380cf0339 Mon Sep 17 00:00:00 2001 From: Hussain Sultan Date: Sun, 23 Aug 2026 12:43:31 -0400 Subject: [PATCH 1/5] fix: recover join leaves poisoned by internal __bsl_jk_ temporaries MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When a join leaf is an into_backend seam (RemoteTable placeholder + payload relation), round-trip leaf recovery cannot walk to a single base relation and keeps the lowered leaf projection as the model's table. If join-key names collide across the joined tables, SemanticJoinOp.to_untagged has renamed the left predicate columns to __bsl_jk_ temporaries inside that projection (the _RenamedResolver ibis workaround), so the declared dimensions no longer resolve and from_tagged raises "Round-trip could not recover the left join table ...". Observed in the field: a 6-table banking model over Snowflake-backed into_backend sources with customer_id/account_id shared across legs failed every read-back with this error, while the same model built with globally unique physical column names round-tripped fine. Fix: _strip_internal_join_temps inverts the reserved temporaries on the recovered leaf table, restoring the schema the model was authored against. Only exact-prefix temporaries whose original name is free are inverted; the __bsl_jk__N overflow spelling (a user column literally named __bsl_jk_ existed) is left alone to avoid corrupting that column. Regression tests build the collision model over memtable into_backend seams: round-trip recovery, plan lowering against original names, one in-budget execution, and the user look-alike column guard. (Cross-leg EXECUTION through in-process seams is not asserted: such seams only support a bounded number of reads per plan — an xorq-side limitation, unchanged by this fix.) Co-Authored-By: Claude Fable 5 --- .../serialization/reconstruct.py | 41 +++++- .../tests/test_xorq_join_leaf_recovery.py | 126 ++++++++++++++++++ 2 files changed, 166 insertions(+), 1 deletion(-) create mode 100644 src/boring_semantic_layer/tests/test_xorq_join_leaf_recovery.py diff --git a/src/boring_semantic_layer/serialization/reconstruct.py b/src/boring_semantic_layer/serialization/reconstruct.py index c5b0baec..820b3d26 100644 --- a/src/boring_semantic_layer/serialization/reconstruct.py +++ b/src/boring_semantic_layer/serialization/reconstruct.py @@ -168,7 +168,7 @@ def _reconstruct_table(): ) return bsl_expr.SemanticModel( - table=_reconstruct_table(), + table=_strip_internal_join_temps(_reconstruct_table()), dimensions=dimensions, measures=measures, calc_measures=calc_measures, @@ -326,6 +326,45 @@ def _reconstruct_limit(metadata: dict, xorq_expr, source, context: BSLSerializat return source.limit(n=int(metadata.get("n", 0)), offset=int(metadata.get("offset", 0))) +def _strip_internal_join_temps(expr): + """Invert BSL's temporary join-key renames on a recovered leaf table. + + ``SemanticJoinOp.to_untagged`` renames left-side predicate columns that + collide across the join to ``__bsl_jk_`` (see ``_RenamedResolver``) + to sidestep ibis ambiguous-deref errors. Those temporaries live in the + lowered leaf projections. When leaf recovery cannot walk to the base + relation and keeps a lowered projection as the model's table — e.g. an + ``into_backend`` seam makes the leaf multi-relation — the declared + dimensions/measures reference the ORIGINAL names and no longer resolve. + Rename the reserved temporaries back so the recovered leaf carries the + schema the model was authored against. + + Only exact-prefix temporaries whose original name is free are inverted; + the ``__bsl_jk__N`` overflow spelling (a user column literally + named ``__bsl_jk_`` existed) is left alone — inverting it could + corrupt that user column. + """ + from ..ops._normalize import _BSL_JOIN_KEY_TMP_PREFIX + + try: + columns = list(expr.columns) + except Exception: + return expr + renames = {} + for col in columns: + if not col.startswith(_BSL_JOIN_KEY_TMP_PREFIX): + continue + original = col[len(_BSL_JOIN_KEY_TMP_PREFIX) :] + if original and original not in columns and original not in renames: + renames[original] = col + if not renames: + return expr + try: + return expr.rename(**renames) + except Exception: + return expr + + def _validate_join_leaf(model, metadata, side: str) -> None: """Check a reconstructed join leaf against its declared fields. diff --git a/src/boring_semantic_layer/tests/test_xorq_join_leaf_recovery.py b/src/boring_semantic_layer/tests/test_xorq_join_leaf_recovery.py new file mode 100644 index 00000000..35b39430 --- /dev/null +++ b/src/boring_semantic_layer/tests/test_xorq_join_leaf_recovery.py @@ -0,0 +1,126 @@ +"""Round-trip recovery of joined models whose leaves are into_backend seams. + +Regression for the pennybank failure: when a join's leaf table is an +``into_backend`` seam, leaf recovery cannot walk to a single base relation +and keeps the lowered leaf projection as the model's table. With colliding +join-key names across the joined tables, ``SemanticJoinOp.to_untagged`` +renames left predicate columns to ``__bsl_jk_`` temporaries inside +that projection, so the declared dimensions no longer resolved and +``from_tagged`` raised "Round-trip could not recover the left join table". + +The fix inverts BSL's reserved temporaries on the recovered leaf +(``_strip_internal_join_temps``), restoring the schema the model was +authored against. +""" + +from __future__ import annotations + +import pytest + +from boring_semantic_layer import to_semantic_table +from boring_semantic_layer.serialization import from_tagged, to_tagged + +xorq = pytest.importorskip("xorq", reason="xorq not installed") + +import pandas as pd # noqa: E402 +import xorq.api as xo # noqa: E402 + + +@pytest.fixture +def seamed_tables(): + """Three collision-heavy tables, each an into_backend seam. + + The seam matters: a leaf whose expression holds more than one relation + (RemoteTable placeholder + in-memory payload) cannot be recovered by + walking to a single base table, so recovery keeps the lowered leaf + projection — the one carrying the ``__bsl_jk_`` temporaries. + """ + frames = { + "accounts": pd.DataFrame( + { + "account_id": [1, 2, 3], + "customer_id": [10, 10, 20], + "credit_limit": [1000.0, 2000.0, 500.0], + } + ), + "customers": pd.DataFrame({"customer_id": [10, 20], "state": ["CA", "NY"]}), + "transactions": pd.DataFrame( + { + "transaction_id": [1, 2, 3, 4], + "account_id": [1, 1, 2, 3], + "amount": [5.0, 6.0, 7.0, 8.0], + } + ), + } + con = xo.connect() + return { + name: xo.memtable(df, name=name).into_backend(con, f"{name}_rt") + for name, df in frames.items() + } + + +def _build_model(tables): + accounts = ( + to_semantic_table(tables["accounts"], name="accounts") + .with_dimensions( + account_id=lambda t: t.account_id, + customer_id=lambda t: t.customer_id, + ) + .with_measures(account_count=lambda t: t.count()) + ) + customers = to_semantic_table(tables["customers"], name="customers").with_dimensions( + customer_id=lambda t: t.customer_id, + state=lambda t: t.state, + ) + transactions = to_semantic_table(tables["transactions"], name="transactions").with_measures( + total_amount=lambda t: t.amount.sum(), + ) + return accounts.join_one(customers, on=lambda a, c: a.customer_id == c.customer_id).join_many( + transactions, on=lambda a, t: a.account_id == t.account_id + ) + + +def test_seamed_collision_join_round_trips(seamed_tables): + tagged = to_tagged(_build_model(seamed_tables)) + + recovered = from_tagged(tagged) + + dims = recovered.get_dimensions() + assert "accounts.customer_id" in dims + assert "customers.customer_id" in dims + meas = recovered.get_measures() + assert "accounts.account_count" in meas + assert "transactions.total_amount" in meas + + +def test_seamed_collision_join_lowers_and_executes(seamed_tables): + """The recovered model must lower against ORIGINAL column names. + + Execution through in-process ``into_backend`` seams only supports a + bounded number of reads per seam, so this asserts one single-leg query + (which fits the budget) and plan-lowering for a cross-leg query. + """ + tagged = to_tagged(_build_model(seamed_tables)) + recovered = from_tagged(tagged) + + cross_leg = recovered.group_by("customers.state").aggregate( + "accounts.account_count", "transactions.total_amount" + ) + assert cross_leg.to_untagged() is not None + + result = recovered.aggregate("accounts.account_count").to_untagged().execute() + assert list(result["accounts.account_count"]) == [3] + + +def test_recovered_leaf_keeps_user_temp_lookalike_column(): + """A user column literally named __bsl_jk_x must not be renamed away.""" + con = xo.connect() + df = pd.DataFrame({"__bsl_jk_x": [1, 2], "x": [3, 4], "y": [5.0, 6.0]}) + weird = xo.memtable(df, name="weird").into_backend(con, "weird_rt") + + model = to_semantic_table(weird, name="weird").with_dimensions( + jk=lambda t: t["__bsl_jk_x"], + x=lambda t: t.x, + ) + recovered = from_tagged(to_tagged(model)) + assert set(recovered.get_dimensions()) == {"jk", "x"} From 155dcaf15346da1cad4e556f8dbf51bf962a39e1 Mon Sep 17 00:00:00 2001 From: Hussain Sultan Date: Sun, 23 Aug 2026 16:35:33 -0400 Subject: [PATCH 2/5] fix: leaf recovery preserves authored deferred shaping (star-schema views) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Leaf recovery re-derived each model table by walking the lowered expression to its bare base relation, discarding everything the author put between the base and the model: mutates, selects, renames. A model built the dimensional-modeling way — shape conformed dimension / fact views as deferred xorq expressions first (validated at authoring), then a thin semantic layer of names and simple reductions on top — failed round-trip at the first derived column ("'Table' object has no attribute 'is_open'"), forcing all derivation logic into measure/dimension lambdas, the one layer with no authoring-time validation. _reconstruct_table now returns the leaf chain AS-IS when it is a pure per-row chain over exactly one relation (no Aggregate, no JoinChain): the chain IS the model's table. Lowered query entries still fall through to the base walk (digging under the aggregate is what recovery is for there), memtable leaves keep their from_ibis conversion, and preserved chains still get __bsl_jk_ temporaries inverted at the call site. Also: _validate_join_leaf now quotes the underlying exception instead of unconditionally blaming pre-aggregation lowering — a measure written with an API this ibis runtime lacks (the Column.filter misdiagnosis: "AttributeError: 'StringColumn' object has no attribute 'filter'" surfaced as a round-trip failure) now names the real error and the fix direction (.sum(where=...) forms). Known gap, marked xfail: a lowered QUERY entry over a shaped view still base-walks — BSL's query-time injected mutates and authored shaping are indistinguishable in the lowered tree; needs a lowering-time boundary marker. MODEL entries — what the doctrine catalogs — are covered. Suite: 1749 passed, 12 xfailed (the malloy-reader collection error and two import-graph failures in the primary checkout come from an in-progress malloy merge there, unrelated to this change). Co-Authored-By: Claude Fable 5 --- .../serialization/reconstruct.py | 45 ++++++- .../tests/test_xorq_join_leaf_recovery.py | 115 ++++++++++++++++++ 2 files changed, 156 insertions(+), 4 deletions(-) diff --git a/src/boring_semantic_layer/serialization/reconstruct.py b/src/boring_semantic_layer/serialization/reconstruct.py index 820b3d26..f026b4c8 100644 --- a/src/boring_semantic_layer/serialization/reconstruct.py +++ b/src/boring_semantic_layer/serialization/reconstruct.py @@ -122,6 +122,39 @@ def _reconstruct_table(): ) return from_ibis(expr) if not hasattr(expr.op(), "source") else expr + # Preserve authored deferred shaping: when the leaf is a pure per-row + # chain (mutate/select/filter — no aggregation, no join) over exactly + # one relation, the chain IS the model's table. Walking to the bare + # base relation here discarded the shaping, so a model built on a + # deferred star-schema view (e.g. columns like `is_open` derived via + # .mutate) recovered against the RAW source and every field + # referencing a derived column failed to resolve. Query entries + # (lowered aggregations) still fall through to the base walk below — + # digging under the aggregate is what recovery is FOR there. Reserved + # __bsl_jk_ join-key temporaries a preserved chain may carry are + # inverted by _strip_internal_join_temps at the call site. + from .._xorq import JoinChain + + leaf_op = unwrapped_expr.op() + is_bare_leaf = isinstance( + leaf_op, + ( + Read, + xorq_rel.InMemoryTable, + xorq_rel.DatabaseTable, + xorq_rel.UnboundTable, + xorq_rel.SelfReference, + ), + ) + if ( + total_leaf_tables == 1 + and not is_bare_leaf + # memtable leaves keep the from_ibis conversion below + and not in_memory_tables + and not walk_nodes((xorq_rel.Aggregate, JoinChain), unwrapped_expr) + ): + return unwrapped_expr + if read_ops: base = read_ops[0].to_expr() return base.view() if is_self_ref else base @@ -390,10 +423,14 @@ def _validate_join_leaf(model, metadata, side: str) -> None: raise ValueError( f"Round-trip could not recover the {side} join table " f"{name!r}: its {kind} {fname!r} does not resolve against " - "the recovered table. Queries lowered through the " - "pre-aggregation path cannot be reconstructed from the " - "lowered expression — serialize the model (or the " - "un-aggregated join) instead." + f"the recovered table ({type(exc).__name__}: {exc}). " + "If the underlying error names a missing METHOD, the " + "field's expression uses an API this ibis runtime does " + "not have (e.g. Column.filter — use .sum(where=...) " + "forms instead). If it names a missing COLUMN, the " + "expression was lowered through the pre-aggregation " + "path and cannot be reconstructed — serialize the model " + "(or the un-aggregated join) instead." ) from exc except Exception: continue diff --git a/src/boring_semantic_layer/tests/test_xorq_join_leaf_recovery.py b/src/boring_semantic_layer/tests/test_xorq_join_leaf_recovery.py index 35b39430..d7803f1d 100644 --- a/src/boring_semantic_layer/tests/test_xorq_join_leaf_recovery.py +++ b/src/boring_semantic_layer/tests/test_xorq_join_leaf_recovery.py @@ -124,3 +124,118 @@ def test_recovered_leaf_keeps_user_temp_lookalike_column(): ) recovered = from_tagged(to_tagged(model)) assert set(recovered.get_dimensions()) == {"jk", "x"} + + +def _seam_free_tables(): + """Shaped deferred views over ONE backend (no seams): the star-schema- + first doctrine — derivations authored in bare xorq, thin BSL on top.""" + con = xo.connect() + raw_accounts = con.register( + pd.DataFrame( + { + "account_id": [1, 2, 3], + "customer_id": [10, 10, 20], + "close_date": [None, "2025-01-01", None], + } + ), + "raw_accounts", + ) + raw_txn = con.register( + pd.DataFrame( + { + "transaction_id": [1, 2, 3, 4], + "account_id": [1, 1, 2, 3], + "amount": [-5.0, -6.0, 3.0, -8.0], + "transaction_type": ["Purchase", "Purchase", "Purchase Return", "ACH Payment"], + } + ), + "raw_transactions", + ) + accounts_view = raw_accounts.mutate(is_open=xo._.close_date.isnull()) + txn_view = raw_txn.mutate( + purchase_amount=(xo._.transaction_type == "Purchase").ifelse(-xo._.amount, 0.0), + ) + return accounts_view, txn_view + + +def test_shaped_single_table_round_trips(): + """A model over a deferred shaped view must recover WITH its shaping. + + Regression: _reconstruct_table walked to the bare base relation, + discarding authored mutates — `t.is_open.sum()` then failed with + "'Table' object has no attribute 'is_open'" wrapped in the round-trip + error, breaking the star-schema-view doctrine at the first step. + """ + accounts_view, _ = _seam_free_tables() + model = to_semantic_table(accounts_view, name="accounts").with_measures( + open_account_count=lambda t: t.is_open.sum(), + ) + recovered = from_tagged(to_tagged(model)) + result = recovered.aggregate("open_account_count").to_untagged().execute() + assert list(result["open_account_count"]) == [2] + + +def test_shaped_join_leaves_round_trip(): + accounts_view, txn_view = _seam_free_tables() + accounts = to_semantic_table(accounts_view, name="accounts").with_dimensions( + account_id=lambda t: t.account_id, + is_open=lambda t: t.is_open, + ) + transactions = to_semantic_table(txn_view, name="transactions").with_measures( + gross_purchases=lambda t: t.purchase_amount.sum(), + ) + model = accounts.join_many(transactions, on=lambda a, t: a.account_id == t.account_id) + + recovered = from_tagged(to_tagged(model)) + result = ( + recovered.group_by("accounts.is_open") + .aggregate("transactions.gross_purchases") + .to_untagged() + .execute() + .sort_values("accounts.is_open") + .reset_index(drop=True) + ) + # closed account 2: purchase-return row only → 0.0; open accounts 1+3: 5+6+0=11.0 + assert list(result["transactions.gross_purchases"]) == [0.0, 11.0] + + +@pytest.mark.xfail( + strict=True, + reason="Known gap: a lowered QUERY entry over a shaped view still base-walks " + "— BSL's query-time injected mutates and authored shaping are " + "indistinguishable in the lowered tree. Needs a lowering-time base-boundary " + "marker. MODEL entries (what the star-schema doctrine catalogs) are covered " + "by the tests above.", +) +def test_query_entry_still_digs_under_the_aggregate(): + """Recovery of a tagged aggregate over a SHAPED view should replay the + query against the shaped base — today the base walk discards the shaping.""" + accounts_view, _ = _seam_free_tables() + model = ( + to_semantic_table(accounts_view, name="accounts") + .with_dimensions( + customer_id=lambda t: t.customer_id, + ) + .with_measures(open_account_count=lambda t: t.is_open.sum()) + ) + tagged_query = to_tagged(model.group_by("customer_id").aggregate("open_account_count")) + recovered = from_tagged(tagged_query) + # reconstructed chain replays the query over the recovered base + result = recovered.to_untagged().execute().sort_values("customer_id").reset_index(drop=True) + assert list(result["open_account_count"]) == [1, 1] + + +def test_guard_message_names_the_underlying_error(): + """A field using a nonexistent API must surface the REAL exception, not + only the canned pre-aggregation explanation (the pennybank Column.filter + misdiagnosis).""" + accounts_view, txn_view = _seam_free_tables() + accounts = to_semantic_table(accounts_view, name="accounts").with_measures( + bad=lambda t: t.account_id.filter(t.is_open).nunique(), + ) + transactions = to_semantic_table(txn_view, name="transactions").with_measures( + gross_purchases=lambda t: t.purchase_amount.sum(), + ) + model = accounts.join_many(transactions, on=lambda a, t: a.account_id == t.account_id) + with pytest.raises(ValueError, match="AttributeError.*filter"): + from_tagged(to_tagged(model)) From 4a4e26f8769d958d32944b60e3f92a807a96bb69 Mon Sep 17 00:00:00 2001 From: Hussain Sultan Date: Sun, 23 Aug 2026 17:26:43 -0400 Subject: [PATCH 3/5] fix: stop one bad dimension/measure from silently wiping its siblings MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit serialize_dimensions/serialize_measures/serialize_calc_measures each wrap their whole per-field loop in one @safe, so a single unserializable entry (an untrusted callable, an unencodable constant) turned the entire collection into a Failure. _extract_semantic_table defaulted that away with .value_or({}), so to_tagged() returned successfully with every dimension or measure on that table gone and no error — surfacing later, if at all, as a confusing "unknown dimension" error on whatever query happened to reference a dropped field. Fix: raise the per-entry error the loop already constructs (naming the offending field) instead of swallowing it. Also adds regression coverage for tagging an already-aggregated join query (to_tagged(aggregate_cache_storage=...) explicitly supports this): a plain join_one + aggregate round-trips correctly, while a join_many fan-out leg under an aggregated query still cannot be recovered — pre-agg compilation rewrites the join into a decomposed tree with no JoinChain matching the original leaves, so leaf recovery can't isolate each side. _validate_join_leaf already catches this and raises loudly rather than returning wrong numbers; the new test pins that the failure stays loud rather than regressing into a silent-wrong one. Fully supporting that case needs tagging join legs at multiple points in the tree, which is out of scope here. Co-Authored-By: Claude Sonnet 5 --- .../serialization/extract.py | 26 ++++- .../test_serialization_trust_boundary.py | 44 +++++++++ .../tests/test_xorq_join_leaf_recovery.py | 94 +++++++++++++++++++ 3 files changed, 161 insertions(+), 3 deletions(-) diff --git a/src/boring_semantic_layer/serialization/extract.py b/src/boring_semantic_layer/serialization/extract.py index df14c1dc..a62bd8d8 100644 --- a/src/boring_semantic_layer/serialization/extract.py +++ b/src/boring_semantic_layer/serialization/extract.py @@ -85,16 +85,36 @@ def _ensure_registered(): # --------------------------------------------------------------------------- +def _unwrap_or_raise(result: Result[dict, Exception]) -> dict: + """Return a successful serialization result, or re-raise its failure. + + ``serialize_dimensions``/``serialize_measures``/``serialize_calc_measures`` + each wrap a whole dict-comprehension-style loop in ``@safe``, so one + unserializable entry (an untrusted callable, an unencodable constant) + turns the *entire* collection into a ``Failure``. Defaulting that away + with ``.value_or({})`` — the previous behavior — silently dropped every + sibling dimension/measure too: ``to_tagged()`` returned successfully + with an empty field set and no indication anything was wrong. Raising + here surfaces the per-entry error message (naming the offending field) + that the loop already constructs, instead of swallowing it. + """ + match result: + case Success(): + return result.unwrap() + case _: + raise result.failure() + + @_register_lazy("SemanticTableOp") def _extract_semantic_table(op, context: BSLSerializationContext) -> dict[str, Any]: dims_result = serialize_dimensions(op.get_dimensions()) meas_result = serialize_measures(op.get_measures()) calc_result = serialize_calc_measures(op.get_calculated_measures()) metadata: dict[str, Any] = { - "dimensions": dims_result.value_or({}), - "measures": meas_result.value_or({}), + "dimensions": _unwrap_or_raise(dims_result), + "measures": _unwrap_or_raise(meas_result), } - calc_data = calc_result.value_or({}) + calc_data = _unwrap_or_raise(calc_result) if calc_data: metadata["calc_measures"] = calc_data if op.name: diff --git a/src/boring_semantic_layer/tests/test_serialization_trust_boundary.py b/src/boring_semantic_layer/tests/test_serialization_trust_boundary.py index 0c36ca03..fca740eb 100644 --- a/src/boring_semantic_layer/tests/test_serialization_trust_boundary.py +++ b/src/boring_semantic_layer/tests/test_serialization_trust_boundary.py @@ -175,6 +175,50 @@ def test_unrepresentable_constant_fails_at_write_time(): serialize_resolver(Just(object())) +# --------------------------------------------------------------------------- +# Field-collection isolation +# --------------------------------------------------------------------------- +# +# ``serialize_dimensions``/``serialize_measures`` each wrap their whole +# per-name loop in a single ``@safe``. The call site used to default a +# ``Failure`` away with ``.value_or({})``, so one poisoned dimension or +# measure silently deleted every sibling on the same table too: +# ``to_tagged()`` returned successfully with an empty field set and no error +# anywhere, surfacing later (if at all) as a confusing "unknown dimension" +# error on a query that happened to reference a dropped field. + + +def test_unserializable_dimension_does_not_wipe_its_siblings(table): + from boring_semantic_layer._xorq import Deferred, Just + + poisoned = Deferred(Just(object())) + model = ( + to_semantic_table(table, "m") + .with_dimensions(good=lambda t: t.a, poison=lambda t: poisoned) + .with_measures(n=lambda t: t.count()) + ) + with pytest.raises(ValueError, match="poison"): + to_tagged(model) + + +def test_unserializable_measure_does_not_wipe_its_siblings(): + from boring_semantic_layer.ops import Measure, SemanticTableOp + from boring_semantic_layer.serialization.extract import extract_op_tree + + op = SemanticTableOp( + table=ibis.memtable({"a": [1, 2, 3]}), + dimensions={}, + measures={ + "good": Measure(expr=lambda t: t.a.sum()), + "poison": Measure(expr=lambda t: object()), + }, + calc_measures={}, + name="m", + ) + with pytest.raises(ValueError, match="poison"): + extract_op_tree(op, BSLSerializationContext()) + + # --------------------------------------------------------------------------- # Aggregate replay # --------------------------------------------------------------------------- diff --git a/src/boring_semantic_layer/tests/test_xorq_join_leaf_recovery.py b/src/boring_semantic_layer/tests/test_xorq_join_leaf_recovery.py index d7803f1d..ce9438a3 100644 --- a/src/boring_semantic_layer/tests/test_xorq_join_leaf_recovery.py +++ b/src/boring_semantic_layer/tests/test_xorq_join_leaf_recovery.py @@ -239,3 +239,97 @@ def test_guard_message_names_the_underlying_error(): model = accounts.join_many(transactions, on=lambda a, t: a.account_id == t.account_id) with pytest.raises(ValueError, match="AttributeError.*filter"): from_tagged(to_tagged(model)) + + +# --------------------------------------------------------------------------- +# Tagging an already-aggregated join query (not just the bare model) +# --------------------------------------------------------------------------- +# +# ``to_tagged()`` explicitly supports tagging a ``SemanticAggregateOp`` +# directly (see its ``aggregate_cache_storage`` parameter, for smart-cube +# caching), which tags the FULLY pre-agg-compiled query rather than a bare +# model. Leaf recovery then has to isolate each join leg from that compiled +# tree instead of the original (unaggregated) join chain. + + +@pytest.fixture +def plain_tables(): + return { + "accounts": pd.DataFrame( + { + "account_id": [1, 2, 3], + "customer_id": [10, 10, 20], + } + ), + "customers": pd.DataFrame({"customer_id": [10, 20], "state": ["CA", "NY"]}), + "transactions": pd.DataFrame( + { + "transaction_id": [1, 2, 3, 4], + "account_id": [1, 1, 2, 3], + "amount": [5.0, 6.0, 7.0, 8.0], + } + ), + } + + +def _build_plain_model(tables): + accounts = ( + to_semantic_table(xo.memtable(tables["accounts"], name="accounts"), name="accounts") + .with_dimensions( + account_id=lambda t: t.account_id, + customer_id=lambda t: t.customer_id, + ) + .with_measures(account_count=lambda t: t.count()) + ) + customers = to_semantic_table( + xo.memtable(tables["customers"], name="customers"), name="customers" + ).with_dimensions( + customer_id=lambda t: t.customer_id, + state=lambda t: t.state, + ) + transactions = to_semantic_table( + xo.memtable(tables["transactions"], name="transactions"), name="transactions" + ).with_measures(total_amount=lambda t: t.amount.sum()) + return accounts, customers, transactions + + +def test_join_one_aggregate_query_round_trips(plain_tables): + """A plain two-table join_one (no fan-out) survives tagging the query itself.""" + accounts, customers, _ = _build_plain_model(plain_tables) + joined = accounts.join_one(customers, on=lambda a, c: a.customer_id == c.customer_id) + query = joined.group_by("customers.state").aggregate("accounts.account_count") + + direct = query.to_untagged().execute().sort_values("customers.state").reset_index(drop=True) + recovered = ( + from_tagged(to_tagged(query)) + .to_untagged() + .execute() + .sort_values("customers.state") + .reset_index(drop=True) + ) + pd.testing.assert_frame_equal(direct, recovered) + + +def test_join_many_aggregate_query_fails_loud_not_silently_wrong(plain_tables): + """A fan-out leg under an aggregated, tagged query cannot be recovered today. + + Pre-agg compilation for ``join_many`` rewrites the join into a decomposed + tree of partial-aggregate/key-bridge joins with no single ``JoinChain`` + corresponding to the original leaves, so ``_split_join_expr`` cannot + isolate each leg from the fully-compiled, tagged expression. + ``_validate_join_leaf`` catches the resulting mismatch and raises — + this pins that the failure stays LOUD (a clear, actionable error) rather + than regressing into silently wrong numbers. Tagging the un-aggregated + join, or the bare model, is unaffected (see the other tests in this + file) and is the supported workaround. + """ + accounts, customers, transactions = _build_plain_model(plain_tables) + joined = accounts.join_one(customers, on=lambda a, c: a.customer_id == c.customer_id).join_many( + transactions, on=lambda a, t: a.account_id == t.account_id + ) + query = joined.group_by("customers.state").aggregate("transactions.total_amount") + + assert query.to_untagged().execute() is not None # the direct query itself is fine + + with pytest.raises(ValueError, match="Round-trip could not recover"): + from_tagged(to_tagged(query)) From 6a68bcfafdce62ced8abc58e1da1baf7ddb2e232 Mon Sep 17 00:00:00 2001 From: Hussain Sultan Date: Sun, 23 Aug 2026 17:51:58 -0400 Subject: [PATCH 4/5] =?UTF-8?q?feat:=20leaf=20expression=20markers=20?= =?UTF-8?q?=E2=80=94=20recovery=20reads=20authored=20leaves,=20never=20gue?= =?UTF-8?q?sses=20(#305)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The last of the leaf-recovery family. to_tagged serialized ONE lowered expression plus metadata, and recovery re-derived each model's table from the lowered plan (join splitting + base-relation walking) — guessing where lowering ends and authored expression begins. Every failure in this family was that guess going wrong: seam projections with __bsl_jk_ temporaries, shaped views discarded, aggregate-grain views unrecoverable, query entries losing their shaped base. to_tagged now stamps each leaf model's AUTHORED table expression with a marker tag inside the lowered payload (__bsl_leaf__, metadata {"leaf": }): _collect_leaf_tables maps leaf SemanticTableOps to their table ops (descending join wrappers' _source_join), and _mark_leaf_tables wraps every structural occurrence via a node-equality rewrite — so markers land wherever lowering placed the leaf: under rename projections, under pre-aggregation legs, under a query's Aggregate. Markers are hashing tags: payload-light, profiles ride along, they serialize through xorq's YAML build path unchanged. Recovery (_find_marked_leaf) returns the marked subtree verbatim before any heuristic runs. This makes first-class, catalog-round-trippable: - shaped deferred views (star-schema doctrine), - to_semantic_table(table.group_by(...).aggregate(...)) — aggregate-grain fact models (dimensional-modeling aggregate fact tables), - query entries over shaped bases (the previously strict-xfail gap — that test now passes and is promoted to a regular test). Compatibility: payloads without markers (pre-change, or a leaf whose table op was rewritten by lowering so no node matches) fall back to the existing heuristics unchanged — covered by an explicit strip-the-markers test. Old readers ignore the inner tags (their walks pass through Tag nodes). Note on the rewrite: plain-callable replacers in ibis's replace() must recreate nodes from _kwargs themselves for child substitutions to propagate (Pattern/Mapping replacers get this for free) — the replacer here does; to_tagged's pre-existing replace_read_parquet callable does not and likely no-ops on nested rewrites (upstream xorq note, untouched). Tests: 11 recovery tests pass incl. grouped-grain in-memory + full disk round-trip; serialization battery 115 passed; full suite in a fresh env: 1307 passed, 6 failed — all 6 reproduce on the base commit (optional-dep environment failures: langgraph/mcp/flavor-routing), zero regressions. Co-authored-by: Claude Fable 5 --- .../serialization/__init__.py | 72 +++++++ .../serialization/reconstruct.py | 31 ++- .../tests/test_xorq_join_leaf_recovery.py | 182 ++++++++++++++++-- 3 files changed, 273 insertions(+), 12 deletions(-) diff --git a/src/boring_semantic_layer/serialization/__init__.py b/src/boring_semantic_layer/serialization/__init__.py index eea30bbe..f89c98be 100644 --- a/src/boring_semantic_layer/serialization/__init__.py +++ b/src/boring_semantic_layer/serialization/__init__.py @@ -52,6 +52,77 @@ def do_import(): # to_tagged # --------------------------------------------------------------------------- +#: Marker tag stamped on each leaf model's AUTHORED table expression inside +#: the lowered payload. Recovery reads the marked subtree back verbatim +#: instead of re-deriving the leaf from the lowered plan (base-relation +#: walking / join splitting), which discarded authored shaping and could not +#: see through aggregation lowering at all. metadata: {"tag": BSL_LEAF_TAG, +#: "leaf": }. +BSL_LEAF_TAG = "__bsl_leaf__" + + +def _collect_leaf_tables(op, out=None): + """Map each leaf SemanticTableOp's name to its authored table op. + + Walks the SEMANTIC op tree (source/left/right chains; join wrappers + descend into their _source_join). First declaration wins on a duplicated + name — an ambiguous name cannot be marked meaningfully and falls back to + heuristic recovery. + """ + from .. import ops as bsl_ops + + if out is None: + out = {} + if op is None: + return out + if isinstance(op, bsl_ops.SemanticTableOp): + source_join = getattr(op, "_source_join", None) + if source_join is not None: + return _collect_leaf_tables(source_join, out) + name = getattr(op, "name", None) + table = getattr(op, "table", None) + if table is not None and hasattr(table, "op"): + table = table.op() + if name and table is not None and name not in out: + out[name] = table + return out + for attr in ("source", "left", "right"): + child = getattr(op, attr, None) + if child is not None: + _collect_leaf_tables(child, out) + return out + + +def _mark_leaf_tables(xorq_table, leaf_tables): + """Wrap every occurrence of an authored leaf table in a marker tag. + + The lowered expression embeds each leaf's table op structurally, so a + node-equality rewrite finds them wherever lowering placed them — under + rename projections, under pre-aggregation legs, under a query's + Aggregate. A leaf whose table was itself rewritten by lowering (so no + node matches) is simply left unmarked and recovers via the heuristics. + """ + from .._xorq import replace_nodes + + by_op = {} + for name, table in leaf_tables.items(): + by_op.setdefault(table, name) + if not by_op: + return xorq_table + + def replacer(node, _kwargs): + # Plain-callable replacers must recreate the node from _kwargs + # themselves — that is how child substitutions propagate upward + # (see ibis graph._coerce_replacer; Pattern/Mapping replacers get + # this for free, callables do not). + rebuilt = node.__recreate__(_kwargs) if _kwargs else node + name = by_op.get(node) + if name is None: + return rebuilt + return rebuilt.to_expr().hashing_tag(tag=BSL_LEAF_TAG, leaf=name).op() + + return replace_nodes(replacer, xorq_table).to_expr() + def to_tagged(semantic_expr, aggregate_cache_storage=None): """Tag a BSL expression with serialized metadata. @@ -112,6 +183,7 @@ def extract_path_from_view(table_name): return node xorq_table = replace_nodes(replace_read_parquet, xorq_table).to_expr() + xorq_table = _mark_leaf_tables(xorq_table, _collect_leaf_tables(op)) metadata = extract_op_tree(op, context) tag_data = {k: freeze(v) for k, v in metadata.items()} diff --git a/src/boring_semantic_layer/serialization/reconstruct.py b/src/boring_semantic_layer/serialization/reconstruct.py index f026b4c8..bdaa169e 100644 --- a/src/boring_semantic_layer/serialization/reconstruct.py +++ b/src/boring_semantic_layer/serialization/reconstruct.py @@ -200,8 +200,13 @@ def _reconstruct_table(): _source_join=join_op, ) + marked_leaf = _find_marked_leaf(xorq_expr, metadata.get("name")) return bsl_expr.SemanticModel( - table=_strip_internal_join_temps(_reconstruct_table()), + table=( + marked_leaf + if marked_leaf is not None + else _strip_internal_join_temps(_reconstruct_table()) + ), dimensions=dimensions, measures=measures, calc_measures=calc_measures, @@ -359,6 +364,30 @@ def _reconstruct_limit(metadata: dict, xorq_expr, source, context: BSLSerializat return source.limit(n=int(metadata.get("n", 0)), offset=int(metadata.get("offset", 0))) +def _find_marked_leaf(xorq_expr, name): + """Return the AUTHORED table expression for leaf *name*, if the payload + carries a leaf marker (BSL_LEAF_TAG, written by to_tagged since the + leaf-payload change). This is the lossless recovery path: the marked + subtree is exactly what the author passed to to_semantic_table — + shaped views, renames, even aggregate-grain rollups — so no heuristic + re-derivation from the lowered plan is needed. Returns None when the + payload predates markers or the leaf was not markable (unnamed model, + or its table op was rewritten by lowering).""" + if not name: + return None + from .._xorq import Tag, walk_nodes + + try: + for tag_op in walk_nodes((Tag,), xorq_expr): + metadata = getattr(tag_op, "metadata", None) or {} + if metadata.get("tag") == "__bsl_leaf__" and metadata.get("leaf") == name: + parent = tag_op.parent + return parent.to_expr() if hasattr(parent, "to_expr") else parent + except Exception: + return None + return None + + def _strip_internal_join_temps(expr): """Invert BSL's temporary join-key renames on a recovered leaf table. diff --git a/src/boring_semantic_layer/tests/test_xorq_join_leaf_recovery.py b/src/boring_semantic_layer/tests/test_xorq_join_leaf_recovery.py index ce9438a3..815505bb 100644 --- a/src/boring_semantic_layer/tests/test_xorq_join_leaf_recovery.py +++ b/src/boring_semantic_layer/tests/test_xorq_join_leaf_recovery.py @@ -199,17 +199,11 @@ def test_shaped_join_leaves_round_trip(): assert list(result["transactions.gross_purchases"]) == [0.0, 11.0] -@pytest.mark.xfail( - strict=True, - reason="Known gap: a lowered QUERY entry over a shaped view still base-walks " - "— BSL's query-time injected mutates and authored shaping are " - "indistinguishable in the lowered tree. Needs a lowering-time base-boundary " - "marker. MODEL entries (what the star-schema doctrine catalogs) are covered " - "by the tests above.", -) -def test_query_entry_still_digs_under_the_aggregate(): - """Recovery of a tagged aggregate over a SHAPED view should replay the - query against the shaped base — today the base walk discards the shaping.""" +def test_query_entry_recovers_the_shaped_base(): + """Recovery of a tagged aggregate over a SHAPED view replays the query + against the shaped base: the leaf marker survives under the Aggregate, + so recovery no longer needs to guess where lowering ends and authored + shaping begins (this was a strict xfail before leaf markers).""" accounts_view, _ = _seam_free_tables() model = ( to_semantic_table(accounts_view, name="accounts") @@ -333,3 +327,169 @@ def test_join_many_aggregate_query_fails_loud_not_silently_wrong(plain_tables): with pytest.raises(ValueError, match="Round-trip could not recover"): from_tagged(to_tagged(query)) + + +# --------------------------------------------------------------------------- +# Leaf markers: shaped and aggregate-grain leaves round-trip losslessly +# --------------------------------------------------------------------------- + + +def test_grouped_grain_fact_round_trips(): + """A fact view PRE-AGGREGATED to a coarser grain is a legal model leaf. + + Dimensional-modeling aggregate fact tables: roll transactions to account + grain as a deferred xorq view, then a thin semantic layer of simple sums + on top. Before leaf markers this failed round-trip — the base walk dug + under the Aggregate and the rollup columns vanished. + """ + con = xo.connect() + raw_txn = con.register( + pd.DataFrame( + { + "transaction_id": [1, 2, 3, 4], + "account_id": [1, 1, 2, 3], + "amount": [5.0, 6.0, 7.0, 8.0], + } + ), + "grain_transactions", + ) + raw_accounts = con.register( + pd.DataFrame({"account_id": [1, 2, 3], "customer_id": [10, 10, 20]}), + "grain_accounts", + ) + txn_by_account = raw_txn.group_by("account_id").aggregate( + txn_count=xo._.count(), txn_amount=xo._.amount.sum() + ) + + accounts = to_semantic_table(raw_accounts, name="accounts").with_dimensions( + account_id=lambda t: t.account_id, + customer_id=lambda t: t.customer_id, + ) + fact = to_semantic_table(txn_by_account, name="txn_rollup").with_measures( + total_amount=lambda t: t.txn_amount.sum(), + total_txns=lambda t: t.txn_count.sum(), + ) + model = accounts.join_one(fact, on=lambda a, f: a.account_id == f.account_id) + + recovered = from_tagged(to_tagged(model)) + result = ( + recovered.group_by("accounts.customer_id") + .aggregate("txn_rollup.total_amount", "txn_rollup.total_txns") + .to_untagged() + .execute() + .sort_values("accounts.customer_id") + .reset_index(drop=True) + ) + assert list(result["txn_rollup.total_amount"]) == [18.0, 8.0] + assert list(result["txn_rollup.total_txns"]) == [3, 1] + + +def test_single_table_model_over_ibis_aggregate_round_trips(): + """to_semantic_table(table.group_by(...).aggregate(...)) — a semantic + model DIRECTLY over an ibis/xorq aggregate query — is a legal model. + + The aggregate view fixes the grain (account level here); the semantic + layer names simple reductions over the rollup columns. Everything stays + deferred; the marker carries the authored aggregate through round-trip. + """ + con = xo.connect() + raw_txn = con.register( + pd.DataFrame( + { + "transaction_id": [1, 2, 3, 4], + "account_id": [1, 1, 2, 3], + "amount": [5.0, 6.0, 7.0, 8.0], + } + ), + "st_transactions", + ) + rollup = raw_txn.group_by("account_id").aggregate( + txn_count=xo._.count(), txn_amount=xo._.amount.sum() + ) + model = ( + to_semantic_table(rollup, name="account_rollup") + .with_dimensions(account_id=lambda t: t.account_id) + .with_measures( + accounts_with_txns=lambda t: t.count(), + total_amount=lambda t: t.txn_amount.sum(), + max_account_amount=lambda t: t.txn_amount.max(), + ) + ) + + recovered = from_tagged(to_tagged(model)) + result = ( + recovered.aggregate("accounts_with_txns", "total_amount", "max_account_amount") + .to_untagged() + .execute() + ) + assert list(result["accounts_with_txns"]) == [3] + assert list(result["total_amount"]) == [26.0] + assert list(result["max_account_amount"]) == [11.0] + + +def test_grouped_grain_model_survives_disk_round_trip(tmp_path): + """The aggregate-grain model survives xorq build -> load_expr -> + ls.builder — the full catalog path.""" + pytest.importorskip("duckdb") + db = str(tmp_path / "wh.ddb") + seed = xo.duckdb.connect(db) + seed.create_table( + "transactions", + pd.DataFrame( + { + "transaction_id": [1, 2, 3, 4], + "account_id": [1, 1, 2, 3], + "amount": [5.0, 6.0, 7.0, 8.0], + } + ), + ) + del seed + con = xo.duckdb.connect(db) + rollup = ( + con.table("transactions") + .group_by("account_id") + .aggregate(txn_amount=xo._.amount.sum()) + ) + model = to_semantic_table(rollup, name="account_rollup").with_measures( + total_amount=lambda t: t.txn_amount.sum(), + ) + tagged = to_tagged(model) + + path = xo.build_expr(tagged, builds_dir=str(tmp_path / "builds")) + loaded = xo.load_expr(path) + recovered = loaded.ls.builder + result = recovered.aggregate("total_amount").to_untagged().execute() + assert list(result["total_amount"]) == [26.0] + + +def test_unmarked_payload_falls_back_to_heuristics(): + """Payloads written before leaf markers (no __bsl_leaf__ tags) must keep + recovering via the heuristic paths — strip the marker tags off a fresh + payload to simulate one.""" + from boring_semantic_layer._xorq import Tag, walk_nodes + from boring_semantic_layer.serialization import BSL_LEAF_TAG + + accounts_view, _txn_view = _seam_free_tables() + accounts = to_semantic_table(accounts_view, name="accounts").with_measures( + open_account_count=lambda t: t.is_open.sum(), + ) + tagged = to_tagged(accounts) + + from xorq.common.utils.graph_utils import replace_nodes + + def strip(node, _kwargs): + rebuilt = node.__recreate__(_kwargs) if _kwargs else node + if isinstance(rebuilt, Tag) and (rebuilt.metadata or {}).get("tag") == BSL_LEAF_TAG: + return rebuilt.parent + return rebuilt + + stripped = replace_nodes(strip, tagged).to_expr() + assert not [ + t + for t in walk_nodes((Tag,), stripped) + if (getattr(t, "metadata", {}) or {}).get("tag") == BSL_LEAF_TAG + ] + recovered = from_tagged(stripped) + result = recovered.aggregate("open_account_count").to_untagged().execute() + # heuristic per-row-chain preservation still recovers the shaped leaf + assert list(result["open_account_count"]) == [2] From 0fb98e28e50ab7e6e169f3b7465d89e3628401ee Mon Sep 17 00:00:00 2001 From: Hussain Sultan Date: Sun, 23 Aug 2026 21:25:34 -0400 Subject: [PATCH 5/5] style: fix ruff format violation in join leaf recovery tests CI lint was failing on `ruff format --check .` for a line that exceeded the formatter's preferred line length. --- .../tests/test_xorq_join_leaf_recovery.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/src/boring_semantic_layer/tests/test_xorq_join_leaf_recovery.py b/src/boring_semantic_layer/tests/test_xorq_join_leaf_recovery.py index 815505bb..0b741c26 100644 --- a/src/boring_semantic_layer/tests/test_xorq_join_leaf_recovery.py +++ b/src/boring_semantic_layer/tests/test_xorq_join_leaf_recovery.py @@ -446,9 +446,7 @@ def test_grouped_grain_model_survives_disk_round_trip(tmp_path): del seed con = xo.duckdb.connect(db) rollup = ( - con.table("transactions") - .group_by("account_id") - .aggregate(txn_amount=xo._.amount.sum()) + con.table("transactions").group_by("account_id").aggregate(txn_amount=xo._.amount.sum()) ) model = to_semantic_table(rollup, name="account_rollup").with_measures( total_amount=lambda t: t.txn_amount.sum(),