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

Code before:
        for node in self.nodes:
            key = node.namespace.split(".")[0] if node.namespace else node.name

            if key not in grouped_nodes_map:
                grouped_nodes_map[key] = GroupedNodes(
                    name=key,
                    type="namespace" if node.namespace else "nodes",
                )

            grouped_nodes_map[key].nodes.append(node.name)

            dependencies = grouped_nodes_map[key].dependencies
            unique_dependencies = set(dependencies)

            for parent in sorted(self.node_dependencies[node], key=lambda n: n.name):
                parent_key = (
                    parent.namespace.split(".")[0] if parent.namespace else parent.name
                )
                if parent_key != key and parent_key not in unique_dependencies:
                    dependencies.append(parent_key)
                    unique_dependencies.add(parent_key)

        return list(grouped_nodes_map.values())

    # ruff: noqa: PLR0912
    def _group_by_memory(self, memory_datasets: set[str]) -> list[GroupedNodes]:
        adj_list: dict[str, set[str]] = {node.name: set() for node in self.nodes}
        parent_to_children: dict[str, set[str]] = {
            node.name: set() for node in self.nodes
        }
        output_to_node = {
            output: node for node in self.nodes for output in node.outputs
        }

        for node in self.nodes:
            for input_ in node.inputs:
                if input_ in output_to_node:
                    parent_name = output_to_node[input_].name
                    parent_to_children[parent_name].add(node.name)
                    if input_ in memory_datasets:
                        adj_list[node.name].add(parent_name)
                        adj_list[parent_name].add(node.name)

        con_components: dict[str, int] = {node.name: -1 for node in self.nodes}

        def dfs(n: str, component: int) -> None:
            if con_components[n] != -1:
                return
            con_components[n] = component
            for neighbour in adj_list[n]:
                dfs(neighbour, component)

Review comment (suggestion):
I don't feel comfortable adding this to the core of Kedro. I remember we had a pretty big discussion when this behaviour was added to Kedro Airflow in https://github.com/kedro-org/kedro-plugins/pull/241. It feels like a rash decision to just add it here without talking about it again.

The main argument against grouping by memory was that namespaces should be the advised way to group things. However, at the time there was no out of the box way to do that. Now we've added that, I feel like the grouping by memory shouldn't be encouraged anymore. 

We can always add this post 1.0.0, since that's a non-breaking change, but I want to discuss it first.

Code after:
                grouped_nodes_map[key] = GroupedNodes(
                    name=key,
                    type="namespace" if node.namespace else "nodes",
                )

            grouped_nodes_map[key].nodes.append(node.name)

            dependencies = grouped_nodes_map[key].dependencies
            unique_dependencies = set(dependencies)

            for parent in sorted(self.node_dependencies[node], key=lambda n: n.name):
                parent_key = (
                    parent.namespace.split(".")[0] if parent.namespace else parent.name
                )
                if parent_key != key and parent_key not in unique_dependencies:
                    dependencies.append(parent_key)
                    unique_dependencies.add(parent_key)

        return list(grouped_nodes_map.values())

    def only_nodes(self, *node_names: str) -> Pipeline:
        """Create a new ``Pipeline`` which will contain only the specified
        nodes by name.

        Args:
            *node_names: One or more node names. The returned ``Pipeline``
                will only contain these nodes.

        Raises:
            ValueError: When some invalid node name is given.

        Returns:
            A new ``Pipeline``, containing only ``nodes``.

        """
        unregistered_nodes = set(node_names) - set(self._nodes_by_name.keys())
        if unregistered_nodes:
            # check if unregistered nodes are available under namespace
            namespaces = []
            for unregistered_node in unregistered_nodes:
                namespaces.extend(
                    [
                        node_name
                        for node_name in self._nodes_by_name.keys()
                        if node_name.endswith(f".{unregistered_node}")
                    ]
                )
            if namespaces:
                raise ValueError(
                    f"Pipeline does not contain nodes named {list(unregistered_nodes)}. "
                    f"Did you mean: {namespaces}?"