fix(DEV-1780): bind or reject dotted dimension join paths

slayer/core/errors.py

@@ -245,3 +245,48 @@ def __init__(self, *, column: str, qualifier: str) -> None:
             f"Project it (add to dimensions/measures), reference it in a filter, or "
             f"order by a projected field instead."
         )
+
+
+class UnresolvableDimensionJoinError(SlayerError, ValueError):
+    """A dimension / time-dimension dotted path that is not a valid direct-join
+    chain and cannot be uniquely routed to its target model (DEV-1780).
+
+    A dotted path resolves only when every hop is a direct join. A short form
+    (``Consumer.name`` — target model only) auto-resolves when exactly one route
+    reaches the target; otherwise (ambiguous route, unreachable target, or an
+    explicit multi-hop chain with a broken hop) the reference is rejected here
+    rather than emitting SQL that references an unbound table alias.
+
+    Multi-inherits ``ValueError`` (like ``UnresolvableOrderColumnError``) so
+    existing ``except ValueError`` sites keep working. ``__str__`` is computed
+    from the fields so ``suggested_path`` set after construction is reflected.
+    """
+
+    def __init__(
+        self,
+        *,
+        reference: str,
+        root_model: str,
+        reason: str | None = None,
+        available_joins: "list[str] | None" = None,
+        suggested_path: str | None = None,
+    ) -> None:
+        self.reference = reference
+        self.root_model = root_model
+        self.reason = reason
+        self.available_joins = available_joins
+        self.suggested_path = suggested_path
+        super().__init__()
+
+    def __str__(self) -> str:
+        msg = (
+            f"Cannot resolve dimension '{self.reference}': not a valid join path "
+            f"from '{self.root_model}'."
+        )
+        if self.reason:
+            msg += f" {self.reason}"
+        if self.available_joins is not None:
+            msg += f" Available joins from '{self.root_model}': {sorted(self.available_joins)}."
+        if self.suggested_path:
+            msg += f" Did you mean '{self.suggested_path}'?"
+        return msg

slayer/engine/enrichment.py

@@ -37,6 +37,7 @@
     parse_filter,
     parse_formula,
 )
+from slayer.core.errors import UnresolvableDimensionJoinError
 from slayer.core.models import Column, SlayerModel
 from slayer.core.query import OrderItem, SlayerQuery, substitute_variables
 from slayer.core.refs import DOTTED_IDENT_REF_RE as _DOTTED_IDENT_REF_RE
@@ -180,6 +181,7 @@ async def enrich_query(
     resolve_model=None,
     dialect: str = "postgres",
     drop_unreachable_filters: bool = False,
+    enforce_join_binding: bool = True,
 ) -> EnrichedQuery:
     """Resolve a SlayerQuery against model definitions into an EnrichedQuery.
 
@@ -345,6 +347,35 @@ def _mark_user_declared(alias: str) -> bool:
                 return True
         return False
 
+    def _repoint_alias(prev_alias: str, new_alias: str) -> None:
+        """DEV-1779: repoint every reference to ``prev_alias`` onto ``new_alias``.
+
+        A formula/transform enriched before the sibling measure it references
+        freezes that sibling's canonical alias (``orders.id_count``) into its
+        expression SQL / transform input. When the sibling is later renamed to
+        its declared name (``orders.order_count``), follow the rename in every
+        carrier: the alias resolver, the provenance-merge index, and the
+        already-frozen ``EnrichedExpression.sql`` / ``EnrichedTransform``.
+        """
+        if prev_alias == new_alias:
+            return
+        for k, v in known_aliases.items():
+            if v == prev_alias:
+                known_aliases[k] = new_alias
+        for k, v in measure_canonical_key_to_alias.items():
+            if v == prev_alias:
+                measure_canonical_key_to_alias[k] = new_alias
+        # Aliases are emitted only as whole quoted identifiers, so matching the
+        # closing quote is exact: ``"orders.id_count"`` never matches the
+        # prefix of ``"orders.id_count_2"``.
+        quoted_prev, quoted_new = f'"{prev_alias}"', f'"{new_alias}"'
+        for e in enriched_expressions:
+            if quoted_prev in e.sql:
+                e.sql = e.sql.replace(quoted_prev, quoted_new)
+        for t in enriched_transforms:
+            if t.measure_alias == prev_alias:
+                t.measure_alias = new_alias
+
     async def _ensure_aggregated_measure(
         alias_key: str,
         measure_name: str,
@@ -1490,12 +1521,11 @@ def _mangled_formula(formula: str) -> str:
                                 break
                         known_aliases[target_name] = target_alias
                         known_aliases[canonical_name] = target_alias
-                        # DEV-1444 provenance merge: any canonical key
-                        # currently pointing at the pre-rename alias must
-                        # follow the rename.
-                        for k, v in list(measure_canonical_key_to_alias.items()):
-                            if v == prev_alias:
-                                measure_canonical_key_to_alias[k] = target_alias
+                        # DEV-1444 provenance merge + DEV-1779 frozen-carrier
+                        # rewrite: repoint resolver / provenance entries AND
+                        # any expression/transform that already froze the
+                        # pre-rename intercept alias onto the new alias.
+                        _repoint_alias(prev_alias, target_alias)
                         # canonical_to_user_name only fires when the
                         # user explicitly renamed via qfield.name; the
                         # auto-rename to cross-model canonical doesn't
@@ -1655,12 +1685,12 @@ def _mangled_formula(formula: str) -> str:
                         break
                 known_aliases[qfield.name] = user_alias
                 known_aliases[canonical_name] = user_alias
-                # DEV-1444 provenance merge: any canonical key currently
-                # pointing at the pre-rename alias must follow the rename
-                # so later auto-extracted refs collapse onto the new alias.
-                for k, v in list(measure_canonical_key_to_alias.items()):
-                    if v == prev_alias:
-                        measure_canonical_key_to_alias[k] = user_alias
+                # DEV-1444 provenance merge + DEV-1779 frozen-carrier rewrite:
+                # any resolver / provenance entry pointing at the pre-rename
+                # alias must follow the rename, AND any expression/transform
+                # that already froze the pre-rename alias must be rewritten so
+                # a formula enriched before this measure doesn't dangle.
+                _repoint_alias(prev_alias, user_alias)
                 # DEV-1443: record the canonical → user-name mapping so
                 # query filters and ORDER BY items referencing the raw
                 # ``col:agg`` formula can be remapped to the user alias
@@ -1878,6 +1908,24 @@ def _mangled_formula(formula: str) -> str:
         dialect=dialect,
     )
 
+    # DEV-1780 safety net: never return a dim/time-dim whose join-path alias is
+    # absent from resolved_joins (it would render an unbound ``A__B`` reference).
+    # Skipped for virtual stages and the re-rooted CTE (enforce_join_binding=False).
+    if enforce_join_binding and model.source_model_origin is None:
+        _bound_aliases = {rj[1] for rj in resolved_joins}
+        _root_prefix = f"{model_name_str}."
+        for _bound_check in list(dimensions) + list(time_dimensions):
+            _mn = _bound_check.model_name
+            if _mn != model_name_str and _mn not in _bound_aliases:
+                _reference = _bound_check.alias
+                if _reference.startswith(_root_prefix):
+                    _reference = _reference[len(_root_prefix):]
+                raise UnresolvableDimensionJoinError(
+                    reference=_reference,
+                    root_model=model_name_str,
+                    available_joins=[j.target_model for j in model.joins],
+                )
+
     # Names that resolve at the query level (named measures, transforms,
     # expressions) — pass through as legitimate filter targets even though
     # they are not Columns / ModelMeasures on the source model.

slayer/engine/join_graph.py

@@ -60,6 +60,61 @@ def reachable_from(self, root: str) -> set[str]:
                     frontier.append(nbr)
         return seen
 
+    def _reverse_reachable_to(self, target: str) -> set[str]:
+        """Nodes that can reach ``target`` via directed edges (incl. ``target``).
+        Used to prune ``count_simple_paths`` to the relevant subgraph."""
+        reverse: dict[str, set[str]] = {}
+        for src, nbrs in self._adj.items():
+            for nbr in nbrs:
+                reverse.setdefault(nbr, set()).add(src)
+        seen: set[str] = {target}
+        frontier: deque[str] = deque([target])
+        while frontier:
+            node = frontier.popleft()
+            for pred in reverse.get(node, ()):  # noqa: SIM118 — .get default
+                if pred not in seen:
+                    seen.add(pred)
+                    frontier.append(pred)
+        return seen
+
+    def count_simple_paths(self, root: str, target: str, *, cap: int = 2) -> int:
+        """Number of distinct simple (acyclic) directed paths ``root → target``,
+        capped at ``cap`` with early-stop.
+
+        ``0`` = unreachable, ``1`` = unique route, ``>= cap`` = ambiguous. Counts
+        ALL simple paths, not just shortest ones: a 2-hop plus a 3-hop route to
+        the same target is genuinely ambiguous (auto-picking the shorter would
+        silently change join semantics). The DFS is confined to nodes that can
+        still reach ``target`` (reverse-reachability prune) and iterates
+        adjacency in sorted order; the visited set keeps it finite on cyclic
+        (symmetric INNER) graphs. ``root == target`` returns ``1`` (trivial
+        empty route)."""
+        if root == target:
+            return 1
+        relevant = self._reverse_reachable_to(target)
+        if root not in relevant:
+            return 0
+
+        count = 0
+        visited: set[str] = {root}
+
+        def dfs(node: str) -> None:
+            nonlocal count
+            for nbr in sorted(self._adj.get(node, ())):  # noqa: SIM118 — .get default
+                if count >= cap:
+                    return
+                if nbr == target:
+                    count += 1
+                    continue
+                if nbr in visited or nbr not in relevant:
+                    continue
+                visited.add(nbr)
+                dfs(nbr)
+                visited.discard(nbr)
+
+        dfs(root)
+        return min(count, cap)
+
     def shortest_path(self, root: str, target: str) -> list[str] | None:
         """Return the hop-name sequence from ``root`` to ``target``
         (excluding ``root``), or ``None`` if unreachable.