We can't properly deal with delayed inputs

[phofl]

Every argument could theoretically be a delayed object, similar to how every argument could be a dask-expr collection, we can't deal with this yet since we never check for them

I created a very naive implementation of a delayed expression _Delayed to capture some things, but we should think critically about it before moving ahead with this issue.

I suggest that we unpack delayed object similar to what we do with collections, in the constructor of Expr

[delucchi-cmu]

This introduced some big performance regressions in our framework that makes heavy use of @dask.delayed. We do some silly things like create 300k delayed objects so we have extreme control over joining dataframes from disparate sources.

We're getting around it for now by using the legacy dataframe directly.

[phofl]

Do you have a reproducer? I suspect that this is actually something different that we can address more easily

[delucchi-cmu]

Simple code snippet.py:

import numpy as np
import dask.dataframe as dd
import pandas as pd

if __name__ == "__main__":
    a = dd.from_dict({"a": np.arange(300_000)}, npartitions=30_000)
    parts = a.to_delayed()
    dd.from_delayed(
        parts[0], meta=pd.DataFrame.from_dict({"a": pd.Series(dtype=np.int64)})
    ).compute()

Starting from a fresh virtual environment:

>> pip install numpy pandas 'dask<=2024.2.0'
>> time python snippet.py

real	0m1.094s
user	0m1.435s
sys	0m1.383s

>> pip install 'dask>=2024.3.1'
>> time python snippet.py

real	0m1.127s
user	0m1.507s
sys	0m1.360s

>> pip install dask-expr
>> time python snippet.py

real	0m48.497s
user	0m48.845s
sys	0m1.381s

[phofl]

Thx, that's helpful

put up a fix here: #1048

[delucchi-cmu]

I suspect there is more than one operation that is being slowed by the addition of dask-expr. Using a similar reproducer as before:

    a = dd.from_dict({"a": np.arange(300_000)}, npartitions=300_000)
    parts = a.to_delayed(optimize_graph=False)

3x slowdown, unchanged by #1048

>> pip install numpy pandas 'dask<=2024.2.0'
>> time python snippet.py

real	0m5.882s
user	0m7.840s
sys	0m1.521s

>> pip install 'dask>=2024.3.1'
>> time python snippet.py

real	0m5.936s
user	0m7.882s
sys	0m1.554s

>> pip install dask-expr
Successfully installed dask-expr-1.0.14
>> time python snippet.py

real	0m20.619s
user	0m22.504s
sys	0m1.365s

>> pip install <dask-expr from source to include PR #1048>
>> time python snippet.py

real	0m20.285s
user	0m22.208s
sys	0m1.568s

[phofl]

Yeah this is a know limitation unfortunately. Going to delayed objects roundtrips through the legacy implementation, which materialises the graph. That causes the slowdown here. Improvements are certainly very welcome