Repository: kedro-org/kedro
Pull request: Improve resume pipeline suggestion for SequentialRunner
## Description
<!-- Why was this PR created? -->
Resolves [#1477](https://github.com/kedro-org/kedro/issues/1477)

## Development notes
<!-- What have you changed, and how has this been tested? -->
After a failed run, Kedro suggests a command to the user:

`You can resume the pipeline run by adding the following argument to your previous command:
                               --from-nodes "node4_B"`

Before this PR, the suggested command will run from the last nodes to be executed, regardless of whether their input was persisted or not. If any of the inputs to the listed nodes is not persisted, the run immediately fails again.

After this PR, the suggested command will run from the closest successfully executed nodes with persisted inputs:

`You can resume the pipeline run from the nearest nodes with persisted inputs by adding the following argument to your previous command:
                               --from-nodes "node1_B,node1_A"`

This is achieved by performing a breadth-first search, starting at the last successfully executed nodes. This backward search yields a set of the nearest nodes that have persisted inputs.

Six tests are added to the `test_sequential_runner` test suite to test different cases on an X-shaped pipeline.

## Limitations
This change is a significant improvement, but there are still two important limitations:

- Persisted inputs are defined to be any that are not `MemoryDataSet`s. This definition has limitations; it does not account for custom datasets that are not persisted.
- Neither the approach in this PR nor the previous one handle the case where nodes append to datasets. Running these nodes repeatedly could have unintended consequences.

In the future, I think it would be a good idea to add a method to the API of AbstractDataSet that checks for persistence. I would love to hear thoughts on this.

## Checklist

- [x] Read the [contributing](https://github.com/kedro-org/kedro/blob/main/CONTRIBUTING.md) guidelines
- [x] Opened this PR as a 'Draft Pull Request' if it is work-in-progress
- [ ] Updated the documentation to reflect the code changes
- [ ] Added a description of this change in the [`RELEASE.md`](https://github.com/kedro-org/kedro/blob/main/RELEASE.md) file
- [x] Added tests to cover my changes


<a href="https://gitpod.io/#https://github.com/kedro-org/kedro/pull/1795"><img src="https://gitpod.io/button/open-in-gitpod.svg"/></a>



Problem statement:
[KED-2857] Suggestion for re-run command on failure, i.e. kedro run --from-nodes, should take into account memory dataset
(transfer from Jira, created by @limdauto)

Currently, when a pipeline fails, there is a suggested command to re-run the pipeline from a particular node. However, if the inputs to this node are MemoryDataSet, the pipeline won't be run correctly. A user suggested that Kedro would back track to the last succesful node with a persisted dataset and continue from there.

This ticket is to update the error message.

Hints from issue discussion:

Commit message: Merge branch 'main' into feat/improve-resume-scenario-suggestion

Patch to review:
diff --git a/kedro/runner/parallel_runner.py b/kedro/runner/parallel_runner.py
index ee744d5871..89dd4ce41c 100644
--- a/kedro/runner/parallel_runner.py
+++ b/kedro/runner/parallel_runner.py
@@ -335,7 +335,7 @@ def _run(  # pylint: disable=too-many-locals,useless-suppression
                     try:
                         node = future.result()
                     except Exception:
-                        self._suggest_resume_scenario(pipeline, done_nodes)
+                        self._suggest_resume_scenario(pipeline, done_nodes, catalog)
                         raise
                     done_nodes.add(node)
 
diff --git a/kedro/runner/runner.py b/kedro/runner/runner.py
index 79ffd10d48..ca5e6faea3 100644
--- a/kedro/runner/runner.py
+++ b/kedro/runner/runner.py
@@ -4,6 +4,7 @@
 
 import logging
 from abc import ABC, abstractmethod
+from collections import deque
 from concurrent.futures import (
     ALL_COMPLETED,
     Future,
@@ -11,12 +12,12 @@
     as_completed,
     wait,
 )
-from typing import Any, Dict, Iterable
+from typing import Any, Dict, Iterable, List, Set
 
 from pluggy import PluginManager
 
 from kedro.framework.hooks.manager import _NullPluginManager
-from kedro.io import AbstractDataSet, DataCatalog
+from kedro.io import AbstractDataSet, DataCatalog, MemoryDataSet
 from kedro.pipeline import Pipeline
 from kedro.pipeline.node import Node
 
@@ -157,31 +158,121 @@ def create_default_data_set(self, ds_name: str) -> AbstractDataSet:
         Returns:
             An instance of an implementation of ``AbstractDataSet`` to be
             used for all unregistered datasets.
-
         """
         pass
 
     def _suggest_resume_scenario(
-        self, pipeline: Pipeline, done_nodes: Iterable[Node]
+        self,
+        pipeline: Pipeline,
+        done_nodes: Iterable[Node],
+        catalog: DataCatalog,
     ) -> None:
+        """
+        Suggest a command to the user to resume a run after it fails.
+        The run should be started from the point closest to the failure
+        for which persisted input exists.
+
+        Args:
+            pipeline: the ``Pipeline`` of the run.
+            done_nodes: the ``Node``s that executed successfully.
+            catalog: the ``DataCatalog`` of the run.
+
+        """
         remaining_nodes = set(pipeline.nodes) - set(done_nodes)
 
         postfix = ""
         if done_nodes:
             node_names = (n.name for n in remaining_nodes)
             resume_p = pipeline.only_nodes(*node_names)
-
             start_p = resume_p.only_nodes_with_inputs(*resume_p.inputs())
-            start_node_names = (n.name for n in start_p.nodes)
+
+            # find the nearest persistent ancestors of the nodes in start_p
+            start_p_persistent_ancestors = _find_persistent_ancestors(
+                pipeline, start_p.nodes, catalog
+            )
+
+            start_node_names = (n.name for n in start_p_persistent_ancestors)
             postfix += f"  --from-nodes \"{','.join(start_node_names)}\""
 
-        self._logger.warning(
-            "There are %d nodes that have not run.\n"
-            "You can resume the pipeline run by adding the following "
-            "argument to your previous command:\n%s",
-            len(remaining_nodes),
-            postfix,
-        )
+        if not postfix:
+            self._logger.warning(
+                "No nodes ran. Repeat the previous command to attempt a new run."
+            )
+        else:
+            self._logger.warning(
+                "There are %d nodes that have not run.\n"
+                "You can resume the pipeline run from the nearest nodes with "
+                "persisted inputs by adding the following "
+                "argument to your previous command:\n%s",
+                len(remaining_nodes),
+                postfix,
+            )
+
+
+def _find_persistent_ancestors(
+    pipeline: Pipeline, children: Iterable[Node], catalog: DataCatalog
+) -> Set[Node]:
+    """Breadth-first search approach to finding the complete set of
+    persistent ancestors of an iterable of ``Node``s. Persistent
+    ancestors exclusively have persisted ``Dataset``s as inputs.
+
+    Args:
+        pipeline: the ``Pipeline`` to find ancestors in.
+        children: the iterable containing ``Node``s to find ancestors of.
+        catalog: the ``DataCatalog`` of the run.
+
+    Returns:
+        A set containing first persistent ancestors of the given
+        ``Node``s.
+
+    """
+    ancestor_nodes_to_run = set()
+    queue, visited = deque(children), set()
+    while queue:
+        current_node = queue.popleft()
+        visited.add(current_node)
+        if _has_persistent_inputs(current_node, catalog):
+            ancestor_nodes_to_run.add(current_node)
+            continue
+        for parent in _enumerate_parents(pipeline, current_node):
+            if parent not in visited:
+                queue.append(parent)
+    return ancestor_nodes_to_run
+
+
+def _enumerate_parents(pipeline: Pipeline, child: Node) -> List[Node]:
+    """For a given ``Node``, returns a list containing the direct parents
+    of that ``Node`` in the given ``Pipeline``.
+
+    Args:
+        pipeline: the ``Pipeline`` to search for direct parents in.
+        child: the ``Node`` to find parents of.
+
+    Returns:
+        A list of all ``Node``s that are direct parents of ``child``.
+
+    """
+    parent_pipeline = pipeline.only_nodes_with_outputs(*child.inputs)
+    return parent_pipeline.nodes
+
+
+def _has_persistent_inputs(node: Node, catalog: DataCatalog) -> bool:
+    """Check if a ``Node`` exclusively has persisted Datasets as inputs.
+    If at least one input is a ``MemoryDataSet``, return False.
+
+    Args:
+        node: the ``Node`` to check the inputs of.
+        catalog: the ``DataCatalog`` of the run.
+
+    Returns:
+        True if the ``Node`` being checked exclusively has inputs that
+        are not ``MemoryDataSet``, else False.
+
+    """
+    for node_input in node.inputs:
+        if isinstance(vars(catalog.datasets)[node_input], MemoryDataSet):
+            return False
+    return True
 
 
 def run_node(
diff --git a/kedro/runner/sequential_runner.py b/kedro/runner/sequential_runner.py
index dcad9324fe..6a9becb868 100644
--- a/kedro/runner/sequential_runner.py
+++ b/kedro/runner/sequential_runner.py
@@ -70,7 +70,7 @@ def _run(
                 run_node(node, catalog, hook_manager, self._is_async, session_id)
                 done_nodes.add(node)
             except Exception:
-                self._suggest_resume_scenario(pipeline, done_nodes)
+                self._suggest_resume_scenario(pipeline, done_nodes, catalog)
                 raise
 
             # decrement load counts and release any data sets we've finished with
diff --git a/kedro/runner/thread_runner.py b/kedro/runner/thread_runner.py
index 57136f67e0..6e34484860 100644
--- a/kedro/runner/thread_runner.py
+++ b/kedro/runner/thread_runner.py
@@ -131,7 +131,7 @@ def _run(  # pylint: disable=too-many-locals,useless-suppression
                     try:
                         node = future.result()
                     except Exception:
-                        self._suggest_resume_scenario(pipeline, done_nodes)
+                        self._suggest_resume_scenario(pipeline, done_nodes, catalog)
                         raise
                     done_nodes.add(node)
                     self._logger.info("Completed node: %s", node.name)
diff --git a/tests/runner/conftest.py b/tests/runner/conftest.py
index ddde22059a..ac4e9d8e1e 100644
--- a/tests/runner/conftest.py
+++ b/tests/runner/conftest.py
@@ -22,7 +22,7 @@ def fan_in(*args):
     return args
 
 
-def exception_fn(arg):
+def exception_fn(*args):
     raise Exception("test exception")
 
 
diff --git a/tests/runner/test_sequential_runner.py b/tests/runner/test_sequential_runner.py
index 6bcc01a139..4684f7c0ad 100644
--- a/tests/runner/test_sequential_runner.py
+++ b/tests/runner/test_sequential_runner.py
@@ -1,3 +1,4 @@
+import re
 from typing import Any, Dict
 
 import pandas as pd
@@ -13,7 +14,7 @@
 )
 from kedro.pipeline import Pipeline, node
 from kedro.runner import SequentialRunner
-from tests.runner.conftest import identity, sink, source
+from tests.runner.conftest import exception_fn, identity, sink, source
 
 
 @pytest.fixture
@@ -278,3 +279,77 @@ def test_confirms(self, mocker, pipeline, is_async):
         catalog = DataCatalog(data_sets={"ds1": fake_dataset_instance})
         SequentialRunner(is_async=is_async).run(pipeline, catalog)
         fake_dataset_instance.confirm.assert_called_once_with()
+
+
+@pytest.fixture
+def resume_scenario_pipeline():
+    return Pipeline(
+        [
+            node(identity, "ds0_A", "ds1_A", name="node1_A"),
+            node(identity, "ds0_B", "ds1_B", name="node1_B"),
+            node(
+                multi_input_list_output,
+                ["ds1_A", "ds1_B"],
+                ["ds2_A", "ds2_B"],
+                name="node2",
+            ),
+            node(identity, "ds2_A", "ds3_A", name="node3_A"),
+            node(identity, "ds2_B", "ds3_B", name="node3_B"),
+            node(identity, "ds3_A", "ds4_A", name="node4_A"),
+            node(identity, "ds3_B", "ds4_B", name="node4_B"),
+        ]
+    )
+
+
+@pytest.fixture
+def resume_scenario_catalog():
+    def _load():
+        return 0
+
+    def _save(arg):
+        assert arg == 0
+
+    persistent_dataset = LambdaDataSet(load=_load, save=_save)
+    return DataCatalog(
+        {
+            "ds0_A": persistent_dataset,
+            "ds0_B": persistent_dataset,
+            "ds2_A": persistent_dataset,
+            "ds2_B": persistent_dataset,
+        }
+    )
+
+
+@pytest.mark.parametrize(
+    "failing_node_indexes,expected_pattern",
+    [
+        ([0], r"No nodes ran."),
+        ([2], r"(node1_A,node1_B|node1_B,node1_A)"),
+        ([3], r"(node3_A,node3_B|node3_B,node3_A)"),
+        ([5], r"(node3_A,node3_B|node3_B,node3_A)"),
+        ([3, 5], r"(node3_A,node3_B|node3_B,node3_A)"),
+        ([2, 5], r"(node1_A,node1_B|node1_B,node1_A)"),
+    ],
+)
+class TestSuggestResumeScenario:
+    def test_suggest_resume_scenario(
+        self,
+        caplog,
+        resume_scenario_catalog,
+        resume_scenario_pipeline,
+        failing_node_indexes,
+        expected_pattern,
+    ):
+        for idx in failing_node_indexes:
+            failing_node = resume_scenario_pipeline.nodes[idx]
+            resume_scenario_pipeline -= Pipeline([failing_node])
+            resume_scenario_pipeline += Pipeline(
+                [failing_node._copy(func=exception_fn)]
+            )
+        with pytest.raises(Exception):
+            SequentialRunner().run(
+                resume_scenario_pipeline,
+                resume_scenario_catalog,
+                hook_manager=_create_hook_manager(),
+            )
+        assert re.search(expected_pattern, caplog.text)


Code review comments:
diff_hunk: @@ -157,31 +158,121 @@ def create_default_data_set(self, ds_name: str) -> AbstractDataSet:
         Returns:
             An instance of an implementation of ``AbstractDataSet`` to be
             used for all unregistered datasets.
-
         """
         pass
 
     def _suggest_resume_scenario(
-        self, pipeline: Pipeline, done_nodes: Iterable[Node]
+        self,
+        pipeline: Pipeline,
+        done_nodes: Iterable[Node],
+        catalog: DataCatalog,
     ) -> None:
+        """
+        Suggest a command to the user to resume a run after it fails.
+        The run should be started from the point closest to the failure
+        for which persisted input exists.
+
+        Args:
+            pipeline: the ``Pipeline`` of the run.
+            done_nodes: the ``Node``s that executed successfully.
+            catalog: the ``DataCatalog`` of the run.
+
+        """
         remaining_nodes = set(pipeline.nodes) - set(done_nodes)
 
         postfix = ""
         if done_nodes:
             node_names = (n.name for n in remaining_nodes)
             resume_p = pipeline.only_nodes(*node_names)
-
             start_p = resume_p.only_nodes_with_inputs(*resume_p.inputs())
-            start_node_names = (n.name for n in start_p.nodes)
+
+            # find the nearest persistent ancestors of the nodes in start_p
+            start_p_persistent_ancestors = _find_persistent_ancestors(
+                pipeline, start_p.nodes, catalog
+            )
+
+            start_node_names = (n.name for n in start_p_persistent_ancestors)
             postfix += f"  --from-nodes \"{','.join(start_node_names)}\""
 
-        self._logger.warning(
-            "There are %d nodes that have not run.\n"
-            "You can resume the pipeline run by adding the following "
-            "argument to your previous command:\n%s",
-            len(remaining_nodes),
-            postfix,
-        )
+        if not postfix:
+            self._logger.warning(
+                "No nodes ran. Repeat the previous command to attempt a new run."
+            )
+        else:
+            self._logger.warning(
+                "There are %d nodes that have not run.\n"
+                "You can resume the pipeline run from the nearest nodes with "
+                "persisted inputs by adding the following "
+                "argument to your previous command:\n%s",
+                len(remaining_nodes),
+                postfix,
+            )
+
+
+def _find_persistent_ancestors(
+    pipeline: Pipeline, children: Iterable[Node], catalog: DataCatalog
+) -> Set[Node]:
+    """Breadth-first search approach to finding the complete set of
+    persistent ancestors of an iterable of ``Node``s. Persistent
+    ancestors exclusively have persisted ``Dataset``s as inputs.
+
+    Args:
+        pipeline: the ``Pipeline`` to find ancestors in.
+        children: the iterable containing ``Node``s to find ancestors of.
+        catalog: the ``DataCatalog`` of the run.
+
+    Returns:
+        A set containing first persistent ancestors of the given
+        ``Node``s.
+
+    """
+    ancestor_nodes_to_run = set()
+    queue, visited = deque(children), set()
+    while queue:
+        current_node = queue.popleft()
+        visited.add(current_node)
+        if _has_persistent_inputs(current_node, catalog):
+            ancestor_nodes_to_run.add(current_node)
+            continue
+        for parent in _enumerate_parents(pipeline, current_node):
+            if parent not in visited:
+                queue.append(parent)
+    return ancestor_nodes_to_run
+
+
+def _enumerate_parents(pipeline: Pipeline, child: Node) -> List[Node]:
+    """For a given ``Node``, returns a list containing the direct parents
+    of that ``Node`` in the given ``Pipeline``.
+
+    Args:
+        pipeline: the ``Pipeline`` to search for direct parents in.
+        child: the ``Node`` to find parents of.
+
+    Returns:
+        A list of all ``Node``s that are direct parents of ``child``.
+
+    """
+    parent_pipeline = pipeline.only_nodes_with_outputs(*child.inputs)
+    return parent_pipeline.nodes
+
+
+def _has_persistent_inputs(node: Node, catalog: DataCatalog) -> bool:
+    """Check if a ``Node`` exclusively has persisted Datasets as inputs.
+    If at least one input is a ``MemoryDataSet``, return False.
+
+    Args:
+        node: the ``Node`` to check the inputs of.
+        catalog: the ``DataCatalog`` of the run.
+
+    Returns:
+        True if the ``Node`` being checked exclusively has inputs that
+        are not ``MemoryDataSet``, else False.
+
+    """
+    for node_input in node.inputs:
+        if isinstance(vars(catalog.datasets)[node_input], MemoryDataSet):
path: kedro/runner/runner.py
text: @user1:
Why do you need to do `vars(catalog.datasets)`?

@author:
catalog.datasets is an object of `_FrozenDatasets`, which has no `__getitem__` defined for it. It stores datasets in the `__dict__` attribute, which can be accessed in a readable way by using `vars()`.

The alternative would be `catalog.datasets.__dict__`, or to use the protected `catalog._data_sets` (which is already a dict) but linters won't allow these.

@user2:
TIL! I've literally never used `vars` before so this is really interesting to hear! 👏 

That said, the way I would recommend doing this would actually be `catalog._get_dataset(dataset_name)` and ignore any lint error it might give. This is a bit more obvious and would be consistent with[ how it's done in kedro-viz](https://github.com/kedro-org/kedro-viz/blob/1d7f37eab55fb12f596f5581997b9fbffefd8735/package/kedro_viz/data_access/repositories/catalog.py#L55), which I tend to regard as the source of truth for best practice in such matters.

@user2:
P.S. yes, it is kind of ridiculous we don't have a nicer way of doing this. It's one thing I would love to improve: see the first bullet point in https://github.com/kedro-org/kedro/issues/1778.

@user1:
@user3 but isn't the problem here that you don't have the actual dataset names?

@user2:
`node_input` is the dataset name here, no?

@user1:
oh! yeah probably! I keep getting so confused by this part of the code base 😅 

@author:
> That said, the way I would recommend doing this would actually be `catalog._get_dataset(dataset_name)` and ignore any lint error it might give.

Thanks for the feedback! Yeah ``vars()`` is quite uncommon I feel 😁. I made this change in [7c7e795](https://github.com/kedro-org/kedro/pull/1795/commits/7c7e79546c2739d034d03444a00dc0fdb6304414).

diff_hunk: @@ -278,3 +279,77 @@ def test_confirms(self, mocker, pipeline, is_async):
         catalog = DataCatalog(data_sets={"ds1": fake_dataset_instance})
         SequentialRunner(is_async=is_async).run(pipeline, catalog)
         fake_dataset_instance.confirm.assert_called_once_with()
+
+
+@pytest.fixture
+def resume_scenario_pipeline():
+    return Pipeline(
path: tests/runner/test_sequential_runner.py
text: @user1:
```suggestion
    return pipeline(
```
very minor, but ideally all tests would use `pipeline` rather than `Pipeline` now (there's lots of other instances of this which seem to have been missed outside your code here - feel free to change them all).

@author:
Sure, I made this change for all instances of Pipeline in the file in [e4a6525](https://github.com/kedro-org/kedro/pull/1795/commits/e4a65257d878d8aa9776923c2c6d58ad25417610).

Where this created naming conflicts with variables / parameters, I renamed these `test_pipeline`.

diff_hunk: @@ -278,3 +279,77 @@ def test_confirms(self, mocker, pipeline, is_async):
         catalog = DataCatalog(data_sets={"ds1": fake_dataset_instance})
         SequentialRunner(is_async=is_async).run(pipeline, catalog)
         fake_dataset_instance.confirm.assert_called_once_with()
+
+
+@pytest.fixture
+def resume_scenario_pipeline():
+    return Pipeline(
+        [
+            node(identity, "ds0_A", "ds1_A", name="node1_A"),
+            node(identity, "ds0_B", "ds1_B", name="node1_B"),
+            node(
+                multi_input_list_output,
+                ["ds1_A", "ds1_B"],
+                ["ds2_A", "ds2_B"],
+                name="node2",
+            ),
+            node(identity, "ds2_A", "ds3_A", name="node3_A"),
+            node(identity, "ds2_B", "ds3_B", name="node3_B"),
+            node(identity, "ds3_A", "ds4_A", name="node4_A"),
+            node(identity, "ds3_B", "ds4_B", name="node4_B"),
+        ]
+    )
+
+
+@pytest.fixture
+def resume_scenario_catalog():
+    def _load():
+        return 0
+
+    def _save(arg):
+        assert arg == 0
+
+    persistent_dataset = LambdaDataSet(load=_load, save=_save)
+    return DataCatalog(
+        {
+            "ds0_A": persistent_dataset,
+            "ds0_B": persistent_dataset,
+            "ds2_A": persistent_dataset,
+            "ds2_B": persistent_dataset,
+        }
+    )
+
+
+@pytest.mark.parametrize(
+    "failing_node_indexes,expected_pattern",
+    [
+        ([0], r"No nodes ran."),
path: tests/runner/test_sequential_runner.py
text: @user1:
Would it be possible to label these by node name rather than index? e.g. `["node1_A"]` here and then maybe `failing_nodes = pipeline.only_nodes(*failing_node_names)`. This way you don't have to understand how node index maps onto the pipeline topology.

@author:
Very good point, I have implemented this in [1bf5f1d](https://github.com/kedro-org/kedro/pull/1795/commits/1bf5f1de48ff9634b2c207212db0afb23f537b06).

diff_hunk: @@ -157,31 +158,121 @@ def create_default_data_set(self, ds_name: str) -> AbstractDataSet:
         Returns:
             An instance of an implementation of ``AbstractDataSet`` to be
             used for all unregistered datasets.
-
         """
         pass
 
     def _suggest_resume_scenario(
-        self, pipeline: Pipeline, done_nodes: Iterable[Node]
+        self,
+        pipeline: Pipeline,
+        done_nodes: Iterable[Node],
+        catalog: DataCatalog,
     ) -> None:
+        """
+        Suggest a command to the user to resume a run after it fails.
+        The run should be started from the point closest to the failure
+        for which persisted input exists.
+
+        Args:
+            pipeline: the ``Pipeline`` of the run.
+            done_nodes: the ``Node``s that executed successfully.
+            catalog: the ``DataCatalog`` of the run.
+
+        """
         remaining_nodes = set(pipeline.nodes) - set(done_nodes)
 
         postfix = ""
         if done_nodes:
             node_names = (n.name for n in remaining_nodes)
             resume_p = pipeline.only_nodes(*node_names)
-
             start_p = resume_p.only_nodes_with_inputs(*resume_p.inputs())
-            start_node_names = (n.name for n in start_p.nodes)
+
+            # find the nearest persistent ancestors of the nodes in start_p
+            start_p_persistent_ancestors = _find_persistent_ancestors(
+                pipeline, start_p.nodes, catalog
+            )
+
+            start_node_names = (n.name for n in start_p_persistent_ancestors)
             postfix += f"  --from-nodes \"{','.join(start_node_names)}\""
 
-        self._logger.warning(
-            "There are %d nodes that have not run.\n"
-            "You can resume the pipeline run by adding the following "
-            "argument to your previous command:\n%s",
-            len(remaining_nodes),
-            postfix,
-        )
+        if not postfix:
+            self._logger.warning(
+                "No nodes ran. Repeat the previous command to attempt a new run."
+            )
+        else:
+            self._logger.warning(
+                "There are %d nodes that have not run.\n"
+                "You can resume the pipeline run from the nearest nodes with "
+                "persisted inputs by adding the following "
+                "argument to your previous command:\n%s",
+                len(remaining_nodes),
+                postfix,
+            )
+
+
+def _find_persistent_ancestors(
+    pipeline: Pipeline, children: Iterable[Node], catalog: DataCatalog
+) -> Set[Node]:
+    """Breadth-first search approach to finding the complete set of
+    persistent ancestors of an iterable of ``Node``s. Persistent
+    ancestors exclusively have persisted ``Dataset``s as inputs.
+
+    Args:
+        pipeline: the ``Pipeline`` to find ancestors in.
+        children: the iterable containing ``Node``s to find ancestors of.
+        catalog: the ``DataCatalog`` of the run.
+
+    Returns:
+        A set containing first persistent ancestors of the given
+        ``Node``s.
+
+    """
+    ancestor_nodes_to_run = set()
+    queue, visited = deque(children), set()
+    while queue:
+        current_node = queue.popleft()
+        visited.add(current_node)
+        if _has_persistent_inputs(current_node, catalog):
+            ancestor_nodes_to_run.add(current_node)
+            continue
+        for parent in _enumerate_parents(pipeline, current_node):
+            if parent not in visited:
path: kedro/runner/runner.py
text: @user1:
As I understand it, this condition is to prevent nodes from being added to the queue several times over. Could this be extended to exclude parents that may already be in the queue already? For example, two nodes A and B that both share parents C and D, and a queue [A|B]; the current implementation would lead to a queue [C|D|C|D] with the set `visited` being {A,B}.

@author:
Thanks for the review Ahdra! You are right about this, nicely spotted. The solution for this is to check for presence in visited when iterating over a `Node`'s parents, not when the node is taken off the queue. The improvement is in [836d583](https://github.com/kedro-org/kedro/pull/1795/commits/836d58364712e900dc794823e1eb8875beaf0ac5).

diff_hunk: @@ -157,31 +158,121 @@ def create_default_data_set(self, ds_name: str) -> AbstractDataSet:
         Returns:
             An instance of an implementation of ``AbstractDataSet`` to be
             used for all unregistered datasets.
-
         """
         pass
 
     def _suggest_resume_scenario(
-        self, pipeline: Pipeline, done_nodes: Iterable[Node]
+        self,
+        pipeline: Pipeline,
+        done_nodes: Iterable[Node],
+        catalog: DataCatalog,
     ) -> None:
+        """
+        Suggest a command to the user to resume a run after it fails.
+        The run should be started from the point closest to the failure
+        for which persisted input exists.
+
+        Args:
+            pipeline: the ``Pipeline`` of the run.
+            done_nodes: the ``Node``s that executed successfully.
+            catalog: the ``DataCatalog`` of the run.
+
+        """
         remaining_nodes = set(pipeline.nodes) - set(done_nodes)
 
         postfix = ""
         if done_nodes:
             node_names = (n.name for n in remaining_nodes)
             resume_p = pipeline.only_nodes(*node_names)
-
             start_p = resume_p.only_nodes_with_inputs(*resume_p.inputs())
-            start_node_names = (n.name for n in start_p.nodes)
+
+            # find the nearest persistent ancestors of the nodes in start_p
+            start_p_persistent_ancestors = _find_persistent_ancestors(
+                pipeline, start_p.nodes, catalog
+            )
+
+            start_node_names = (n.name for n in start_p_persistent_ancestors)
             postfix += f"  --from-nodes \"{','.join(start_node_names)}\""
 
-        self._logger.warning(
-            "There are %d nodes that have not run.\n"
-            "You can resume the pipeline run by adding the following "
-            "argument to your previous command:\n%s",
-            len(remaining_nodes),
-            postfix,
-        )
+        if not postfix:
+            self._logger.warning(
+                "No nodes ran. Repeat the previous command to attempt a new run."
+            )
+        else:
+            self._logger.warning(
+                "There are %d nodes that have not run.\n"
+                "You can resume the pipeline run from the nearest nodes with "
+                "persisted inputs by adding the following "
+                "argument to your previous command:\n%s",
+                len(remaining_nodes),
+                postfix,
+            )
+
+
+def _find_persistent_ancestors(
+    pipeline: Pipeline, children: Iterable[Node], catalog: DataCatalog
+) -> Set[Node]:
+    """Breadth-first search approach to finding the complete set of
+    persistent ancestors of an iterable of ``Node``s. Persistent
+    ancestors exclusively have persisted ``Dataset``s as inputs.
+
+    Args:
+        pipeline: the ``Pipeline`` to find ancestors in.
+        children: the iterable containing ``Node``s to find ancestors of.
+        catalog: the ``DataCatalog`` of the run.
+
+    Returns:
+        A set containing first persistent ancestors of the given
+        ``Node``s.
+
+    """
+    ancestor_nodes_to_run = set()
+    queue, visited = deque(children), set()
+    while queue:
+        current_node = queue.popleft()
+        visited.add(current_node)
+        if _has_persistent_inputs(current_node, catalog):
+            ancestor_nodes_to_run.add(current_node)
+            continue
+        for parent in _enumerate_parents(pipeline, current_node):
+            if parent not in visited:
path: kedro/runner/runner.py
text: @user1:
Could there be a case where the parent node hasn't been visited but is already in the queue?

@author:
This is a very good question. The only scenario I can see is that the queue is initialised with a group of nodes where one is the parent of another, but I don't think the calling code allows for this.

Initialising `visited` with the same nodes as the queue would avoid this ever becoming a problem.

@user1:
yeah that's probably very unlikely. I'm happy to leave it as is.

diff_hunk: @@ -278,3 +279,77 @@ def test_confirms(self, mocker, pipeline, is_async):
         catalog = DataCatalog(data_sets={"ds1": fake_dataset_instance})
         SequentialRunner(is_async=is_async).run(pipeline, catalog)
         fake_dataset_instance.confirm.assert_called_once_with()
+
+
+@pytest.fixture
+def resume_scenario_pipeline():
path: tests/runner/test_sequential_runner.py
text: @user1:
Should we define this in conftest where the other pipeline fixtures are? Would be good to give it a more descriptive name also like `X_shaped_pipeline` (c.f. `branchless_pipeline`) and maybe a short docstring describing or picturing the structure (like `branchless_no_input_pipeline` has).

@author:
I think you're right about this. I modified the names and moved my fixtures to conftest. I also moved the remaining fixtures in that file to conftest, in the spirit of leaving the codebase cleaner than when I found it. The remaining fixtures were from 2019 and came from a single contributor, while the more modern fixtures were all defined in conftest and were from multiple contributors, which is a good sign to me that this is the better approach.

These changes are made in [069a6fe](https://github.com/kedro-org/kedro/pull/1795/commits/069a6feede3bf53d7688aeb1a1511d58b9c3d3e9).

@user1:
Perfect, thank you! Definitely sounds like an improvement to the codebase.

diff_hunk: @@ -278,3 +279,77 @@ def test_confirms(self, mocker, pipeline, is_async):
         catalog = DataCatalog(data_sets={"ds1": fake_dataset_instance})
         SequentialRunner(is_async=is_async).run(pipeline, catalog)
         fake_dataset_instance.confirm.assert_called_once_with()
+
+
+@pytest.fixture
+def resume_scenario_pipeline():
+    return Pipeline(
+        [
+            node(identity, "ds0_A", "ds1_A", name="node1_A"),
+            node(identity, "ds0_B", "ds1_B", name="node1_B"),
+            node(
+                multi_input_list_output,
+                ["ds1_A", "ds1_B"],
+                ["ds2_A", "ds2_B"],
+                name="node2",
+            ),
+            node(identity, "ds2_A", "ds3_A", name="node3_A"),
+            node(identity, "ds2_B", "ds3_B", name="node3_B"),
+            node(identity, "ds3_A", "ds4_A", name="node4_A"),
+            node(identity, "ds3_B", "ds4_B", name="node4_B"),
+        ]
+    )
+
+
+@pytest.fixture
+def resume_scenario_catalog():
+    def _load():
+        return 0
+
+    def _save(arg):
+        assert arg == 0
path: tests/runner/test_sequential_runner.py
text: @user1:
Is this checking the `LambdaDataSet` storing the `0`? This seems to be a bit unnatural to do the assertion here, why do you need to test the `LambdaDataSet._load` here?

@author:
This is a holdover from the test I originally yanked the code from, my bad! Fixed in [e92a756](https://github.com/kedro-org/kedro/pull/1795/commits/e92a756fde94bbcac085c816859c449e694e518b).

@user1:
Ah! I see.