Repository: dask/dask
Pull request #11739 (2025-02-12 19:58:57)

Fix graph creation in dask-expr to_delayed
- [ ] Closes #xxxx
- [ ] Tests added / passed
- [ ] Passes `pre-commit run --all-files`

Turns out the graph creation in dask-expr was stupid :( culling isn't really necessary but also doesn't hurt, so I added it back in

github-actions[bot]: ## Unit Test Results
_See [test report](https://dask.github.io/dask/test_report.html) for an extended history of previous test failures. This is useful for diagnosing flaky tests._

     15 files  ±0       15 suites  ±0   4h 11m 57s ⏱️ +16s
 17 829 tests ±0   16 548 ✅ ±0   1 281 💤 ±0  0 ❌ ±0 
216 466 runs  ±0  185 238 ✅  - 3  31 228 💤 +3  0 ❌ ±0 

Results for commit 5e2a8174. ± Comparison against base commit 0b1e23e6.

[test-results]:data:application/gzip;base64,H4sIAHMGrWcC/03Myw6DIBCF4VcxrLtgRgaGvkxDVBJSLw3Cqum7F61Fl/93kvMWPozDKu4N0K0Raw7prD5Hl8Iy7w1gCpU17bthtP9+rLnrNtSk+MRneG2IDNW8C2MxWWGIcYmHxDxv1whaaX10vWbCliv+rltArHa53vv63C3TFFIJQQM6BqOI0UGLrjdEVkorbUveaq08O9VrEJ8vKCb09RoBAAA=


Patch:
diff --git a/dask/dataframe/dask_expr/_collection.py b/dask/dataframe/dask_expr/_collection.py
index c52008b09..70a18d8a3 100644
--- a/dask/dataframe/dask_expr/_collection.py
+++ b/dask/dataframe/dask_expr/_collection.py
@@ -2502,13 +2502,17 @@ Expr={expr}"""
         --------
         dask_expr.from_delayed
         """
+        from dask.highlevelgraph import HighLevelGraph
+
         if optimize_graph:
             frame = self.optimize()
         else:
             frame = self
         keys = frame.__dask_keys__()
         graph = frame.__dask_graph__()
-        return [Delayed(k, graph) for k in keys]
+        layer = "delayed-" + frame._name
+        graph = HighLevelGraph.from_collections(layer, graph, dependencies=())
+        return [Delayed(k, graph, layer=layer) for k in keys]
 
     def to_backend(self, backend: str | None = None, **kwargs):
         """Move to a new DataFrame backend
diff --git a/dask/delayed.py b/dask/delayed.py
index e02b60699..1718ca88a 100644
--- a/dask/delayed.py
+++ b/dask/delayed.py
@@ -20,7 +20,7 @@ from dask.base import (
 )
 from dask.base import tokenize as _tokenize
 from dask.context import globalmethod
-from dask.core import quote
+from dask.core import flatten, quote
 from dask.highlevelgraph import HighLevelGraph
 from dask.optimization import fuse
 from dask.typing import Graph, NestedKeys
@@ -539,6 +539,7 @@ def optimize(dsk, keys, **kwargs):
 
     if not isinstance(dsk, HighLevelGraph):
         dsk = HighLevelGraph.from_collections(id(dsk), dsk, dependencies=())
+    dsk = dsk.cull(set(flatten(keys)))
     return dsk
 
 


Test patch:
diff --git a/dask/dataframe/dask_expr/io/tests/test_delayed.py b/dask/dataframe/dask_expr/io/tests/test_delayed.py
index bc63534d9..42508dcd3 100644
--- a/dask/dataframe/dask_expr/io/tests/test_delayed.py
+++ b/dask/dataframe/dask_expr/io/tests/test_delayed.py
@@ -14,6 +14,7 @@ pd = _backend_library()
 def test_from_delayed_optimizing():
     parts = from_dict({"a": np.arange(300)}, npartitions=30).to_delayed()
     result = from_delayed(parts[0], meta=pd.DataFrame({"a": pd.Series(dtype=np.int64)}))
+    assert len(result.optimize().dask) == 2
     assert_eq(result, pd.DataFrame({"a": pd.Series(np.arange(10))}))
 
 
diff --git a/dask/tests/test_delayed.py b/dask/tests/test_delayed.py
index 124c53fda..4ce1dbc8e 100644
--- a/dask/tests/test_delayed.py
+++ b/dask/tests/test_delayed.py
@@ -331,8 +331,8 @@ def test_common_subexpressions():
 def test_delayed_optimize():
     x = Delayed("b", {"a": 1, "b": (inc, "a"), "c": (inc, "b")})
     (x2,) = dask.optimize(x)
-    # Delayed's __dask_optimize__ does not cull out 'c'
-    assert sorted(x2.dask.keys()) == ["a", "b", "c"]
+    # Delayed's __dask_optimize__ culls out 'c'
+    assert sorted(x2.dask.keys()) == ["a", "b"]
     assert x2._layer != x2._key
     # Optimize generates its own layer name, which doesn't match the key.
     # `Delayed._rebuild` handles this.
@@ -843,7 +843,7 @@ def test_annotations_survive_optimization():
     (d_opt,) = dask.optimize(d)
     assert type(d_opt.dask) is HighLevelGraph
     assert len(d_opt.dask.layers) == 1
-    assert len(d_opt.dask.layers["b"]) == 3  # c is not culled
+    assert len(d_opt.dask.layers["b"]) == 2  # c is culled
     assert d_opt.dask.layers["b"].annotations == {"foo": "bar"}