fix: recover join leaves poisoned by internal __bsl_jk_ temporaries

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": <model name>}.
+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()}

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:

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
@@ -167,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=_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,
@@ -326,6 +364,69 @@ 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.
+
+    ``SemanticJoinOp.to_untagged`` renames left-side predicate columns that
+    collide across the join to ``__bsl_jk_<name>`` (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_<name>_N`` overflow spelling (a user column literally
+    named ``__bsl_jk_<name>`` 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.
 
@@ -351,10 +452,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

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