Repository: kedro-org/kedro
File: kedro/pipeline/pipeline.py (Python)

Code before:
        visited = set([start])

        while queue:
            node, path = queue.popleft()
            if node == end:
                return path
            for child in self._node_children[node]:
                if child not in visited:
                    visited.add(child)
                    queue.append((child, [*path, child]))
        return []  # pragma: no cover

    def _validate_namespaces(self) -> None:
        from warnings import warn

        node_parents: dict[Node, set[Node]] = self.node_dependencies

        # Here we keep all visited namespaces for each node
        seen_namespaces: dict[Node, dict[str, set[Node]]] = defaultdict(dict)
        for node in self.nodes:
            visited: dict[str, set[Node]] = defaultdict(set)
            for parent in node_parents[node]:
                last_namespace = parent.namespace or ""
                curr_namespace = node.namespace or ""
                # If curr_namespace was visited in paths leading to parent
                if curr_namespace and curr_namespace in seen_namespaces[parent]:
                    start = next(iter(seen_namespaces[parent][curr_namespace]))
                    path = [n.name for n in self._shortest_path(start, node)]
                    warn(
                        f"Namespace '{curr_namespace}' is interrupted by nodes {path[:-1]} and thus invalid.",
                        UserWarning,
                    )
                # All visited namespaces for the current node get updated with the parent's visited namespaces
                for ns, nodes in seen_namespaces[parent].items():
                    visited[ns].update(nodes)
                # If the current namespace is different from the last namespace and isn't a child namespace,
                # mark the last namespace and all unrelated parent namespaces as visited to detect potential future interruptions
                if (
                    last_namespace
                    and curr_namespace != last_namespace
                    and not curr_namespace.startswith(last_namespace + ".")
                ):
                    parts = last_namespace.split(".")
                    prefix = ""
                    for p in parts:
                        prefix += p
                        if not curr_namespace.startswith(prefix):
                            visited[prefix].add(node)
                        prefix += "."
            # Now we have created the visited namespaces for the current node
            seen_namespaces[node] = visited

Review comment (suggestion):
Can you assign `seen_namespaces[parent]` to a variable for readability? I think it will help with especially the line below.

Code after:
    def _shortest_path(self, start: Node, end: Node) -> list[Node]:
        """Find the shortest path between two nodes in the pipeline, if it exists."""
        queue = deque([(start, [start])])
        visited = set([start])

        while queue:
            node, path = queue.popleft()
            if node == end:
                return path
            for child in self._node_children[node]:
                if child not in visited:
                    visited.add(child)
                    queue.append((child, [*path, child]))
        return []  # pragma: no cover

    def _validate_namespaces(self) -> None:
        node_parents: dict[Node, set[Node]] = self.node_dependencies

        # Here we keep all visited namespaces and the nodes that interrupted them for each node
        # the mapping is as follows:
        # {node: {namespace: {interrupting_node}}}
        seen_namespaces: dict[Node, dict[str, set[Node]]] = defaultdict(dict)
        for node in self.nodes:
            visited: dict[str, set[Node]] = defaultdict(set)
            for parent in node_parents[node]:
                last_namespace = parent.namespace or ""
                curr_namespace = node.namespace or ""
                seen = seen_namespaces[parent]
                # If any part of curr_namespace was visited in paths leading to parent
                interrupted = next(
                    (ns for ns in node.namespace_prefixes if ns in seen),
                    None,
                )
                # The second condition checks if the interruption was already reported during visiting the parent node
                if interrupted and not (last_namespace + ".").startswith(
                    interrupted + "."
                ):
                    start = next(iter(seen[interrupted]))
                    path = [n.name for n in self._shortest_path(start, node)]
                    warn(
                        f"Namespace '{interrupted}' is interrupted by nodes {path[:-1]} and thus invalid.",
                        UserWarning,
                    )
                # All visited namespaces for the current node get updated with the parent's visited namespaces
                for ns, nodes in seen.items():
                    visited[ns].update(nodes)
                # If the current namespace is different from the last namespace and isn't a child namespace,
                # mark the last namespace and all unrelated parent namespaces as visited to detect potential future interruptions
                if (
                    last_namespace
                    and curr_namespace != last_namespace