When dimension nodes lose columns, we should validate that no nodes that linked to that dimension or cubes that use that dimension are broken

Author: shangyian

datajunction-server/datajunction_server/internal/deployment/orchestrator.py

@@ -158,20 +158,21 @@ class _DryRunRollback(Exception):
 
 
 class _NodeIdProxy:
-    """Minimal stand-in for Node that carries the DB primary key and node type.
+    """Minimal stand-in for Node that carries the DB primary key, node type, and name.
 
     Used in ``changed_nodes`` passed to ``compute_impact`` after a session
     commit or savepoint rollback, when the actual ORM Node objects may be
     expired or in an inconsistent state.  The BFS needs ``.id`` to traverse
-    ``NodeRelationship`` and ``.type`` to check whether a changed node is a
-    DIMENSION (Stage 3 of the BFS).
+    ``NodeRelationship``, ``.type`` to check whether a changed node is a
+    DIMENSION (Stage 3 of the BFS), and ``.name`` for get_dimension_inbound_bfs.
     """
 
-    __slots__ = ("id", "type")
+    __slots__ = ("id", "type", "name")
 
-    def __init__(self, node_id: int, node_type: "NodeType") -> None:
+    def __init__(self, node_id: int, node_type: "NodeType", name: str) -> None:
         self.id = node_id
         self.type = node_type
+        self.name = name
 
 
 @dataclass
@@ -468,7 +469,7 @@ async def _compute_downstream_impact(
             node_type = pre_deploy_node_types.get(name, node_spec.node_type)
             changed_nodes[name] = (
                 explicit_spec,
-                cast(Node, _NodeIdProxy(node_id, node_type)),
+                cast(Node, _NodeIdProxy(node_id, node_type, name)),
                 set(),
             )
 
@@ -479,7 +480,7 @@ async def _compute_downstream_impact(
                 node_type = pre_deploy_node_types.get(name, node_spec.node_type)
                 changed_nodes[name] = (
                     None,
-                    cast(Node, _NodeIdProxy(node_id, node_type)),
+                    cast(Node, _NodeIdProxy(node_id, node_type, name)),
                     set(),
                 )
 

datajunction-server/datajunction_server/internal/impact.py

@@ -20,6 +20,7 @@
 from sqlalchemy.sql.expression import bindparam, text
 
 from datajunction_server.database.column import Column
+from datajunction_server.database.dimensionlink import DimensionLink
 from datajunction_server.database.node import Node, NodeRevision
 from datajunction_server.internal.deployment.validation import bulk_validate_node_data
 from datajunction_server.models.deployment import (
@@ -41,6 +42,7 @@
     get_downstream_nodes,
 )
 from datajunction_server.sql.parsing.types import PRIMITIVE_TYPES
+from datajunction_server.utils import SEPARATOR
 
 _logger = logging.getLogger(__name__)
 
@@ -795,18 +797,27 @@ async def _seed_bfs(
                 )
 
     # ---------------------------------------------------------------------------
-    # Stage 3 — dimension-loses-columns pass
+    # Stage 3 — dimension-loses-columns (or becomes-invalid) pass
     # ---------------------------------------------------------------------------
-    dim_nodes_losing_cols = [
-        state.node_cache[name]
-        for name, prop in state.propagated_change.items()
-        if prop.columns_removed
-        and state.node_cache.get(name) is not None
-        and state.node_cache[name].type == NodeType.DIMENSION
-    ]
-    if dim_nodes_losing_cols:
+    # Trigger when a dimension node loses specific columns OR becomes entirely
+    # invalid (e.g. new query fails validation).  ``dim_is_invalid=True`` means
+    # we couldn't determine which columns changed — treat all holders as broken.
+    dim_nodes_stage3: list[tuple[Node, set[str], bool]] = []
+    for name, prop in state.propagated_change.items():
+        node = state.node_cache.get(name)
+        if node is None or node.type != NodeType.DIMENSION:
+            continue
+        if prop.columns_removed:
+            dim_nodes_stage3.append((node, prop.columns_removed, False))
+        elif prop.is_invalid or (
+            prop.proposed_columns is not None and not prop.proposed_columns
+        ):
+            # Dimension became invalid — all dimension-link holders are affected
+            dim_nodes_stage3.append((node, set(), True))
+
+    if dim_nodes_stage3:
         holder_names: set[str] = set()
-        for dim_node in dim_nodes_losing_cols:
+        for dim_node, _, _ in dim_nodes_stage3:
             holder_displays, _ = await get_dimension_inbound_bfs(
                 session,
                 dim_node,
@@ -819,7 +830,7 @@ async def _seed_bfs(
             list(holder_names),
             options=list(_node_output_options()),
         )
-        dim_node_names = [n.name for n in dim_nodes_losing_cols]
+        dim_node_names = [n.name for n, _, _ in dim_nodes_stage3]
         for holder in holders:
             if holder.name not in state.propagated_change:  # pragma: no branch
                 state.propagated_change[holder.name] = _PropagatedChange(
@@ -831,6 +842,40 @@ async def _seed_bfs(
             state.node_cache[holder.name] = holder
             state.frontier.add(holder.name)
             state.visited.add(holder.name)
+
+            # Determine whether the holder's dimension link is broken:
+            # - dimension invalid entirely → all links to it are broken
+            # - specific PK-side (join-target) columns removed → link is broken
+            for dim_node, removed_cols, dim_is_invalid in dim_nodes_stage3:
+                for link in holder.current.dimension_links:
+                    if link.dimension.name != dim_node.name:
+                        continue
+                    broken_pk_cols = (
+                        set()
+                        if dim_is_invalid
+                        else _dim_pk_col_names(link) & removed_cols
+                    )
+                    if dim_is_invalid or broken_pk_cols:
+                        reason = (
+                            f"Dimension link to {dim_node.name} is broken: "
+                            "dimension node is invalid"
+                            if dim_is_invalid
+                            else (
+                                f"Dimension link to {dim_node.name} is broken: "
+                                f"join-target columns removed: "
+                                f"{', '.join(sorted(broken_pk_cols))}"
+                            )
+                        )
+                        _record_impact(
+                            state.impacted,
+                            holder,
+                            impact_type="dimension_link",
+                            caused_by=[dim_node.name],
+                            reason=reason,
+                        )
+                        state.propagated_change[holder.name].is_invalid = True
+                        break
+
             await _check_cube_dim_link_impacts(
                 session=session,
                 node_name=holder.name,
@@ -902,10 +947,14 @@ async def _propagate_column_change(
                 caused_by=[parent_name],
                 reason=_column_reason(prop),
             )
+            # Case 3: if this child has dimension links whose FK columns are now
+            # absent from its proposed output, those links are broken too.
+            new_col_names = {c.name for c in new_cols} if new_cols else set()
+            broken_links = _broken_fk_dim_links(child, new_col_names)
             _merge_propagated(
                 state.propagated_change,
                 child_name,
-                dim_links_removed=set(),
+                dim_links_removed=broken_links,
                 is_deleted=False,
                 is_invalid=len(new_cols) == 0,
                 caused_by=[parent_name],
@@ -1112,3 +1161,39 @@ def _column_reason(change: _PropagatedChange) -> str:
 def _dim_link_reason(change: _PropagatedChange) -> str:
     dims = sorted(change.dim_links_removed)
     return f"Dimension links removed: {', '.join(dims)}"
+
+
+def _dim_pk_col_names(link: "DimensionLink") -> set[str]:
+    """Return the dimension-side column names used as join targets in this link.
+
+    Parses the link's join SQL to find which columns on the dimension node appear
+    in the equality conditions.  Returns an empty set if parsing fails.
+    """
+    try:
+        dim_prefix = link.dimension.name + SEPARATOR
+        return {
+            pk_ident[len(dim_prefix) :]
+            if pk_ident.startswith(dim_prefix)
+            else pk_ident.rsplit(SEPARATOR, 1)[-1]
+            for pk_ident in link.foreign_keys.values()
+            if pk_ident is not None
+        }
+    except Exception:
+        return set()
+
+
+def _broken_fk_dim_links(child: "Node", new_col_names: set[str]) -> set[str]:
+    """Return dimension node names whose FK link on ``child`` is now broken.
+
+    A link is broken when one or more of the child's FK columns used in the
+    join condition are absent from ``new_col_names`` (the proposed output columns
+    after an upstream change).
+    """
+    broken: set[str] = set()
+    for link in child.current.dimension_links:
+        try:
+            if link.foreign_key_column_names - new_col_names:
+                broken.add(link.dimension.name)
+        except Exception:
+            pass
+    return broken

datajunction-server/datajunction_server/internal/validation.py

@@ -33,7 +33,8 @@ def update_ast_column_types(node: ast.Node) -> None:
     raw ColumnType strings.
     """
     if isinstance(node, ast.Column) and node._table and hasattr(node._table, "dj_node"):
-        for db_col in node._table.dj_node.columns:
+        dj_node = node._table.dj_node  # type: ignore[attr-defined]
+        for db_col in dj_node.columns:
             if db_col.name == node.name.name:
                 node._type = db_col.type
                 break