diff --git a/examples/dimensional_indexing.py b/examples/dimensional_indexing.py
index 08b68ee8..363cc526 100644
--- a/examples/dimensional_indexing.py
+++ b/examples/dimensional_indexing.py
@@ -210,7 +210,7 @@ def get_autocomplete_suggestions(prefix: str, limit: int = 10):
     flights_with_origin = flights.join_one(airports, lambda f, a: f.origin == a.code)
 
     joined_index = (
-        flights_with_origin.index(s.cols("carrier", "airports__state"))
+        flights_with_origin.index(s.cols("carrier", "airports.state"))
         .order_by(lambda t: t.weight.desc())
         .limit(15)
         .execute()
diff --git a/src/boring_semantic_layer/calc_compiler.py b/src/boring_semantic_layer/calc_compiler.py
index 4ddbbb6c..a6ca50b7 100644
--- a/src/boring_semantic_layer/calc_compiler.py
+++ b/src/boring_semantic_layer/calc_compiler.py
@@ -414,9 +414,12 @@ def compile_calc_measure(
         )
         for col_name in totals_schema:
             prefixed = f"{totals_prefix}{col_name}"
-            target_name = prefixed if prefixed in rwt_columns else col_name
-            if target_name in rwt_columns:
-                subs[Field(totals_vt_op, col_name)] = Field(rwt_op, target_name)
+            # Only substitute the real TOTALS column. Falling back to the
+            # unprefixed per-group column silently replaced the grand total
+            # with the group's own value (every share became 1.0) and made
+            # the unresolved-reference guard below unreachable.
+            if prefixed in rwt_columns:
+                subs[Field(totals_vt_op, col_name)] = Field(rwt_op, prefixed)
 
     rewritten = op.replace(subs)
 
@@ -635,21 +638,19 @@ def attach_windowed_totals(
         try:
             agg_expr = agg_specs[name](new_base)
         except Exception as exc:
-            logger.debug(
-                "could not evaluate agg_spec for %r when attaching windowed totals: %s",
-                name,
-                exc,
-            )
-            continue
+            raise TotalsNotAvailableError(
+                f"Could not evaluate the aggregation for {name!r} while "
+                "attaching totals; a skipped totals column would silently "
+                "substitute the per-group value for the grand total."
+            ) from exc
         try:
             windowed = agg_expr.over(ibis_mod.window())
         except Exception as exc:
-            logger.debug(
-                "could not wrap %r in window() for windowed totals: %s",
-                name,
-                exc,
-            )
-            continue
+            raise TotalsNotAvailableError(
+                f"Could not window the aggregation for {name!r} while "
+                "attaching totals; a skipped totals column would silently "
+                "substitute the per-group value for the grand total."
+            ) from exc
         col = f"{totals_prefix}{name}"
         new_base = new_base.mutate(**{col: windowed})
         arbitrary_specs[col] = (lambda t, _c=col: t[_c].arbitrary())
@@ -772,10 +773,11 @@ def attach_calc_totals(
             else:
                 totals_expr = fn
         except Exception as exc:
-            logger.debug(
-                "calc-of-calc totals evaluation failed for %r: %s", calc_name, exc
-            )
-            continue
+            raise TotalsNotAvailableError(
+                f"Could not evaluate totals for calc measure {calc_name!r}; "
+                "a skipped totals column would silently substitute the "
+                "per-group value for the grand total."
+            ) from exc
         col = f"{totals_prefix}{calc_name}"
         real_agg_tbl = real_agg_tbl.mutate(**{col: totals_expr})
 
diff --git a/src/boring_semantic_layer/expr.py b/src/boring_semantic_layer/expr.py
index db599052..cbe402b3 100644
--- a/src/boring_semantic_layer/expr.py
+++ b/src/boring_semantic_layer/expr.py
@@ -1107,7 +1107,7 @@ def join_one(
         self,
         other: SemanticModel,
         on: Callable[[Any, Any], ir.BooleanValue] | str | Deferred | Sequence[str | Deferred],
-        how: str = "inner",
+        how: str = "left",
     ) -> SemanticJoin:
         """Join with one-to-one relationship semantics."""
         return SemanticJoin(
@@ -1243,7 +1243,7 @@ def join_one(
         self,
         other: SemanticModel,
         on: Callable[[Any, Any], ir.BooleanValue] | str | Deferred | Sequence[str | Deferred],
-        how: str = "inner",
+        how: str = "left",
     ) -> SemanticJoin:
         """Join with one-to-one relationship semantics."""
         return SemanticJoin(
@@ -1289,6 +1289,18 @@ def __init__(self, source: SemanticTableOp, keys: tuple[str, ...]) -> None:
         op = SemanticGroupByOp(source=source, keys=keys)
         super().__init__(op)
 
+    def filter(self, predicate: Callable) -> SemanticGroupBy:
+        """Filter rows before aggregation, keeping the grouping keys.
+
+        A filter between group_by and aggregate is pre-aggregation row
+        filtering, so it commutes with the grouping. The inherited filter
+        wrapped the group-by op itself, and aggregate() — which only
+        recovers keys from its direct source — silently dropped the
+        requested grouping.
+        """
+        filtered = SemanticFilter(source=self.op().source, predicate=predicate)
+        return SemanticGroupBy(source=filtered.op(), keys=self.op().keys)
+
     @property
     def source(self):
         return self.op().source
@@ -1469,7 +1481,7 @@ def join_one(
         self,
         other: SemanticModel,
         on: Callable[[Any, Any], ir.BooleanValue] | str | Deferred | Sequence[str | Deferred],
-        how: str = "inner",
+        how: str = "left",
     ) -> SemanticJoin:
         """Join with one-to-one relationship semantics."""
         return SemanticJoin(
diff --git a/src/boring_semantic_layer/nested_compile.py b/src/boring_semantic_layer/nested_compile.py
index 870e57fc..9c77cbd7 100644
--- a/src/boring_semantic_layer/nested_compile.py
+++ b/src/boring_semantic_layer/nested_compile.py
@@ -140,7 +140,10 @@ def join_tables(by_cols: Iterable[str], tables: list) -> Any:
     by_cols_set = set(by_cols)
 
     def join_step(left, right):
-        predicates = [left[c] == right[c] for c in by_cols]
+        # Null-safe equality: group keys can legitimately be NULL (real NULL
+        # dim values, or keys minted by an outer join). Plain == drops those
+        # groups from every table but the first.
+        predicates = [left[c].identical_to(right[c]) for c in by_cols]
         right_cols = [c for c in right.columns if c not in by_cols_set]
         right_select = [right[c] for c in right_cols]
         return left.left_join(right, predicates).select([left] + right_select)
diff --git a/src/boring_semantic_layer/ops.py b/src/boring_semantic_layer/ops.py
index 8d333b9f..1bc870f4 100644
--- a/src/boring_semantic_layer/ops.py
+++ b/src/boring_semantic_layer/ops.py
@@ -556,6 +556,19 @@ def _mutate_dimensions_with_dependencies(
     """Mutate requested dimensions, recursively materializing derived deps first."""
     resolving: list[str] = []
 
+    # Dim lambdas reference sibling dims by their BARE name (t.region_band),
+    # but merged dimension maps key them by prefixed name on joins
+    # (customers.region_band). Alias unambiguous suffixes so dependency
+    # resolution can materialize them under the name the lambda reads.
+    merged_dimensions = dict(merged_dimensions)
+    _by_suffix: dict[str, list[str]] = {}
+    for _name in merged_dimensions:
+        if "." in _name:
+            _by_suffix.setdefault(_name.split(".", 1)[1], []).append(_name)
+    for _short, _fulls in _by_suffix.items():
+        if _short not in merged_dimensions and len(_fulls) == 1:
+            merged_dimensions[_short] = merged_dimensions[_fulls[0]]
+
     def resolve_one(dim_name: str, current_tbl: ir.Table) -> ir.Table:
         if dim_name not in merged_dimensions:
             return current_tbl
@@ -2188,6 +2201,141 @@ def _table_filter_resolver(raw_tbl, table_op, table_name):
     return _Resolver(raw_tbl, dims)
 
 
+_FIELD_TYPES = tuple({ibis_ops.Field, xorq_ops.Field})
+_AND_TYPES = tuple({ibis_ops.And, xorq_ops.And})
+
+
+def _leaf_rel_types():
+    """Base relation classes for both ibis flavors (plus xorq Read)."""
+    from ._xorq import Read as _XorqRead
+
+    types = {
+        ibis_ops.DatabaseTable,
+        ibis_ops.InMemoryTable,
+        xorq_ops.DatabaseTable,
+        xorq_ops.InMemoryTable,
+    }
+    if _XorqRead is not None:
+        types.add(_XorqRead)
+    return tuple(types)
+
+
+def _flatten_and_legs(expr):
+    """Flatten a boolean expression's top-level AND chain into legs."""
+    op = expr.op()
+    if isinstance(op, _AND_TYPES):
+        return _flatten_and_legs(op.left.to_expr()) + _flatten_and_legs(op.right.to_expr())
+    return [expr]
+
+
+def _value_fields(value_op):
+    """Fields referenced by a value op, without descending into relations.
+
+    Descending into a Field's relation would surface every column of the
+    join tree; provenance only wants the fields the value itself reads.
+    """
+    from .graph_utils import gen_children_of
+
+    out, stack, seen = [], [value_op], set()
+    while stack:
+        node = stack.pop()
+        if node in seen:
+            continue
+        seen.add(node)
+        if isinstance(node, _FIELD_TYPES):
+            out.append(node)
+            continue
+        if isinstance(node, (Relation, xorq_ops.Relation)):
+            continue
+        stack.extend(gen_children_of(node))
+    return out
+
+
+def _field_base_relations(field_op, leaf_types, guard=0):
+    """Resolve a Field down to the base relation(s) its value derives from."""
+    if guard > 100:
+        return {None}
+    rel, name = field_op.rel, field_op.name
+    if isinstance(rel, leaf_types):
+        return {rel}
+    values = getattr(rel, "values", None)
+    if values is not None and name in values:
+        bases = set()
+        for f in _value_fields(values[name]):
+            bases |= _field_base_relations(f, leaf_types, guard + 1)
+        return bases
+    parent = getattr(rel, "parent", None)
+    if parent is not None:
+        return _field_base_relations(field_op.__class__(parent, name), leaf_types, guard + 1)
+    return {rel}
+
+
+def _base_rel_key(rel):
+    """Structural fingerprint for a base relation.
+
+    Separately-built untagged tables wrap distinct backend instances, so
+    node equality fails even for the same physical table; match on class,
+    table name and schema instead.
+    """
+    try:
+        schema_names = tuple(rel.schema.names)
+    except Exception:
+        schema_names = ()
+    return (type(rel).__name__, getattr(rel, "name", None), schema_names)
+
+
+def _leg_source_tables(leg_expr, base_rel_to_table, leaf_types):
+    """Names of the source tables a filter leg's fields derive from."""
+    sources = set()
+    for f in _value_fields(leg_expr.op()):
+        for base in _field_base_relations(f, leaf_types):
+            sources.add(base_rel_to_table.get(_base_rel_key(base), "__unknown__"))
+    return sources
+
+
+def _inline_to_base_op(node, leaf_types, target_tbl=None, guard=0):
+    """Rewrite a value op so every Field references a base relation.
+
+    Projection/filter chains between the joined table and the base are
+    inlined, producing an expression that can be re-applied to the owning
+    table's raw table (row-precise filter pushdown). When ``target_tbl``
+    is given, base fields are rebased onto it by column name — the join's
+    copy of a base relation wraps a different backend instance, so node
+    identity alone would fail the Filter integrity check.
+    """
+    if guard > 200:
+        raise ValueError("expression too deep to rebind")
+    if isinstance(node, _FIELD_TYPES):
+        rel, name = node.rel, node.name
+        if isinstance(rel, leaf_types):
+            if target_tbl is not None:
+                return target_tbl[name].op()
+            return node
+        values = getattr(rel, "values", None)
+        if values is not None and name in values:
+            return _inline_to_base_op(values[name], leaf_types, target_tbl, guard + 1)
+        parent = getattr(rel, "parent", None)
+        if parent is not None:
+            return _inline_to_base_op(
+                node.__class__(parent, name), leaf_types, target_tbl, guard + 1
+            )
+        return node
+    if isinstance(node, (Relation, xorq_ops.Relation)):
+        raise ValueError("cannot rebind a predicate containing a subquery")
+
+    def _tx(a):
+        if isinstance(a, tuple):
+            return tuple(_tx(x) for x in a)
+        if isinstance(a, _FIELD_TYPES) or hasattr(a, "__argnames__"):
+            return _inline_to_base_op(a, leaf_types, target_tbl, guard + 1)
+        return a
+
+    new_args = [_tx(a) for a in node.args]
+    if all(n is o for n, o in zip(new_args, node.args)):
+        return node
+    return node.__class__(**dict(zip(node.__argnames__, new_args)))
+
+
 def _find_deferrable_joins(
     join_op,
     group_by_keys: tuple[str, ...],
@@ -2294,6 +2442,25 @@ def walk(node):
         if not deferred_dims:
             return
 
+        # Deferral attaches dim labels to entity-grain rows WITHOUT
+        # re-aggregating, so it is only sound when the requested group keys
+        # already pin the entity grain. Grouping by a coarser attribute
+        # alone (e.g. customers.region) must go through the pre-agg path,
+        # which re-groups correctly; deferring it returns one row per
+        # entity with duplicated dim values.
+        left_key_names = frozenset(left_cols)
+
+        def _key_covers_entity(entity_name):
+            candidates = {entity_name, f"{right_name}.{entity_name}"}
+            for k in group_by_keys:
+                short = k.split(".", 1)[-1]
+                if k in candidates or short == entity_name or short in left_key_names:
+                    return True
+            return False
+
+        if not all(_key_covers_entity(e) for e in entity_dims):
+            return
+
         deferrable.append(_DeferrableJoin(
             table_name=right_name,
             table_op=right,
@@ -2308,7 +2475,8 @@ def walk(node):
 
 def _left_join_bridge(left, bridge, common_keys):
     """Left-join *bridge* onto *left*, selecting only new columns from bridge."""
-    preds = [left[c] == bridge[c] for c in common_keys]
+    # Null-safe equality so NULL-valued keys still pair up
+    preds = [left[c].identical_to(bridge[c]) for c in common_keys]
     bridge_only = tuple(c for c in bridge.columns if c not in frozenset(common_keys))
     return left.left_join(bridge, preds).select([left] + [bridge[c] for c in bridge_only])
 
@@ -2406,11 +2574,17 @@ def _is_count_distinct_expr(expr):
 
 
 def _reagg_op_for_expr(expr):
-    """Return the correct re-aggregation operation name for an ibis expression.
+    """Return the re-aggregation operation name for an ibis expression.
+
+    Additive measures (SUM, COUNT) re-aggregate with ``sum``; MIN and MAX
+    with ``min``/``max``. MEAN never reaches here (decomposed by
+    ``_is_mean_expr``), nor does COUNT DISTINCT (deferred).
 
-    Additive measures (SUM, COUNT) re-aggregate with ``sum``.
-    MIN and MAX re-aggregate with ``min`` and ``max`` respectively.
-    MEAN should never reach here — it is decomposed by ``_is_mean_expr``.
+    Returns ``None`` for everything else — median, stddev, variance,
+    compound expressions like ``sum()/count()`` — which cannot be
+    re-aggregated from a finer pre-aggregate at all. Those measures must
+    be computed at the exact target grain (``_exact_grain_preagg``);
+    the previous ``"sum"`` default silently summed per-key medians.
     """
     op = expr.op()
     reductions = _reductions_for_expr(expr)
@@ -2418,6 +2592,8 @@ def _reagg_op_for_expr(expr):
         return "min"
     if isinstance(op, reductions.Max):
         return "max"
+    if isinstance(op, (reductions.Sum, reductions.Count, reductions.CountStar)):
+        return "sum"
     if isinstance(op, reductions.Mean):
         raise ValueError(
             f"Mean expression {expr.get_name()!r} was not decomposed — "
@@ -2428,7 +2604,7 @@ def _reagg_op_for_expr(expr):
             f"CountDistinct expression {expr.get_name()!r} was not deferred — "
             "this is a bug in the pre-aggregation logic"
         )
-    return "sum"
+    return None
 
 
 def _build_reagg(col_ref, op_name):
@@ -2436,6 +2612,49 @@ def _build_reagg(col_ref, op_name):
     return getattr(col_ref, op_name)()
 
 
+def _exact_grain_preagg(raw_tbl, tbl, group_by_cols, join_keys, exact_measures):
+    """Aggregate non-decomposable measures at the exact target grain.
+
+    Median, stddev, variance and compound expressions (``sum()/count()``)
+    cannot be re-aggregated from a finer pre-aggregate. Build a
+    (group keys -> join keys) bridge from the joined table and aggregate
+    the raw rows directly per group: each raw row participates once per
+    group-key value it maps to, matching the join-participation semantics
+    of the decomposable path. Raises instead of degrading — the previous
+    behavior summed per-key values silently.
+    """
+    names = ", ".join(sorted(exact_measures))
+    if tbl is None:
+        raise ValueError(
+            f"Cannot compute non-decomposable measure(s) {names} at a "
+            "cross-table grain: the joined table is unavailable."
+        )
+    missing = [c for c in group_by_cols if c not in tbl.columns]
+    if missing:
+        raise ValueError(
+            f"Cannot compute non-decomposable measure(s) {names}: group "
+            f"key(s) {missing} are not materialized on the joined table."
+        )
+    shared_jk = [k for k in join_keys if k in tbl.columns]
+    if not shared_jk:
+        raise ValueError(
+            f"Cannot compute non-decomposable measure(s) {names}: no join "
+            "keys shared with the joined table to bridge the target grain."
+        )
+    # Temp names so bridge group columns can never shadow raw columns the
+    # measure expressions reference
+    tmp = {c: f"__exact_gb_{i}" for i, c in enumerate(group_by_cols)}
+    bridge = tbl.select(
+        [tbl[c].name(tmp[c]) for c in group_by_cols] + [tbl[k] for k in shared_jk]
+    ).distinct()
+    preds = [bridge[k].identical_to(raw_tbl[k]) for k in shared_jk]
+    joined = bridge.inner_join(raw_tbl, preds)
+    aggs = {m: fn(joined) for m, fn in exact_measures.items()}
+    pt = joined.group_by([joined[t] for t in tmp.values()]).aggregate(**aggs)
+    # ibis rename convention: {new_name: old_name}
+    return pt.rename({orig: tmp_name for orig, tmp_name in tmp.items()})
+
+
 def _partition_agg_specs_by_source(
     agg_specs: dict[str, Callable],
     all_roots: list[SemanticTableOp],
@@ -2635,9 +2854,16 @@ def collect_filters_to_join(node):
                         tbl, unprefixed_keys, root_dims,
                     )
                     # Apply pre-aggregation filters on the dimension table.
+                    # Resolve through the table-scoped resolver so prefixed
+                    # (t["customers.region"]) and derived-dim references
+                    # work the same as bare column access.
                     for flt in dim_filters:
                         fn = _unwrap(flt) if hasattr(flt, "unwrap") else flt
-                        tbl = tbl.filter(fn(tbl))
+                        tbl = tbl.filter(
+                            _resolve_expr(
+                                fn, _table_filter_resolver(tbl, root_op, root_op.name)
+                            )
+                        )
                     result = tbl.select(unprefixed_keys).distinct()
                     # Rename columns to their prefixed (dotted) names so that
                     # downstream consumers see the expected column names.
@@ -2791,17 +3017,38 @@ def _to_untagged_with_preagg(
         # owning source table below; anything handled by neither path
         # raises instead of silently dropping the filter.
         filters_on_tbl: set[int] = set()
+        tbl_filter_exprs: dict[int, Any] = {}
         if tbl is not None and filter_fns:
             from .convert import _Resolver
 
+            # Bare aliases for prefixed dims: a filter written `t.size`
+            # against a join where exactly one table declares `size` must
+            # resolve on the joined table too — otherwise the dim bridge is
+            # built from the UNFILTERED join and sibling tables' measures
+            # silently ignore the filter. Physical columns keep priority
+            # (aliases are only added for names that are not columns of the
+            # joined table), and ambiguous suffixes get no alias so they
+            # still hit the loud ownership check below.
+            dims_for_tbl = dict(merged_dimensions)
+            tbl_cols = set(tbl.columns)
+            _by_suffix: dict[str, list[str]] = {}
+            for dname in merged_dimensions:
+                if "." in dname:
+                    _by_suffix.setdefault(dname.split(".", 1)[1], []).append(dname)
+            for short, fulls in _by_suffix.items():
+                if short not in dims_for_tbl and short not in tbl_cols and len(fulls) == 1:
+                    dims_for_tbl[short] = merged_dimensions[fulls[0]]
+
             for i, pred_fn in enumerate(filter_fns):
                 try:
-                    resolver = _Resolver(tbl, merged_dimensions)
-                    filtered = tbl.filter(_resolve_expr(pred_fn, resolver))
+                    resolver = _Resolver(tbl, dims_for_tbl)
+                    pred_expr = _resolve_expr(pred_fn, resolver)
+                    filtered = tbl.filter(pred_expr)
                 except Exception:
                     continue
                 tbl = filtered
                 filters_on_tbl.add(i)
+                tbl_filter_exprs[i] = pred_expr
 
         # --- 1b. Determine which source table(s) each filter belongs to ---
         # Ownership resolution uses each table's own dimensions (bare and
@@ -2846,6 +3093,47 @@ def _to_untagged_with_preagg(
                     'table prefix (e.g. t["orders.status"]).'
                 )
 
+        # --- 1c. Split cross-table conjunctions into per-table legs ---
+        # A compound like (t["orders.status"]=="open") & (t.qty >= 2) has no
+        # single owner, so it used to reach the many side only through a
+        # join-KEY bridge, keeping every item of any qualifying order — the
+        # item-level leg was silently dropped. Split top-level ANDs and track
+        # each leg's source tables via field provenance so legs can be pushed
+        # row-precisely to the table they constrain.
+        filter_legs: dict[int, list] = {}
+        many_side_tables: set[str] = set()
+        if tbl_filter_exprs:
+            leaf_types = _leaf_rel_types()
+            base_rel_to_table: dict = {}
+            for tname, raw in raw_tables.items():
+                for leaf in walk_nodes(leaf_types, raw):
+                    key = _base_rel_key(leaf)
+                    # Same physical table on both sides (self-join): a leg
+                    # can't be attributed to one alias — never match.
+                    if base_rel_to_table.get(key, tname) != tname:
+                        base_rel_to_table[key] = "__ambiguous__"
+                    else:
+                        base_rel_to_table[key] = tname
+            for i, expr in tbl_filter_exprs.items():
+                if filter_owners[i] and len(filter_owners[i]) == 1:
+                    continue  # whole filter pushes to its single owner
+                filter_legs[i] = [
+                    (leg, _leg_source_tables(leg, base_rel_to_table, leaf_types))
+                    for leg in _flatten_and_legs(expr)
+                ]
+
+            def _collect_many_sides(node):
+                if isinstance(node, SemanticJoinOp):
+                    if node.cardinality == "many":
+                        for r in _find_all_root_models(node.right):
+                            if getattr(r, "name", None):
+                                many_side_tables.add(r.name)
+                    _collect_many_sides(node.left)
+                    _collect_many_sides(node.right)
+
+            if filter_legs:
+                _collect_many_sides(join_op)
+
         # --- 2. Build aggregation plan ---
         if tbl is not None:
             scope = MeasureScope(
@@ -2921,6 +3209,7 @@ def _to_untagged_with_preagg(
             # by another table) reach this table via a join-key bridge.
             if filter_fns:
                 needs_bridge = False
+                residual_cross_legs = False
                 for i, pred_fn in enumerate(filter_fns):
                     if filter_owners[i] == frozenset({table_name}):
                         pred_expr = _resolve_expr(
@@ -2928,8 +3217,32 @@ def _to_untagged_with_preagg(
                             _table_filter_resolver(raw_tbl, table_op, table_name),
                         )
                         raw_tbl = raw_tbl.filter(pred_expr)
-                    else:
-                        needs_bridge = True
+                        continue
+                    needs_bridge = True
+                    # Push this table's legs of a cross-table conjunction at
+                    # row grain; legs spanning tables (cross-table OR) keep
+                    # row-level information the key bridge cannot recover.
+                    for leg_expr, leg_srcs in filter_legs.get(i, ()):
+                        if leg_srcs == {table_name}:
+                            try:
+                                leg_op = _inline_to_base_op(
+                                    leg_expr.op(), _leaf_rel_types(), target_tbl=raw_tbl
+                                )
+                                raw_tbl = raw_tbl.filter(leg_op.to_expr())
+                            except Exception:
+                                residual_cross_legs = True
+                        elif table_name in leg_srcs and len(leg_srcs) > 1:
+                            residual_cross_legs = True
+
+                if residual_cross_legs and table_name in many_side_tables and measures:
+                    raise ValueError(
+                        f"A filter mixes columns of {table_name!r} with other "
+                        "tables in a way that cannot be applied row-precisely "
+                        f"to {table_name!r} (e.g. OR across tables); its "
+                        "measures would be silently inflated to join-key "
+                        "grain. Split the condition into separate .filter() "
+                        "calls, or restate it against a single table."
+                    )
 
                 # Filters not pushed here (cross-table, ambiguous, or owned
                 # by another table) restrict via join keys from the filtered
@@ -2979,6 +3292,7 @@ def _to_untagged_with_preagg(
             # Build agg expressions on the raw table
             agg_exprs: dict = {}
             _tot_exprs: dict = {}
+            _exact_measures_t: dict = {}
             for mname, _mfn in measures.items():
                 short = mname.split(".", 1)[1] if "." in mname else mname
                 if short in table_measures:
@@ -2990,10 +3304,15 @@ def _to_untagged_with_preagg(
                     if _is_mean_expr(expr):
                         mean_op = expr.op()
                         base_col = mean_op.arg.to_expr()
+                        # mean(where=...) must filter both legs, or the
+                        # decomposed mean silently ignores its condition
+                        mean_where = (
+                            mean_op.where.to_expr() if mean_op.where is not None else None
+                        )
                         sum_col = f"_sum__{mname}"
                         count_col = f"_count__{mname}"
-                        agg_exprs[sum_col] = base_col.sum()
-                        agg_exprs[count_col] = base_col.count()
+                        agg_exprs[sum_col] = base_col.sum(where=mean_where)
+                        agg_exprs[count_col] = base_col.count(where=mean_where)
                         _decomposed_means[mname] = (sum_col, count_col)
                     elif _is_count_distinct_expr(expr):
                         # COUNT DISTINCT is immune to fan-out — defer past pre-agg
@@ -3001,13 +3320,21 @@ def _to_untagged_with_preagg(
                             table_name, short, raw_tbl, table_measures[short],
                         )
                     else:
-                        _reagg_ops[mname] = _reagg_op_for_expr(expr)
-                        agg_exprs[mname] = expr
+                        reagg = _reagg_op_for_expr(expr)
+                        if reagg is None and group_by_cols:
+                            # Non-decomposable (median, stddev, compound
+                            # ratio): computed at the exact target grain
+                            # after the grain decision below
+                            _exact_measures_t[mname] = table_measures[short]
+                        else:
+                            if reagg is not None:
+                                _reagg_ops[mname] = reagg
+                            agg_exprs[mname] = expr
 
             if _tot_exprs:
                 _totals_sources[table_name] = (raw_tbl, _tot_exprs)
 
-            if not agg_exprs:
+            if not agg_exprs and not _exact_measures_t:
                 continue
 
             # --- Compute grain ---
@@ -3031,18 +3358,39 @@ def _to_untagged_with_preagg(
                     if prefix == table_name and short in table_dims:
                         dim_fn = table_dims[short]
                         if callable(dim_fn):
-                            dim_expr = dim_fn(raw_tbl)
+                            resolved_via_deps = False
+                            try:
+                                dim_expr = dim_fn(raw_tbl)
+                            except Exception:
+                                # Derived dim referencing other derived dims
+                                raw_tbl = _mutate_dimensions_with_dependencies(
+                                    raw_tbl, [short], table_dims
+                                )
+                                raw_columns = set(raw_tbl.columns)
+                                dim_expr = raw_tbl[short]
+                                resolved_via_deps = True
                             col_name = dim_expr.get_name()
-                            if col_name == short and col_name in raw_columns:
+                            if (
+                                not resolved_via_deps
+                                and col_name == short
+                                and col_name in raw_columns
+                            ):
                                 # Simple column reference — use directly
                                 if col_name not in _local_dims:
                                     _local_dims.append(col_name)
-                            elif col_name in raw_columns or short not in raw_columns:
-                                # Derived dimension — materialize on raw_tbl
-                                raw_tbl = raw_tbl.mutate(**{short: dim_expr})
+                            elif (
+                                col_name in raw_columns
+                                or short not in raw_columns
+                                or resolved_via_deps
+                            ):
+                                # Derived dimension — materialize under the
+                                # requested (prefixed) name so the pre-agg
+                                # grain matches the group-by keys and the
+                                # key column survives into the result
+                                raw_tbl = raw_tbl.mutate(**{gb_key: dim_expr})
                                 raw_columns = set(raw_tbl.columns)
-                                if short not in _local_dims:
-                                    _local_dims.append(short)
+                                if gb_key not in _local_dims:
+                                    _local_dims.append(gb_key)
                     elif prefix != table_name:
                         has_cross_table_gb = True
                 elif gb_key in merged_dimensions:
@@ -3086,12 +3434,26 @@ def _to_untagged_with_preagg(
                 case _:
                     grain = tuple(_local_dims)
 
-            if grain:
-                _preagg_results.append(
-                    raw_tbl.group_by([raw_tbl[c] for c in grain]).aggregate(**agg_exprs)
-                )
-            else:
-                _preagg_results.append(raw_tbl.aggregate(**agg_exprs))
+            if _exact_measures_t:
+                if not has_cross_table_gb:
+                    # Local grain IS the target grain — aggregate the
+                    # original expressions directly, no re-agg happens
+                    for m, fn in _exact_measures_t.items():
+                        agg_exprs[m] = fn(raw_tbl)
+                else:
+                    _preagg_results.append(
+                        _exact_grain_preagg(
+                            raw_tbl, tbl, group_by_cols, available_jk, _exact_measures_t
+                        )
+                    )
+
+            if agg_exprs:
+                if grain:
+                    _preagg_results.append(
+                        raw_tbl.group_by([raw_tbl[c] for c in grain]).aggregate(**agg_exprs)
+                    )
+                else:
+                    _preagg_results.append(raw_tbl.aggregate(**agg_exprs))
 
         # Freeze mutable accumulators
         preagg_results = tuple(_preagg_results)
@@ -3390,18 +3752,27 @@ def strip_deferred(node):
                     if callable(dim_fn):
                         try:
                             expr = dim_fn(dim_tbl)
-                            col_name = expr.get_name()
-                            if col_name in dim_tbl.columns:
-                                # Direct column — use as-is
-                                dim_cols_to_add.append((dim_name, col_name))
-                            else:
-                                # Derived expression — mutate onto dim table
-                                # Use a temp name to avoid collisions
-                                temp_name = f"__deferred_{short}"
-                                dim_tbl = dim_tbl.mutate(**{temp_name: expr})
-                                dim_cols_to_add.append((dim_name, temp_name))
                         except Exception:
-                            pass
+                            # Derived dims may reference other derived dims;
+                            # materialize dependencies first. Failures beyond
+                            # that raise — silently dropping the requested
+                            # dimension returned unlabeled rows at a hidden
+                            # grain.
+                            dim_tbl = _mutate_dimensions_with_dependencies(
+                                dim_tbl, [short], right_dims
+                            )
+                            dim_cols_to_add.append((dim_name, short))
+                            continue
+                        col_name = expr.get_name()
+                        if col_name in dim_tbl.columns:
+                            # Direct column — use as-is
+                            dim_cols_to_add.append((dim_name, col_name))
+                        else:
+                            # Derived expression — mutate onto dim table
+                            # Use a temp name to avoid collisions
+                            temp_name = f"__deferred_{short}"
+                            dim_tbl = dim_tbl.mutate(**{temp_name: expr})
+                            dim_cols_to_add.append((dim_name, temp_name))
 
             if dim_cols_to_add:
                 # Perform the LEFT JOIN
@@ -3478,7 +3849,10 @@ def _rejoin_one(pt):
             if not common:
                 return pt
 
-            preds = [dim_bridge[c] == pt[c] for c in common]
+            # Null-safe equality: a NULL group key (real NULL dim value, or
+            # minted by the outer join for parents with no children) must
+            # still match its pre-agg row
+            preds = [dim_bridge[c].identical_to(pt[c]) for c in common]
             joined_pt = dim_bridge.left_join(pt, preds).select(
                 [dim_bridge] + [pt[c] for c in pt_meas]
             )
@@ -4738,14 +5112,31 @@ def _build_numeric_index_fragment(
 def _resolve_selector(
     selector: str | list[str] | Callable | None,
     base_tbl: ir.Table,
+    known_fields=frozenset(),
 ) -> tuple[str, ...]:
     if selector is None:
         return tuple(base_tbl.columns)
-    try:
-        selected = base_tbl.select(selector)
-        return tuple(selected.columns)
-    except Exception:
-        return []
+    names = None
+    if isinstance(selector, str):
+        names = [selector]
+    elif isinstance(selector, (list, tuple)) and all(isinstance(n, str) for n in selector):
+        names = list(selector)
+    if names is not None:
+        known = set(known_fields) | set(base_tbl.columns)
+        unknown = [n for n in names if n not in known]
+        if unknown:
+            raise ValueError(
+                f"index() selector matched no dimension or column: {unknown}. "
+                f"Available fields: {sorted(known)}"
+            )
+        return tuple(names)
+    # Callable / ibis selector: let resolution errors propagate loudly — an
+    # empty fallback here made a failing selector index every field.
+    if callable(selector) and not isinstance(selector, s.Selector):
+        resolved = selector(base_tbl)
+        exprs = resolved if isinstance(resolved, (list, tuple)) else [resolved]
+        return tuple(e.get_name() for e in exprs)
+    return tuple(base_tbl.select(selector).columns)
 
 
 def _get_fields_to_index(
@@ -4756,7 +5147,7 @@ def _get_fields_to_index(
     if selector is None:
         selector = s.all()
 
-    raw_fields = _resolve_selector(selector, base_tbl)
+    raw_fields = _resolve_selector(selector, base_tbl, known_fields=merged_dimensions.keys())
 
     if not raw_fields:
         result = list(merged_dimensions.keys())
@@ -5033,7 +5424,17 @@ def _dimension_only_source_table(
                 # columns from other tables we cannot use the shortcut.
                 if filters:
                     tbl = _to_untagged(root)
-                    tbl_cols = frozenset(tbl.columns) | frozenset(root_dims)
+                    # Accept bare and table-prefixed spellings: filters
+                    # written t["customers.region"] (the qualified form
+                    # BSL's own errors recommend) must not silently
+                    # disable the zero-fact-rows guarantee.
+                    tbl_cols = (
+                        frozenset(tbl.columns)
+                        | frozenset(root_dims)
+                        | frozenset(f"{target_prefix}.{d}" for d in root_dims)
+                        | frozenset(f"{target_prefix}.{c}" for c in tbl.columns)
+                    )
+                    resolver = _table_filter_resolver(tbl, root, target_prefix)
                     for flt in filters:
                         fn = _unwrap(flt) if hasattr(flt, "unwrap") else flt
                         # Dict/string filters resolve deferred through the
@@ -5042,7 +5443,7 @@ def _dimension_only_source_table(
                         # risk a wrong source table. See query.Filter.to_callable.
                         if getattr(fn, "__bsl_deferred_resolution__", False):
                             return None
-                        extraction = _extract_columns_from_callable(fn, tbl)
+                        extraction = _extract_columns_from_callable(fn, resolver)
                         if extraction.extraction_failed:
                             return None  # Can't determine — bail out
                         if not extraction.columns <= tbl_cols:
diff --git a/src/boring_semantic_layer/serialization/extract.py b/src/boring_semantic_layer/serialization/extract.py
index b4cd375c..12f9e96e 100644
--- a/src/boring_semantic_layer/serialization/extract.py
+++ b/src/boring_semantic_layer/serialization/extract.py
@@ -101,6 +101,13 @@ def _extract_semantic_table(op, context: BSLSerializationContext) -> dict[str, A
         metadata["name"] = op.name
     if op.description:
         metadata["description"] = op.description
+    # Wrapper tables from SemanticJoin.with_measures()/with_dimensions()
+    # carry the join topology in _source_join. Serializing the wrapper
+    # flat loses it, and the reconstructed model then executes on the
+    # lowered (fanned-out) join — bypassing pre-aggregation entirely.
+    source_join = getattr(op, "_source_join", None)
+    if source_join is not None:
+        metadata["source_join"] = extract_op_tree(source_join, context)
     return metadata
 
 
diff --git a/src/boring_semantic_layer/serialization/reconstruct.py b/src/boring_semantic_layer/serialization/reconstruct.py
index f9aa7a22..b65e7c1b 100644
--- a/src/boring_semantic_layer/serialization/reconstruct.py
+++ b/src/boring_semantic_layer/serialization/reconstruct.py
@@ -138,6 +138,25 @@ def _reconstruct_table():
     measures = {name: _create_measure(name, data) for name, data in meas_meta.items()}
     calc_measures = deserialize_calc_measures(calc_meta) if calc_meta else {}
 
+    # Wrapper tables (join.with_measures()/with_dimensions()) must be
+    # rebuilt AROUND the reconstructed join: without _source_join the
+    # model executes on the lowered fanned-out join and the pre-agg
+    # machinery (fan-out-safe sums, t.all() totals, filter pushdown)
+    # never runs.
+    source_join_meta = context.parse_field(metadata, "source_join")
+    if source_join_meta:
+        join_model = reconstruct_bsl_operation(source_join_meta, xorq_expr, context)
+        join_op = join_model.op() if hasattr(join_model, "op") else join_model
+        return bsl_expr.SemanticModel(
+            table=join_op.to_untagged(),
+            dimensions=dimensions,
+            measures=measures,
+            calc_measures=calc_measures,
+            name=metadata.get("name"),
+            description=metadata.get("description"),
+            _source_join=join_op,
+        )
+
     return bsl_expr.SemanticModel(
         table=_reconstruct_table(),
         dimensions=dimensions,
@@ -252,6 +271,40 @@ def _reconstruct_limit(
     return source.limit(n=int(metadata.get("n", 0)), offset=int(metadata.get("offset", 0)))
 
 
+def _validate_join_leaf(model, metadata, side: str) -> None:
+    """Check a reconstructed join leaf against its declared fields.
+
+    Only missing-column failures (AttributeError/KeyError) are treated as
+    misassignment — other resolution errors (e.g. measures that need an
+    unnest context) are not evidence the table is wrong.
+    """
+    from .. import ops as bsl_ops
+
+    op = model.op() if hasattr(model, "op") else model
+    if not isinstance(op, bsl_ops.SemanticTableOp):
+        return
+    try:
+        tbl = op.table.to_expr() if hasattr(op.table, "to_expr") else op.table
+    except Exception:
+        return
+    name = metadata.get("name") or side
+    for kind, fields in (("dimension", op.get_dimensions()), ("measure", op.get_measures())):
+        for fname, fn in fields.items():
+            try:
+                fn(tbl)
+            except (AttributeError, KeyError) as exc:
+                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."
+                ) from exc
+            except Exception:
+                continue
+
+
 @register_reconstructor("SemanticJoinOp")
 def _reconstruct_join(
     metadata: dict, xorq_expr, source, context: BSLSerializationContext
@@ -277,6 +330,15 @@ def _reconstruct_join(
     left_model = reconstruct_bsl_operation(left_metadata, left_xorq_expr, context)
     right_model = reconstruct_bsl_operation(right_metadata, right_xorq_expr, context)
 
+    # Guard against leaf misassignment: expressions lowered through the
+    # pre-agg path put partial-aggregate/key-bridge joins where the raw
+    # join used to be, so _split_join_expr can hand back the wrong table
+    # for a side. When shapes happen to align this silently returns wrong
+    # numbers — validate that each leaf's declared fields resolve against
+    # its recovered table and raise otherwise.
+    _validate_join_leaf(left_model, left_metadata, "left")
+    _validate_join_leaf(right_model, right_metadata, "right")
+
     how = metadata.get("how", "inner")
     # Default to "many" for payloads serialized before cardinality was
     # emitted — join_many is a safe superset of join_one behaviour, while
diff --git a/src/boring_semantic_layer/tests/test_soundness_round2.py b/src/boring_semantic_layer/tests/test_soundness_round2.py
new file mode 100644
index 00000000..c1a0596d
--- /dev/null
+++ b/src/boring_semantic_layer/tests/test_soundness_round2.py
@@ -0,0 +1,562 @@
+"""Regression tests for the July 2026 round-2 soundness evaluation.
+
+Each test pins a confirmed silent-wrong-answer defect (or its loud-error
+replacement) against pandas-derived ground truth. Finding IDs reference
+the round-2 soundness report:
+
+- A1/A2  non-decomposable measures re-aggregated with SUM
+- A3     mean(where=...) losing its condition under pre-aggregation
+- B1     bare derived-dim filters restricting only the owning table
+- B2     cross-table compound predicates inflating many-side measures
+- B3     group_by().filter().aggregate() discarding the grouping
+- B4     derived dims as group keys missing from pre-agg output
+- C1     NULL group keys dropped by plain equi-joins in the re-join
+- D1     join_one default join type depending on the receiver class
+- E1/E2  serialization round-trips changing results
+- F1     deferred-join path dropping dims / returning hidden grain
+- F2     dimension-only shortcut disabled by prefixed filter spelling
+- F3     index() selector typos silently indexing every field
+"""
+
+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 with an uneven 1:N fan into items (3/1/2 line items)."""
+    orders = con.create_table(
+        "orders",
+        pd.DataFrame(
+            {
+                "order_id": [1, 2, 3],
+                "customer_id": [10, 10, 20],
+                "status": ["open", "closed", "open"],
+                "amount": [100.0, 120.0, 80.0],
+            }
+        ),
+    )
+    items = con.create_table(
+        "items",
+        pd.DataFrame(
+            {
+                "item_id": [1, 2, 3, 4, 5, 6],
+                "order_id": [1, 1, 1, 2, 3, 3],
+                "qty": [1, 2, 1, 3, 1, 1],
+                "sku": ["a", "b", "a", "c", "a", "b"],
+            }
+        ),
+    )
+    o_st = (
+        to_semantic_table(orders, name="orders")
+        .with_dimensions(
+            customer_id=lambda t: t.customer_id,
+            status=lambda t: t.status,
+            size=lambda t: (t.amount > 90).ifelse("big", "small"),
+        )
+        .with_measures(
+            total_amount=lambda t: t.amount.sum(),
+            avg_amount=lambda t: t.amount.mean(),
+            avg_open_amount=lambda t: t.amount.mean(where=t.status == "open"),
+            median_amount=lambda t: t.amount.median(),
+            aov=lambda t: t.amount.sum() / t.count(),
+        )
+    )
+    i_st = (
+        to_semantic_table(items, name="items")
+        .with_dimensions(sku=lambda t: t.sku)
+        .with_measures(
+            item_count=lambda t: t.count(),
+            total_qty=lambda t: t.qty.sum(),
+        )
+    )
+    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 TestNonDecomposableReagg:
+    """A1/A2: median/stddev/ratios must not be summed at cross-table grain."""
+
+    def test_median_by_cross_table_dim(self, orders_items):
+        df = (
+            _joined(orders_items)
+            .group_by("items.sku")
+            .aggregate("orders.median_amount")
+            .execute()
+            .set_index("items.sku")
+        )
+        # sku a touches orders 1 (100) and 3 (80) -> median 90, not 180
+        assert df.loc["a", "orders.median_amount"] == pytest.approx(90.0)
+        assert df.loc["c", "orders.median_amount"] == pytest.approx(120.0)
+
+    def test_ratio_measure_by_cross_table_dim(self, orders_items):
+        df = (
+            _joined(orders_items)
+            .group_by("items.sku")
+            .aggregate("orders.aov")
+            .execute()
+            .set_index("items.sku")
+        )
+        # sku a: (100 + 80) / 2 = 90, not the summed per-order ratios (180)
+        assert df.loc["a", "orders.aov"] == pytest.approx(90.0)
+
+    def test_items_median_by_orders_dim(self, con):
+        orders = con.create_table(
+            "orders_m",
+            pd.DataFrame({"order_id": [1, 2], "band": ["A", "A"]}),
+        )
+        items = con.create_table(
+            "items_m",
+            pd.DataFrame(
+                {"order_id": [1, 1, 1, 2, 2], "qty": [1, 5, 9, 1, 3]}
+            ),
+        )
+        o = to_semantic_table(orders, name="orders").with_dimensions(
+            band=lambda t: t.band
+        )
+        i = to_semantic_table(items, name="items").with_measures(
+            median_qty=lambda t: t.qty.median(),
+            avg_qty=lambda t: t.qty.mean(),
+        )
+        df = (
+            o.join_many(i, lambda a, b: a.order_id == b.order_id)
+            .group_by("orders.band")
+            .aggregate("items.median_qty", "items.avg_qty")
+            .execute()
+        )
+        # median over item rows in band A = 3, not sum of per-order medians (5+2)
+        assert df["items.median_qty"].iloc[0] == pytest.approx(3.0)
+        assert df["items.avg_qty"].iloc[0] == pytest.approx(19 / 5)
+
+
+class TestConditionalMeanDecomposition:
+    """A3: mean(where=...) keeps its condition on joined grains."""
+
+    def test_scalar_grain(self, orders_items):
+        df = _joined(orders_items).aggregate("orders.avg_open_amount").execute()
+        # mean of open orders [100, 80] = 90, not mean of all three (100)
+        assert df["orders.avg_open_amount"].iloc[0] == pytest.approx(90.0)
+
+    def test_grouped_grain(self, orders_items):
+        df = (
+            _joined(orders_items)
+            .group_by("orders.customer_id")
+            .aggregate("orders.avg_open_amount")
+            .order_by("orders.customer_id")
+            .execute()
+        )
+        # customer 10's only open order is 100 (closed 120 excluded)
+        assert df["orders.avg_open_amount"].tolist() == pytest.approx([100.0, 80.0])
+
+
+class TestNullGroupKeys:
+    """C1: NULL group keys keep their measures, independent of measure order."""
+
+    @pytest.fixture
+    def null_qty(self, con):
+        orders = con.create_table(
+            "orders_n",
+            pd.DataFrame(
+                {
+                    "order_id": [1, 2, 3, 4],
+                    "customer_id": [10, 10, 20, 30],
+                    "amount": [100.0, 120.0, 80.0, 50.0],
+                }
+            ),
+        )
+        items = con.create_table(
+            "items_n",
+            pd.DataFrame(
+                {
+                    "item_id": [1, 2, 3, 4, 5, 6],
+                    "order_id": [1, 1, 1, 2, 3, 3],
+                    "qty": [1.0, 2.0, 1.0, 3.0, None, 1.0],
+                }
+            ),
+        )
+        o = (
+            to_semantic_table(orders, name="orders")
+            .with_dimensions(customer_id=lambda t: t.customer_id)
+            .with_measures(total_amount=lambda t: t.amount.sum())
+        )
+        i = (
+            to_semantic_table(items, name="items")
+            .with_dimensions(qty=lambda t: t.qty)
+            .with_measures(item_count=lambda t: t.count())
+        )
+        return o.join_many(i, lambda a, b: a.order_id == b.order_id)
+
+    @staticmethod
+    def _null_row(df):
+        return df[df["items.qty"].isna()].iloc[0]
+
+    def test_null_group_measures_present(self, null_qty):
+        # Both a real NULL qty (order 3) and the no-item order 4 land in
+        # the NULL group: item_count 1, distinct-order amount 80 + 50.
+        df = null_qty.group_by("items.qty").aggregate(
+            "items.item_count", "orders.total_amount"
+        ).execute()
+        row = self._null_row(df)
+        assert row["items.item_count"] == 1
+        assert row["orders.total_amount"] == pytest.approx(130.0)
+
+    def test_measure_order_does_not_change_answer(self, null_qty):
+        df1 = null_qty.group_by("items.qty").aggregate(
+            "items.item_count", "orders.total_amount"
+        ).execute()
+        df2 = null_qty.group_by("items.qty").aggregate(
+            "orders.total_amount", "items.item_count"
+        ).execute()
+        r1, r2 = self._null_row(df1), self._null_row(df2)
+        assert r1["items.item_count"] == r2["items.item_count"]
+        assert r1["orders.total_amount"] == r2["orders.total_amount"]
+
+
+class TestJoinOneDefaults:
+    """D1: join_one defaults to how="left" on every receiver class."""
+
+    def test_noop_filter_does_not_change_totals(self, con):
+        orders = con.create_table(
+            "orders_d",
+            pd.DataFrame(
+                {"order_id": [1, 2, 3], "customer_id": [10, 10, 20], "amount": [100, 120, 80]}
+            ),
+        )
+        customers = con.create_table(
+            "customers_d",
+            pd.DataFrame({"customer_id": [10], "region": ["west"]}),
+        )
+        o = (
+            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 = to_semantic_table(customers, name="customers").with_dimensions(
+            customer_id=lambda t: t.customer_id, region=lambda t: t.region
+        )
+        on = lambda a, b: a.customer_id == b.customer_id  # noqa: E731
+        direct = o.join_one(c, on).aggregate("orders.total_amount").execute()
+        filtered = (
+            o.filter(lambda t: t.amount > 0)
+            .join_one(c, on)
+            .aggregate("orders.total_amount")
+            .execute()
+        )
+        assert (
+            direct["orders.total_amount"].iloc[0]
+            == filtered["orders.total_amount"].iloc[0]
+            == 300
+        )
+
+
+class TestFilterRouting:
+    """B1/B2: derived-dim and compound filters restrict every table."""
+
+    def test_bare_derived_dim_filter_matches_prefixed(self, orders_items):
+        joined = _joined(orders_items)
+        bare = (
+            joined.filter(lambda t: t.size == "big")
+            .group_by("orders.customer_id")
+            .aggregate("orders.total_amount", "items.item_count")
+            .order_by("orders.customer_id")
+            .execute()
+        )
+        prefixed = (
+            joined.filter(lambda t: t["orders.size"] == "big")
+            .group_by("orders.customer_id")
+            .aggregate("orders.total_amount", "items.item_count")
+            .order_by("orders.customer_id")
+            .execute()
+        )
+        pd.testing.assert_frame_equal(bare, prefixed)
+        # big orders: 1 (100) and 2 (120), both customer 10 — no ghost rows
+        assert bare["orders.customer_id"].tolist() == [10]
+        assert bare["orders.total_amount"].tolist() == [220.0]
+        assert bare["items.item_count"].tolist() == [4]
+
+    def test_compound_and_matches_chained_filters(self, orders_items):
+        joined = _joined(orders_items)
+        compound = (
+            joined.filter(lambda t: (t["orders.status"] == "open") & (t.qty >= 2))
+            .group_by("orders.customer_id")
+            .aggregate("items.item_count", "items.total_qty")
+            .order_by("orders.customer_id")
+            .execute()
+        )
+        chained = (
+            joined.filter(lambda t: t["orders.status"] == "open")
+            .filter(lambda t: t.qty >= 2)
+            .group_by("orders.customer_id")
+            .aggregate("items.item_count", "items.total_qty")
+            .order_by("orders.customer_id")
+            .execute()
+        )
+        pd.testing.assert_frame_equal(compound, chained)
+        # open orders 1 & 3; items with qty>=2: item 2 (order 1) only
+        assert compound["items.item_count"].tolist() == [1]
+
+    def test_cross_table_or_raises(self, orders_items):
+        expr = (
+            _joined(orders_items)
+            .filter(lambda t: (t["orders.status"] == "closed") | (t.qty >= 5))
+            .aggregate("items.item_count")
+        )
+        with pytest.raises(ValueError, match="row-precisely"):
+            expr.execute()
+
+
+class TestGroupByFilterAggregate:
+    """B3: the grouping survives a filter between group_by and aggregate."""
+
+    def test_keys_preserved(self, orders_items):
+        df = (
+            _joined(orders_items)
+            .group_by("orders.customer_id")
+            .filter(lambda t: t["orders.status"] == "open")
+            .aggregate("orders.total_amount", "items.item_count")
+            .order_by("orders.customer_id")
+            .execute()
+        )
+        assert "orders.customer_id" in df.columns
+        assert df["orders.customer_id"].tolist() == [10, 20]
+        assert df["orders.total_amount"].tolist() == [100.0, 80.0]
+        assert df["items.item_count"].tolist() == [3, 2]
+
+
+class TestDerivedDimGroupKeys:
+    """B4: derived dims as group keys appear in the pre-agg output."""
+
+    def test_key_column_present(self, orders_items):
+        df = (
+            _joined(orders_items)
+            .group_by("orders.size")
+            .aggregate("orders.total_amount")
+            .execute()
+            .set_index("orders.size")
+        )
+        assert df.loc["big", "orders.total_amount"] == pytest.approx(220.0)
+        assert df.loc["small", "orders.total_amount"] == pytest.approx(80.0)
+
+    def test_with_many_side_measure(self, orders_items):
+        df = (
+            _joined(orders_items)
+            .group_by("orders.size")
+            .aggregate("orders.total_amount", "items.item_count")
+            .execute()
+            .set_index("orders.size")
+        )
+        assert df.loc["big", "items.item_count"] == 4
+        assert df.loc["small", "items.item_count"] == 2
+
+
+class TestDeferredDimensionJoins:
+    """F1: deferral keeps the requested grain and never drops dims."""
+
+    @pytest.fixture
+    def entity_join(self, con):
+        orders = con.create_table(
+            "orders_e",
+            pd.DataFrame(
+                {
+                    "order_id": [1, 2, 3, 4],
+                    "customer_id": [10, 10, 20, 30],
+                    "amount": [100.0, 120.0, 80.0, 50.0],
+                }
+            ),
+        )
+        customers = con.create_table(
+            "customers_e",
+            pd.DataFrame({"customer_id": [10, 20, 30], "region": ["east", "west", "east"]}),
+        )
+        o = (
+            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 = to_semantic_table(customers, name="customers").with_dimensions(
+            customer_id={"expr": lambda t: t.customer_id, "is_entity": True},
+            region=lambda t: t.region,
+            region_band=lambda t: t.region.upper(),
+            region_display=lambda t: t.region_band + "!",
+        )
+        return o.join_one(c, lambda l, r: l.customer_id == r.customer_id)
+
+    def test_coarser_dim_grain_is_regrouped(self, entity_join):
+        df = (
+            entity_join.group_by("customers.region")
+            .aggregate("orders.total_amount")
+            .execute()
+            .set_index("customers.region")
+        )
+        # 2 region rows, not 3 entity rows with duplicate labels
+        assert len(df) == 2
+        assert df.loc["east", "orders.total_amount"] == pytest.approx(270.0)
+        assert df.loc["west", "orders.total_amount"] == pytest.approx(80.0)
+
+    def test_derived_of_derived_dim_resolves(self, entity_join):
+        df = (
+            entity_join.group_by("customers.region_display")
+            .aggregate("orders.total_amount")
+            .execute()
+            .set_index("customers.region_display")
+        )
+        assert "EAST!" in df.index and "WEST!" in df.index
+        assert df.loc["EAST!", "orders.total_amount"] == pytest.approx(270.0)
+
+    def test_entity_grain_deferral_still_works(self, entity_join):
+        df = (
+            entity_join.group_by("customers.customer_id", "customers.region")
+            .aggregate("orders.total_amount")
+            .order_by("customers.customer_id")
+            .execute()
+        )
+        assert df["orders.total_amount"].tolist() == pytest.approx([220.0, 80.0, 50.0])
+        assert df["customers.region"].tolist() == ["east", "west", "east"]
+
+
+class TestDimensionOnlyShortcut:
+    """F2: zero-fact members survive every filter spelling."""
+
+    @pytest.fixture
+    def dim_join(self, con):
+        orders = con.create_table(
+            "orders_s",
+            pd.DataFrame({"order_id": [1, 2], "customer_id": [10, 20], "amount": [1, 2]}),
+        )
+        customers = con.create_table(
+            "customers_s",
+            pd.DataFrame(
+                {"customer_id": [10, 20, 30], "region": ["west", "east", "north"]}
+            ),
+        )
+        o = (
+            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 = to_semantic_table(customers, name="customers").with_dimensions(
+            customer_id=lambda t: t.customer_id,
+            region=lambda t: t.region,
+            region_band=lambda t: t.region.upper(),
+        )
+        return o.join_one(c, lambda l, r: l.customer_id == r.customer_id)
+
+    def test_prefixed_filter_keeps_zero_fact_members(self, dim_join):
+        df = (
+            dim_join.filter(lambda t: t["customers.region"] != "zzz")
+            .group_by("customers.region")
+            .aggregate()
+            .execute()
+        )
+        # 'north' has no orders but must still be returned (#224)
+        assert sorted(df["customers.region"]) == ["east", "north", "west"]
+
+    def test_derived_dim_filter_keeps_zero_fact_members(self, dim_join):
+        df = (
+            dim_join.filter(lambda t: t.region_band != "ZZZ")
+            .group_by("customers.region")
+            .aggregate()
+            .execute()
+        )
+        assert sorted(df["customers.region"]) == ["east", "north", "west"]
+
+
+class TestIndexSelector:
+    """F3: a selector that matches nothing raises instead of matching all."""
+
+    @pytest.fixture
+    def flat(self, con):
+        tbl = con.create_table(
+            "flat_i",
+            pd.DataFrame({"status": ["a", "b"], "region": ["x", "y"], "amount": [1, 2]}),
+        )
+        return to_semantic_table(tbl, name="flat").with_dimensions(
+            status=lambda t: t.status, region=lambda t: t.region
+        )
+
+    def test_typo_raises(self, flat):
+        with pytest.raises(ValueError, match="staus"):
+            flat.index("staus").execute()
+
+    def test_exact_field_indexes_only_that_field(self, flat):
+        df = flat.index("status").execute()
+        assert set(df["fieldName"]) == {"status"}
+
+
+class TestSerializationRoundTrip:
+    """E1/E2: round-trips preserve results or fail loudly."""
+
+    @pytest.fixture(autouse=True)
+    def _requires_xorq(self):
+        pytest.importorskip("xorq")
+
+    def _wrapper_model(self, con):
+        orders = con.create_table(
+            "orders_rt",
+            pd.DataFrame(
+                {
+                    "order_id": [1, 2, 3],
+                    "customer_id": [10, 10, 20],
+                    "amount": [100.0, 120.0, 80.0],
+                }
+            ),
+        )
+        items = con.create_table(
+            "items_rt",
+            pd.DataFrame(
+                {"item_id": [1, 2, 3, 4, 5, 6], "order_id": [1, 1, 1, 2, 3, 3], "qty": [1] * 6}
+            ),
+        )
+        o = (
+            to_semantic_table(orders, name="orders")
+            .with_dimensions(customer_id=lambda t: t.customer_id)
+            .with_measures(
+                total_amount=lambda t: t.amount.sum(),
+                avg_amount=lambda t: t.amount.mean(),
+            )
+        )
+        i = to_semantic_table(items, name="items").with_measures(
+            item_count=lambda t: t.count()
+        )
+        return o.join_many(i, lambda a, b: a.order_id == b.order_id).with_measures(
+            pot=lambda t: t["orders.total_amount"] / t.all(t["orders.total_amount"]),
+        )
+
+    def test_wrapper_roundtrip_keeps_preagg(self, con):
+        from boring_semantic_layer.serialization import from_tagged, to_tagged
+
+        model = self._wrapper_model(con)
+        query = lambda m: (  # noqa: E731
+            m.group_by("orders.customer_id")
+            .aggregate("orders.total_amount", "orders.avg_amount", "pot")
+            .order_by("orders.customer_id")
+            .execute()
+        )
+        before = query(model)
+        restored = from_tagged(to_tagged(model))
+        after = query(restored)
+        pd.testing.assert_frame_equal(before, after)
+        # Fan-out-safe numbers, not the lowered-join ones (420/160)
+        assert after["orders.total_amount"].tolist() == pytest.approx([220.0, 80.0])
+        assert after["pot"].sum() == pytest.approx(1.0)
+
+    def test_preagg_query_roundtrip_raises_not_wrong(self, con):
+        from boring_semantic_layer.serialization import from_tagged, to_tagged
+
+        model = self._wrapper_model(con)
+        expr = model.filter(lambda t: t.qty >= 1).aggregate("items.item_count")
+        tagged = to_tagged(expr)
+        with pytest.raises(ValueError, match="could not recover"):
+            from_tagged(tagged).execute()