Fix delayed in fusing with multipled dependencies

[phofl]

xref dask/dask#11067

[fjetter]

can you help me out...? 10 x load + 10 x lambda = 20. What am I missing?

[fjetter]

if you want to test that it doesn't fuse, maybe

df.map_partitions(lambda x: None, meta={}).optimize(fuse=False).dask == df.map_partitions(lambda x: None, meta={}).optimize().dask

is a better test. or even

ddf = df.map_partitions(lambda x: None, meta={})
dsk_opt = ddf.optimize().dask
dsk_raw = ddf.lower_completely().dask

[phofl]

10 x an intermediate task representing the from_delayed thing that isn't fused anymore. Previously that was fused into the map_partitions thing as well. This is the tradeoff for now

[phofl]

I like that, adjusted

[jtilly]

Thank you for the PR!

I just wanted to flag that this PR doesn't (yet) change the behavior for the example that I provided in dask/dask#11067:

(dask) ➜  ~/dask (main) ✗ pip install git+https://github.com/phofl/dask-expr@11067
...
(dask) ➜  ~/dask (main) ✗ python example.py                                             
Loading chunk 9
Loading chunk 8
Loading chunk 7
Loading chunk 6
Loading chunk 5
Loading chunk 4
Loading chunk 3
Loading chunk 2
Loading chunk 1
Loading chunk 0
Storing chunk 9
Storing chunk 8
Storing chunk 7
Storing chunk 6
Storing chunk 5
Storing chunk 4
Storing chunk 3
Storing chunk 2
Storing chunk 1
Storing chunk 0
(dask) ➜  ~/dask (main) ✗ DASK_DATAFRAME__QUERY_PLANNING=False python example.py                     
Loading chunk 9
Storing chunk 9
Loading chunk 8
Storing chunk 8
Loading chunk 7
Storing chunk 7
Loading chunk 6
Storing chunk 6
Loading chunk 5
Storing chunk 5
Loading chunk 4
Storing chunk 4
Loading chunk 3
Storing chunk 3
Loading chunk 2
Storing chunk 2
Loading chunk 1
Storing chunk 1
Loading chunk 0
Storing chunk 0

[phofl]

Yes we still have some overhead. It works mostly as expected if your delayed function runs longer (e.g. increase the size of your DataFrame), but this will still need a follow up on our end since the behaviour isn't as intended yet.

On a side note: You might want to use from_map if your delayed function is indeed only loading data from disk, that is better suited for your task anyway and doesn't suffer from this problem.