chore: bump Ray to 2.9.0 (#356)

# The problem

New Ray version was released. It seems to include a fix for handling big
workflow graphs.

# This PR's solution

Bumps Ray.

# Checklist

_Check that this PR satisfies the following items:_

- [x] Tests have been added for new features/changed behavior (if no new
features have been added, check the box).
- [x] The [changelog file](CHANGELOG.md) has been updated with a
user-readable description of the changes (if the change isn't visible to
the user in any way, check the box).
- [x] The PR's title is prefixed with
`<feat/fix/chore/imp[rovement]/int[ernal]/docs>[!]:`
- [x] The PR is linked to a JIRA ticket (if there's no suitable ticket,
check the box).

---------

Co-authored-by: Sebastian Morawiec <[email protected]>

CHANGELOG.md

@@ -19,6 +19,8 @@
 
 💅 *Improvements*
 
+* Bumped Ray to 2.9.0.
+
 🥷 *Internal*
 
 📃 *Docs*

setup.cfg

@@ -84,7 +84,7 @@ where = src
 
 [options.extras_require]
 ray =
-    ray[default]==2.6.3
+    ray[default]==2.9.0
     async-timeout # Ray depends on this, but its missing from ray requirements. Can be removed when https://github.com/ray-project/ray/issues/41267 is fixed.
     pyarrow # Ray workflows depend on this, but it's missing from ray requirements.
 all =
@@ -124,6 +124,7 @@ dev =
     pytest-cov>=2.12
     pytest-dependency
     pytest_httpserver
+    pytest-timeout>=2.0.0
     responses~=0.20,!=0.24
     ruff
     scikit-learn

tests/runtime/ray/test_integration.py

@@ -1300,9 +1300,43 @@ def wf():
 
         # When
         # The function-under-test is called inside the workflow.
-        wf_run_id = runtime.create_workflow_run(wf_model, None, False)
+        wf_run_id = runtime.create_workflow_run(wf_model, project=None, dry_run=False)
         _wait_to_finish_wf(wf_run_id, runtime)
 
         # Precondition
         wf_run = runtime.get_workflow_run_status(wf_run_id)
         assert wf_run.status.state == State.SUCCEEDED
+
+
+@pytest.mark.slow
+class TestGraphComplexity:
+    """
+    Ray <2.9 had an issue where certain workflows with high graph complexity would
+    evaluate exponentially long. This test ensures that we can resolve high-complexity
+    graph workflows in reasonable time.
+    """
+
+    @pytest.mark.timeout(20)
+    def test_high_graph_complexity_workflow(self, runtime: _dag.RayRuntime):
+        @sdk.task
+        def generic_task(*args):
+            return 42
+
+        @sdk.workflow
+        def wf():
+            reduced = generic_task()
+            for _ in range(10):
+                fanned_out = [generic_task(reduced) for _ in range(10)]
+                reduced = generic_task(*fanned_out)
+            return reduced
+
+        wf_model = wf().model
+
+        # When
+        # The function-under-test is called inside the workflow.
+        wf_run_id = runtime.create_workflow_run(wf_model, project=None, dry_run=False)
+
+        _wait_to_finish_wf(wf_run_id, runtime)
+        # Precondition
+        wf_run = runtime.get_workflow_run_status(wf_run_id)
+        assert wf_run.status.state == State.SUCCEEDED