Repository: kedro-org/kedro
Pull request: Consistent node execution order by sorting node with `Sequentialrunner`
Signed-off-by: Nok Chan <nok.lam.chan@quantumblack.com>

## Description
<!-- Why was this PR created? -->
#1350 

## Development notes
<!-- What have you changed, and how has this been tested? -->
The trickiest part of the PR is actually the test. It's not straightforward to reproduce the random behaviour. If you just create 10 pipelines in the same python session. It is actually deterministic, which is different from doing 10 `kedro run`. 

After some investigation, I believe this is the whole story
1. Set is not "random", but "arbitrary ordered", at least for CPython
2. `Node` has a custom `__hash__` function, which actually dictated how `set` order them.
3. In the same python session, the hash will be identical, even if they are different instances because how the `__hash__` is implemented. Therefore when

```python
  def __hash__(self):
      return hash(self._unique_key)
```
The way to test this properly is likely just to do some mocking (I still need to think about how to do this)

`_topologically_sorted` is the main function that gets changed, it changes the order of the arbitrary nodes to a deterministic order. The node dependencies are still respected. It only has effects when nodes don't have dependencies on each other. The new implementation will give a deterministic order to these nodes, so the run's results will be consistent.

```python
  def __lt__(self, other):
      if not isinstance(other, Node):
          return NotImplemented
      return self._unique_key < other._unique_key
```

# How this is tested
In the end, I go with manual testing, as turns out it is very tricky to test this non-deterministic behavior due to `hash`.

For reference, this is the test that I have written originally to test out this behavior, which should help understanding what this PR is trying to do. 

```python
    def test_pipeline_consistent_nodes_order(self, mocker):
        """
            Pipeline that have multiple possible execution orders should have consistent
            solutions
            Possible Solutions:
            1. A -> B -> C -> D -> E -> F
            2. B -> A -> C -> D -> E -> F
            3 ... Any permutation as long as F is executed last.

            Although we are not sure which permutation it is, but it should always output
            the same permutation.

            A-- \
            B--- \
            C---- F
            D--- /
            E-- /
        """

        def multiconcat(*args):
            return sum(args)

        mock_hash = mocker.patch(f"{__name__}.Node.__hash__", lambda x: random.int())
        expected_sorted_nodes: List[List[Node]] = None

        # Repeat 10 times so we can be sure it is not purely by chance
        for _ in range(10):

            inverted_fork_dags = Pipeline(
                [
                    node(constant_output, None, "A"),  # 3 4 1 0
                    node(constant_output, None, "B"),  # 9 4 1 4
                    node(constant_output, None, "C"),  # 2
                    node(constant_output, None, "D"),  # 6
                    node(constant_output, None, "E"),
                    node(multiconcat, ["A", "B", "C", "D", "E"], "F"),
                ]
            )
            if not expected_sorted_nodes:
                expected_sorted_nodes = inverted_fork_dags.nodes

            else:

                assert expected_sorted_nodes == inverted_fork_dags.nodes
```

## Checklist

- [ ] Read the [contributing](https://github.com/kedro-org/kedro/blob/main/CONTRIBUTING.md) guidelines
- [ ] 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
- [ ] Added tests to cover my changes


<a href="https://gitpod.io/#https://github.com/kedro-org/kedro/pull/1604"><img src="https://gitpod.io/button/open-in-gitpod.svg"/></a>



Problem statement:
Consistent nodes execution order with `SequentialRunner`
## Description
I'm always frustrated when I execute an identical pipeline it can yield different results. The root cause of this issue is due to the fact that multiple combinations of nodes exists. Kedro only try to solve the DAGs by finding 1 possible solution, but it is not guaranteed to be the same. 

One workaround is to specify input/output of nodes to make sure there is only 1 possible solution, but this is not ideal as users has to maintain arbitrary dummy variables.

### What is missing in the DAGs?
Seed of random number generator. Consider a simple pipeline with 3 nodes:
```
A-- \    
      \   
       C   
      /   
    /   
B--
```
In this pipeline, there are 2 possible execution order with SequentialRunner, 1. A->B->C, 2. B->A->C. Although there are no strong preference whether 1/2 is better, it is better to stick with one of them, as the output can be changed. 

## Context
In data science/machine learning pipeline, setting a seed to ensure **reproducible** result are very common, and currently there are no easy way to achieve this. 

## Possible Implementation
Ensure the resolved nodes are sorted so it always run in the same order with SequentialRunner.

## Possible Alternatives
(Optional) Describe any alternative solutions or features you've considered.

Hints from issue discussion:
So on this - we use the `toposort~=1.5` external library, since Kedro was released there is now a stdlib https://docs.python.org/3/library/graphlib.html module that does the same thing. If we do look into making this deterministic, it might be a good opportunity to adopt this too.
Note: currently we are using a function called `toposort`, but actually there is a function call `toposort_flatten` will guarantee the node order
@noklam awesome!
This is interesting - I'd never thought that individual node ordering would make any difference in practice so long as things are toposorted, even if that sort isn't deterministic.

@noklam just for my understanding, can you give a concrete example of something you might do where changing node execution order can alter the final output? Your diagram of nodes didn't come out correctly so it's a bit hard to understand.
I'll just add users do ask for this maybe once every two months, I've even seen people introduce fake nodes to force ordering.
@AntonyMilneQB 
This has to deal with the fact that the random number generator do have an specific order.

Imagine a simple pipeline like this
```
train_test_split_data-- \    
                         \   
                         train_model   
                         /   
                        /   
Model initialization--
```
Model initialization and train_test_split is 2 independent node, where it doesn't really matter who goes first. However, it is a common practice to set random seed to ensure the results are identical.

If we try to do kedro run and unfold the execution order, these are 2 possible scenraio.
Case 1
```
np.random.seed(2022)
model = model_initialization()
x_train, x_val, y_train, y_val = train_test_split(data)
train_model(x_train, x_val, y_train, y_val, model)
```

Case 2
```
np.random.seed(2022)
x_train, x_val, y_train, y_val = train_test_split(data)
model = model_initialization()
train_model(x_train, x_val, y_train, y_val, model)
```
Neither Case 1 / Case 2 is preferred, but one would preferred it is always Case1 or always Case 2 instead of bouncing between the 2.
@noklam Got it, thanks! I guess I usually pass `random_seed` into each function explicitly so didn't notice this before, but what you say makes perfect sense - it would be better if node execution order was completely deterministic.
Just seeing this. I'm not sure I like the idea of making any guarantees of deterministic node execution order, if people will start relying on it for anything beyond tests.

1. Guaranteed execution order only makes sense for `SequentialRunner`.
2. In a distributed/parallelized setting, any decision as to which node you run first could be made by a smart scheduler rather than user definition.

edit: Just to be clear, I'm fine with it being deterministic, but question making a guarantee that it would be deterministic. No code should depend on this. Trust me when I say, some users will find a way to depend on guaranteed execution order--I'm pretty sure I've seen requests for this sort of functionality--so I would go so far as to put a big warning to not depend on it lol.
The deterministic pipeline will work for `SequentialRunner` only indeed.

I would argue it is useful for the reproducible pipeline. My question would be what are the benefit of having random execution order with `SequentialRunner`? 

Currently, the only way to achieve this is either
1. dummy input/output
2. Reset seed on every node
I had exactly the same problem, that my machine learning pipeline returned different results with runs having different orders of nodes, some of which involved random seed settings.

A workaround solution is to pass a dummy I/O object across nodes, however, that is not a good solution in terms of coding design. (And what if there are many nodes invloved?)

I wonder if there is any follow up on this issue. Thank you!

Commit message: Merge branch 'main' into feat/consistent_node_execution_order_with_sequential_runner

Patch to review:
diff --git a/RELEASE.md b/RELEASE.md
index ae71953088..e40bae5552 100644
--- a/RELEASE.md
+++ b/RELEASE.md
@@ -25,9 +25,11 @@
 * Reduced number of log lines by changing the logging level from `INFO` to `DEBUG` for low priority messages.
 * Kedro's framework-side logging configuration no longer performs file-based logging. Hence superfluous `info.log`/`errors.log` files are no longer created in your project root, and running Kedro on read-only file systems such as Databricks Repos is now possible.
 * The `root` logger is now set to the Python default level of `WARNING` rather than `INFO`. Kedro's logger is still set to emit `INFO` level messages.
+* Kedro pipeline will have consistent execution order given the same set of nodes when using with `SequentialRunner`.
 * `kedro jupyter notebook/lab` no longer reuses a Jupyter kernel.
 * Required `cookiecutter>=2.1.1` to address a [known command injection vulnerability](https://security.snyk.io/vuln/SNYK-PYTHON-COOKIECUTTER-2414281).
 
+
 ## Upcoming deprecations for Kedro 0.19.0
 * `kedro.extras.ColorHandler` will be removed in 0.19.0.
 
diff --git a/kedro/pipeline/pipeline.py b/kedro/pipeline/pipeline.py
index dff0b214d7..637b6975dd 100644
--- a/kedro/pipeline/pipeline.py
+++ b/kedro/pipeline/pipeline.py
@@ -352,7 +352,7 @@ def nodes(self) -> List[Node]:
         return list(chain.from_iterable(self._topo_sorted_nodes))
 
     @property
-    def grouped_nodes(self) -> List[Set[Node]]:
+    def grouped_nodes(self) -> List[List[Node]]:
         """Return a list of the pipeline nodes in topologically ordered groups,
         i.e. if node A needs to be run before node B, it will appear in an
         earlier group.
@@ -870,7 +870,7 @@ def _validate_transcoded_inputs_outputs(nodes: List[Node]) -> None:
         )
 
 
-def _topologically_sorted(node_dependencies) -> List[Set[Node]]:
+def _topologically_sorted(node_dependencies) -> List[List[Node]]:
     """Topologically group and sort (order) nodes such that no node depends on
     a node that appears in the same or a later group.
 
@@ -894,7 +894,11 @@ def _circle_error_message(error_data: Dict[str, str]) -> str:
         return f"Circular dependencies exist among these items: {circular}"
 
     try:
-        return list(toposort(node_dependencies))
+        result = []
+        for dependencies in toposort(node_dependencies):
+            # Sort it so it has consistent order when run with SequentialRunner
+            result.append(sorted(dependencies))
+        return result
     except ToposortCircleError as exc:
         message = _circle_error_message(exc.data)
         raise CircularDependencyError(message) from exc
diff --git a/tests/pipeline/test_pipeline.py b/tests/pipeline/test_pipeline.py
index 5b5cfbecd2..2065bd706e 100644
--- a/tests/pipeline/test_pipeline.py
+++ b/tests/pipeline/test_pipeline.py
@@ -1,10 +1,13 @@
+import random
 import re
 from itertools import chain
+from typing import List
 
 import pytest
 
 import kedro
 from kedro.pipeline import Pipeline, node
+from kedro.pipeline.node import Node
 from kedro.pipeline.pipeline import (
     CircularDependencyError,
     ConfirmNotUniqueError,
@@ -253,8 +256,9 @@ def test_grouped_nodes(self, input_data):
         grouped = pipeline.grouped_nodes
         # Flatten a list of grouped nodes
         assert pipeline.nodes == list(chain.from_iterable(grouped))
-        # Check each grouped node matches with expected group
-        assert all(g == e for g, e in zip(grouped, expected))
+        # Check each grouped node matches with the expected group, the order is
+        # non-deterministic, so we are only checking they have the same set of nodes.
+        assert all(set(g) == e for g, e in zip(grouped, expected))
 
     def test_free_input(self, input_data):
         nodes = input_data["nodes"]
@@ -588,6 +592,52 @@ def test_connected_pipeline(self, disjoint_pipeline):
         assert len(pipeline.inputs()) == 1
         assert len(pipeline.outputs()) == 1
 
+    def test_pipeline_consistent_nodes_order(self, mocker):
+        """
+            Pipeline that have multiple possible execution orders should have consistent
+            solutions
+            Possible Solutions:
+            1. A -> B -> C -> D -> E -> F
+            2. B -> A -> C -> D -> E -> F
+            3 ... Any permutation as long as F is executed last.
+
+            Although we are not sure which permutation it is, but it should always output
+            the same permutation.
+
+            A-- \
+            B--- \
+            C---- F
+            D--- /
+            E-- /
+        """
+
+        def multiconcat(*args):
+            return sum(args)
+
+        mock_hash = mocker.patch(f"{__name__}.Node.__hash__")
+        expected_sorted_nodes: List[List[Node]] = None
+
+        # Repeat 10 times so we can be sure it is not purely by chance
+        for _ in range(10):
+            mock_hash.return_value = random.randint(1, 1e20)
+
+            inverted_fork_dags = Pipeline(
+                [
+                    node(constant_output, None, "A"),
+                    node(constant_output, None, "B"),
+                    node(constant_output, None, "C"),
+                    node(constant_output, None, "D"),
+                    node(constant_output, None, "E"),
+                    node(multiconcat, ["A", "B", "C", "D", "E"], "F"),
+                ]
+            )
+            if not expected_sorted_nodes:
+                expected_sorted_nodes = inverted_fork_dags.nodes
+
+            else:
+
+                assert expected_sorted_nodes == inverted_fork_dags.nodes
+
 
 class TestPipelineDescribe:
     def test_names_only(self, str_node_inputs_list):
diff --git a/tests/pipeline/test_pipeline_with_transcoding.py b/tests/pipeline/test_pipeline_with_transcoding.py
index a2936a8e3e..ea4fcc2020 100644
--- a/tests/pipeline/test_pipeline_with_transcoding.py
+++ b/tests/pipeline/test_pipeline_with_transcoding.py
@@ -121,7 +121,7 @@ def test_grouped_nodes(self, input_data):
         # Flatten a list of grouped nodes
         assert pipeline.nodes == list(chain.from_iterable(grouped))
         # Check each grouped node matches with expected group
-        assert all(g == e for g, e in zip(grouped, expected))
+        assert all(set(g) == e for g, e in zip(grouped, expected))
 
     def test_free_input(self, input_data):
         nodes = input_data["nodes"]


Code review comments:
diff_hunk: @@ -588,6 +592,52 @@ def test_connected_pipeline(self, disjoint_pipeline):
         assert len(pipeline.inputs()) == 1
         assert len(pipeline.outputs()) == 1
 
+    def test_pipeline_consistent_nodes_order(self, mocker):
+        """
+            Pipeline that have multiple possible execution orders should have consistent
+            solutions
+            Possible Solutions:
+            1. A -> B -> C -> D -> E -> F
+            2. B -> A -> C -> D -> E -> F
+            3 ... Any permutation as long as F is executed last.
+
+            Although we are not sure which permutation it is, but it should always output
+            the same permutation.
+
+            A-- \
+            B--- \
+            C---- F
+            D--- /
+            E-- /
+        """
+
+        def multiconcat(*args):
+            return sum(args)
+
+        mock_hash = mocker.patch(f"{__name__}.Node.__hash__")
+        expected_sorted_nodes: List[List[Node]] = None
+
+        # Repeat 10 times so we can be sure it is not purely by chance
+        for _ in range(10):
+            mock_hash.return_value = random.randint(1, 1e20)
path: tests/pipeline/test_pipeline.py
text: @user1:
I don't think this is doing what you want it to do. This is currently fixing the hash of every `Node` instance in this pipeline to be the same. We don't want `hash(node1)` to be the same as `hash(node2)`. What we should have is:
```
        n1 = node(constant_output, None, "A")
        n2 = node(constant_output, None, "B")
        n3 = node(constant_output, None, "C")
        n4 = node(constant_output, None, "D")
        n5 = node(constant_output, None, "E")
        n6 = node(multiconcat, ["A", "B", "C", "D", "E"], "F")

       # You don't have to nest these, you can put them all in one with block
       # But actually for Python < 3.10 it's super ugly formatting still: https://stackoverflow.com/questions/3024925/create-a-with-block-on-several-context-managers
        for _ in range(10):
            with mock.patch.object(n1, "__hash__", random.randint(1, 1e20)):
                with mock.patch.object(n2, "__hash__", random.randint(1, 1e20)):
                    with mock.patch.object(n3, "__hash__", random.randint(1, 1e20)):
                        with mock.patch.object(n4, "__hash__", random.randint(1, 1e20)):
                            with mock.patch.object(n5, "__hash__", random.randint(1, 1e20)):
                                with mock.patch.object(n6, "__hash__", random.randint(1, 1e20)):
                                    inverted_fork_dags = Pipeline([n1, n2, n3, n4, n5, n6])
                                    # use inverted_fork_dags.nodes
```
... but this is still not a great test because the current code in `main` doesn't fail it 😬 

After spending a looooong time playing around with this, I think it might just not be worth writing a test for it at all all... So long as it works as it should in manual testing then I think we're fine.

Happy to explain more about what I discovered while playing around with the testing here if you want to hear. It's certainly a tricky one.

@author:
Let's chat tomorrow!

diff_hunk: @@ -588,6 +592,52 @@ def test_connected_pipeline(self, disjoint_pipeline):
         assert len(pipeline.inputs()) == 1
         assert len(pipeline.outputs()) == 1
 
+    def test_pipeline_consistent_nodes_order(self, mocker):
+        """
+            Pipeline that have multiple possible execution orders should have consistent
+            solutions
+            Possible Solutions:
+            1. A -> B -> C -> D -> E -> F
+            2. B -> A -> C -> D -> E -> F
+            3 ... Any permutation as long as F is executed last.
+
+            Although we are not sure which permutation it is, but it should always output
+            the same permutation.
+
+            A-- \
+            B--- \
+            C---- F
+            D--- /
+            E-- /
+        """
+
+        def multiconcat(*args):
+            return sum(args)
path: tests/pipeline/test_pipeline.py
text: @user1:
Doesn't really matter because this function isn't actually executed anywhere... but this won't actually work if you try to run it:
```
TypeError: unsupported operand type(s) for +: 'int' and 'str'
```
Because Python initialises the sum with 0, so it tries to do 0 + "A". The correct thing to do would be `sum(args, "")` BUT even then it won't work:
```
TypeError: sum() can't sum strings [use ''.join(seq) instead]
```

So what you really want here is `"".join(args)`.