Merge pull request #1736 from bnaul/patch-4

Respect order of inputs in merge

CHANGELOG.md

@@ -27,6 +27,7 @@ These changes are available in the [master branch](https://github.com/PrefectHQ/
 - Fix executor initialization missing `self` in `KubernetesJobEnvironment` - [#1713](https://github.com/PrefectHQ/prefect/pull/1713)
 - Fix `identifier_label` not being generated on each run for Kubernetes based environments - [#1718](https://github.com/PrefectHQ/prefect/pull/1718)
 - Fix issue where users could not override their user config path when deploying Docker to Cloud - [#1719](https://github.com/PrefectHQ/prefect/pull/1719)
+- Respect order of inputs in merge - [#1736](https://github.com/~/1736)
 
 ### Deprecations
 

src/prefect/tasks/control_flow/conditional.py

@@ -16,7 +16,12 @@ def __init__(self, **kwargs) -> None:
 
     def run(self, **task_results: Any) -> Any:
         return next(
-            (v for v in task_results.values() if not isinstance(v, NoResultType)), None
+            (
+                v
+                for k, v in sorted(task_results.items())
+                if not isinstance(v, NoResultType)
+            ),
+            None,
         )
 
 

tests/tasks/test_control_flow.py

@@ -222,6 +222,22 @@ def true_branch():
         assert state.result[merge_task].result == [1, 2]
 
 
+def test_merge_order():
+    @task
+    def x():
+        return "x"
+
+    @task
+    def y():
+        return "y"
+
+    with Flow(name="test") as flow:
+        merge_task = merge(x(), y())
+
+    state = flow.run()
+    assert state.result[merge_task].result == "x"
+
+
 class TestFilterTask:
     def test_empty_initialization(self):
         task = FilterTask()