feat(relations): cross-family coercion for derived DAG dependency refs

src/mountainash/relations/dag/dag.py

@@ -4,11 +4,18 @@
 from typing import Any, Optional, TYPE_CHECKING
 
 from mountainash.core.constants import CONST_BACKEND
-from mountainash.relations.core.relation_nodes import ReadRelNode
 from mountainash.relations.core.relation_nodes.extensions_mountainash import RefRelNode
 from mountainash.relations.dag.traversal import walk_refs as _walk_refs
 
 
+def _consumer_prototype(family: CONST_BACKEND, dialect: str | None) -> Any:
+    """Alias for :func:`_anchor_prototype` (the consumer-side prototype
+    ``_coerce_to_match`` needs as its target object). Same function, renamed
+    at the call site so item 97's resolver-time coercion reads naturally
+    without rewriting item 92's territory."""
+    return _anchor_prototype(family, dialect)
+
+
 def _anchor_prototype(family: CONST_BACKEND, dialect: str | None) -> Any:
     """A lightweight empty object of *family* for coercion to target."""
     if family is CONST_BACKEND.POLARS:
@@ -398,24 +405,47 @@ def _compile_with_refs(
         expression_system = expression_system_cls(dialect=dialect)
         expr_visitor = UnifiedExpressionVisitor(expression_system)
 
+        # Item 97 (inherits item 92's Revision 4 upfront guard, broadened to
+        # ANY foreign-family ref -- bare or derived -- not just a bare
+        # ReadRelNode root): a lazy narwhals ANCHOR consuming foreign refs
+        # must reject before caching -- _coerce_to_match's eager-over-lazy
+        # handling is order-dependent and must not be relied upon.
         if backend_target_name is not None or ref_names:
             anchor_name = backend_target_name or sorted(ref_names)[0]
             anchor_family, _, anchor_leaf = self._resolve_identity_leaf(anchor_name)
             if anchor_leaf is not None and _is_lazy_narwhals(anchor_leaf.dataframe):
                 if any(
                     self._resolve_actual_identity_for(n)[0]
                     not in (None, resolved_backend)
-                    and isinstance(getattr(self.relations[n], "_node", None), ReadRelNode)
                     for n in ref_names
                 ):
                     raise TypeError(
                         "Cross-family DAG coercion is not supported with a lazy Narwhals anchor."
                     )
 
-        cache: dict[str, Any] = {}
+        canonical: dict[str, "tuple[Any, CONST_BACKEND | None, str | None]"] = {}
+        coerced: dict[tuple[str, CONST_BACKEND, str | None], Any] = {}
 
         def resolver(n: str) -> Any:
-            return cache[n]
+            value, src_family, src_dialect = canonical[n]
+            if src_family is None:
+                return value  # no-leaf ref: already anchor-family
+            cons_family = visitor.backend.backend_type
+            cons_dialect = visitor.backend.dialect
+            needs_coercion = (
+                src_family != cons_family
+                or (
+                    src_family is CONST_BACKEND.NARWHALS
+                    and src_dialect != cons_dialect
+                )
+            )
+            if not needs_coercion:
+                return value
+            key = (n, cons_family, cons_dialect)
+            if key not in coerced:
+                proto = _consumer_prototype(cons_family, cons_dialect)
+                coerced[key] = UnifiedRelationVisitor._coerce_to_match(proto, value)
+            return coerced[key]
 
         # KeyDriftContext (item 48 PR-D): +1 optional visitor param,
         # analogous to ref_resolver. The context's resource_name is the
@@ -464,7 +494,15 @@ def resolver(n: str) -> Any:
         if ref_names:
             from mountainash.relations.dag.key_context import KeyDriftContext
 
-            # Get full topological order and filter to only the needed refs
+            # Item 97: canonical materialization -- every ref is compiled
+            # exactly once, with ITS OWN (family, dialect) identity, and
+            # stored as (value, family, dialect) in `canonical`. A consumer
+            # of that ref coerces it lazily via `resolver()` above (memoised
+            # per (name, consumer_family, consumer_dialect)). This replaces
+            # item 89's four-way branch (anchor-for-foreign-bare-read /
+            # own-dialect / reuse-anchor) with a three-way branch that still
+            # reuses the anchor's objects for the same-family-same-dialect
+            # case (item 89's zero-cost homogeneous-DAG path).
             full_order = self.topological_order(target=None)
             for n in full_order:
                 if n not in ref_names:
@@ -474,78 +512,50 @@ def resolver(n: str) -> Any:
                 if root is None:
                     raise ValueError(f"relation {n!r} has no _node attribute")
 
-                # Item 89: give each dependency its OWN physical
-                # (family, dialect) identity for the duration of ITS OWN
-                # root.accept(visitor) -- mirrors the key_context per-ref
-                # swap immediately below. Without this, every dependency
-                # was gated/enriched against the ANCHOR's dialect
-                # regardless of its own, silently leaking a raw native
-                # exception whenever a dialect-scoped CapabilityFact
-                # (BUILD-time GATE or MATERIALIZE_RESIDUE) belonged to the
-                # ref's own dialect, not the anchor's.
                 ref_family, ref_dialect = self._resolve_actual_identity_for(n)
 
+                # Each dependency is key-assessed against ITS OWN
+                # constraints, unconditionally -- including a no-leaf
+                # SourceRelNode ref -- independent of whether the target
+                # itself has a key identity.
+                visitor.key_context = KeyDriftContext(
+                    resource_name=n,
+                    constraints_for=self.constraints_for,
+                    schema_of=self.schema,
+                )
+
                 if ref_family is None:
                     # No physical read identity (pure SourceRelNode/inline-
-                    # data ref). Nothing to compare against -- inherit the
-                    # anchor unconditionally.
+                    # data ref). Materialise with the anchor pair.
                     visitor.backend, visitor.expr_visitor = (
                         relation_system,
                         expr_visitor,
                     )
-                elif ref_family != resolved_backend:
-                    if isinstance(root, ReadRelNode):
-                        anchor_proto = _anchor_prototype(resolved_backend, dialect)
-                        visitor.backend = get_relation_system(ref_family)(dialect=ref_dialect)
-                        visitor.expr_visitor = UnifiedExpressionVisitor(
-                            get_expression_system(ref_family)(dialect=ref_dialect)
-                        )
-                        visitor.key_context = KeyDriftContext(
-                            resource_name=n,
-                            constraints_for=self.constraints_for,
-                            schema_of=self.schema,
-                        )
-                        result = root.accept(visitor)
-                        cache[n] = UnifiedRelationVisitor._coerce_to_match(
-                            anchor_proto, result
-                        )
-                        visitor.backend, visitor.expr_visitor = relation_system, expr_visitor
-                        continue
-                    visitor.backend, visitor.expr_visitor = relation_system, expr_visitor
-                elif ref_dialect != dialect:
-                    # Same family, dialect differs from the anchor's --
-                    # covers BOTH a genuinely different known dialect AND
-                    # "ref_dialect is None but the anchor's dialect is
-                    # known" (a same-family-unknown-dialect ref must get
-                    # dialect=None explicitly, never silently inherit the
-                    # anchor's specific dialect).
-                    visitor.backend = relation_system_cls(dialect=ref_dialect)
-                    visitor.expr_visitor = UnifiedExpressionVisitor(
-                        expression_system_cls(dialect=ref_dialect)
-                    )
-                else:
-                    # Same family, same dialect as the anchor -- the
-                    # common case. Reuse the anchor's ORIGINAL objects: no
-                    # reconstruction, no new identity, no behaviour change
-                    # for a homogeneous DAG.
+                    canonical[n] = (root.accept(visitor), None, None)
+                elif ref_family == resolved_backend and ref_dialect == dialect:
+                    # Same family + same dialect as the anchor: reuse the
+                    # anchor's ORIGINAL objects (item 89's zero-cost path).
                     visitor.backend, visitor.expr_visitor = (
                         relation_system,
                         expr_visitor,
                     )
+                    canonical[n] = (root.accept(visitor), ref_family, ref_dialect)
+                else:
+                    # Foreign family, or same family with a different
+                    # dialect: compile with the ref's OWN (family, dialect)
+                    # identity -- never the anchor's -- so a dialect-scoped
+                    # CapabilityFact gates/enriches correctly, and store the
+                    # raw canonical value uncoerced; coercion happens lazily
+                    # at resolver() call time, against the ACTUAL consumer.
+                    visitor.backend = get_relation_system(ref_family)(dialect=ref_dialect)
+                    visitor.expr_visitor = UnifiedExpressionVisitor(
+                        get_expression_system(ref_family)(dialect=ref_dialect)
+                    )
+                    canonical[n] = (root.accept(visitor), ref_family, ref_dialect)
 
-                # Each dependency is key-assessed against ITS OWN
-                # constraints, unconditionally — independent of whether the
-                # target itself has a key identity (key_context may be None
-                # for an ad-hoc execute() target; that must not suppress
-                # dependency assessment).
-                visitor.key_context = KeyDriftContext(
-                    resource_name=n,
-                    constraints_for=self.constraints_for,
-                    schema_of=self.schema,
-                )
-                cache[n] = root.accept(visitor)
             # Restore the anchor's ORIGINAL backend/expr_visitor/key_context
-            # (None for ad-hoc execute()) before compiling the target itself.
+            # ONCE, after the loop -- never per-branch (a trailing no-leaf
+            # ref must not leak its key_context into the target compile).
             visitor.backend, visitor.expr_visitor, visitor.key_context = (
                 relation_system,
                 expr_visitor,

tests/relations/dag/test_dag_cross_family_bare.py

@@ -3,7 +3,6 @@
 
 import pandas as pd
 import polars as pl
-import pytest
 
 import mountainash as ma
 from mountainash.relations.dag import RelationDAG
@@ -50,12 +49,14 @@ def test_narwhals_anchor_coerces_bare_polars_dependency_to_exact_dialect(self):
         result = dag.collect("target")
         assert result is not None
 
-    def test_derived_foreign_dependency_is_not_coerced(self):
-        # Boundary (item 97): a FilterRelNode-rooted foreign dependency stays on
-        # the anchor pair and raises the raw TypeError.
+    def test_derived_foreign_dependency_is_coerced(self):
+        # Item 97 inverts this item-92 boundary: a FilterRelNode-rooted
+        # foreign dependency is now materialised in its own family and
+        # coerced at resolver time, rather than raising from the anchor's
+        # read().
         dag = RelationDAG()
-        dag.add("a_anchor", ma.relation(pl.DataFrame({"id": [1]})))
+        dag.add("a_anchor", ma.relation(pl.DataFrame({"id": [1, 2]})))
         dag.add("z_dep", ma.relation(pd.DataFrame({"id": [2]})).filter(ma.col("id") > 0))
         dag.add("target", dag.ref("a_anchor").join(dag.ref("z_dep"), on="id"))
-        with pytest.raises(TypeError, match="cannot read DataFrame"):
-            dag.collect("target")
+        result = dag.collect("target")
+        assert result.collect().to_dict(as_series=False) == {"id": [2]}

tests/relations/dag/test_dag_cross_family_derived.py

@@ -0,0 +1,89 @@
+"""Cross-family coercion for derived/transitive DAG dependency refs (item 97).
+
+Item 92 coerces only *bare* foreign-family refs (root is a ReadRelNode).
+A *derived* ref whose own root tree contains an INLINE foreign ReadRelNode
+(e.g. ma.relation(pandas_df).filter(...)) is left on the anchor pair and
+raises `TypeError: <Family> backend cannot read DataFrame.` This item
+materialises each ref once in its own family and coerces at resolver time.
+
+Design: mountainash-central
+2026-08-14-dag-cross-family-derived-dependency-coercion-design.md
+(Revision 6, 6 GLM-5.2 adversarial review rounds -- SOUND_WITH_CONCERNS).
+"""
+from __future__ import annotations
+
+import pandas as pd
+import polars as pl
+import pytest
+
+import mountainash as ma
+from mountainash.relations.dag import RelationDAG
+
+import mountainash.relations.backends  # noqa: F401
+import mountainash.expressions.backends  # noqa: F401
+
+
+def _pl(data: dict):
+    return pl.DataFrame(data)
+
+
+def _pd(data: dict):
+    return pd.DataFrame(data)
+
+
+class TestShapeBDerivedInlineRead:
+    def test_filter_rooted_inline_pandas_read_coerces_to_polars_anchor(self):
+        dag = RelationDAG()
+        dag.add("a_pol", ma.relation(_pl({"id": [1, 2], "name": ["a", "b"]})))
+        # Shape B: the foreign ReadRelNode(pandas) is INLINE in m_der's tree.
+        dag.add(
+            "m_der",
+            ma.relation(_pd({"id": [2, 3], "name": ["c", "d"]})).filter(
+                ma.col("id").gt(0)
+            ),
+        )
+        dag.add("target", dag.ref("a_pol").join(dag.ref("m_der"), on="id"))
+
+        result = dag.collect("target")  # Polars LazyFrame for a Polars anchor
+
+        assert result.collect().to_dict(as_series=False) == {
+            "id": [2],
+            "name": ["b"],
+            "name_right": ["c"],
+        }
+
+
+class TestBoundariesAndRegressions:
+    def test_project_rooted_inline_pandas_read_in_union(self):
+        dag = RelationDAG()
+        dag.add("a_pol", ma.relation(_pl({"id": [1], "name": ["a"]})))
+        dag.add(
+            "m_proj",
+            ma.relation(_pd({"id": [2], "name": ["c"]})).select("id", "name"),
+        )
+        dag.add("target", ma.concat([dag.ref("a_pol"), dag.ref("m_proj")]))
+        result = dag.collect("target")
+        assert sorted(result.collect().to_dict(as_series=False)["id"]) == [1, 2]
+
+    def test_transitive_ref_chain(self):
+        dag = RelationDAG()
+        # Names chosen so "a_pol" sorts alphabetically before "n_sel" among
+        # target's own direct refs, keeping the anchor-detection walk
+        # (item 89's deterministic "first ref alphabetically") on the
+        # Polars anchor -- this test targets the transitive-chain
+        # materialisation path, not anchor-selection order.
+        dag.add("m_raw", ma.relation(_pd({"id": [2], "name": ["c"]})).filter(ma.col("id").gt(0)))
+        dag.add("n_sel", dag.ref("m_raw").select("id"))
+        dag.add("a_pol", ma.relation(_pl({"id": [1, 2]})))
+        dag.add("target", dag.ref("a_pol").join(dag.ref("n_sel"), on="id"))
+        result = dag.collect("target")
+        assert result.collect().to_dict(as_series=False)["id"] == [2]
+
+    def test_no_leaf_ref_key_context_preserved(self):
+        # A pure inline-data (SourceRelNode) ref materialises with the anchor
+        # pair and is still key-assessed -- no key_context leak into the target.
+        dag = RelationDAG()
+        dag.add("inline", ma.relation({"id": [1]}))
+        dag.add("target", dag.ref("inline").select("id"))
+        result = dag.collect("target")
+        assert result.collect().to_dict(as_series=False)["id"] == [1]