dask / dask/dask

Delayed function raises not implemented for dask dataframes passed as kwargs

Open
#11,949 1 comment 0 reactions 0 assignees View on GitHub
needs attention needs triage
Dominant language
Python
Stars
13.9k
Forks
2k
PR merge metrics
No merged PRs in 30d

Description

**Describe the issue**:

Delayed function raises not implemented error when parameters are passed as kwargs as opposed to positional args for dask dataframes.

This behavior is occurring from v2025.4.0+, works in v2025.3.0. Have not tested in older versions.

**Minimal Complete Verifiable Example**:

```python
from typing import Any

import dask
import pandas as pd
from dask import dataframe as dd
from dask import array as da

print(dask.__version__)

@dask.delayed
def foo(a) -> Any:
return a

n = 1
arr = [1, 2, 3]
df = pd.DataFrame({"a": [1, 2, 3]})
ddf = dd.from_pandas(df, npartitions=1)
darr = da.from_array([1, 2, 3])

print(f"{foo(n)=}") # <- returns Delayed('foo-...') object
print(f"{foo(a=n)=}") # <- returns Delayed('foo-...') object
print(f"{foo(arr)=}") # <- returns Delayed('foo-...') object
print(f"{foo(a=arr)=}") # <- returns Delayed('foo-...') object
print(f"{foo(darr)=}") # <- returns Delayed('foo-...') object
print(f"{foo(a=darr)=}") # <- returns Delayed('foo-...') object
print(f"{foo(df)=}") # <- returns Delayed('foo-...') object
print(f"{foo(a=df)=}") # <- returns Delayed('foo-...') object
print(f"{foo(ddf)=}") # <- returns Delayed('foo-...') object
print(f"{foo(a=ddf)=}") # <- raises NotImplementedError in 2025.4.0+
```

**Anything else we need to know?**:

traceback

```sh
Traceback (most recent call last):
File "/home/aaravind/Documents/Projects/PyProjects/scraps/scraps.py", line 32, in
print(f"{foo(a=ddf)=}") # <- raises NotImplementedError in 2025.4.0+
~~~^^^^^^^
File "/home/aaravind/Documents/Projects/PyProjects/scraps/.venv/lib/python3.13/site-packages/dask/delayed.py", line 846, in __call__
return call_function(
self._obj, self._key, args, kwargs, pure=self._pure, nout=self._nout
)
File "/home/aaravind/Documents/Projects/PyProjects/scraps/.venv/lib/python3.13/site-packages/dask/delayed.py", line 816, in call_function
dask_kwargs, collections2 = unpack_collections(kwargs)
~~~~~~~~~~~~~~~~~~^^^^^^^^
File "/home/aaravind/Documents/Projects/PyProjects/scraps/.venv/lib/python3.13/site-packages/dask/delayed.py", line 222, in unpack_collections
args, collections = unpack_collections(
~~~~~~~~~~~~~~~~~~^
[[k, v] for k, v in expr.items()], _return_collections=False
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
)
^
File "/home/aaravind/Documents/Projects/PyProjects/scraps/.venv/lib/python3.13/site-packages/dask/delayed.py", line 206, in unpack_collections
args, collections = utils.unzip(
~~~~~~~~~~~^
(unpack_collections(e, _return_collections=False) for e in expr), 2
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
)
^
File "/home/aaravind/Documents/Projects/PyProjects/scraps/.venv/lib/python3.13/site-packages/dask/utils.py", line 2298, in unzip
out = list(zip(*ls))
~~~^^^^^
File "/home/aaravind/Documents/Projects/PyProjects/scraps/.venv/lib/python3.13/site-packages/dask/delayed.py", line 207, in
(unpack_collections(e, _return_collections=False) for e in expr), 2
~~~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/aaravind/Documents/Projects/PyProjects/scraps/.venv/lib/python3.13/site-packages/dask/delayed.py", line 206, in unpack_collections
args, collections = utils.unzip(
~~~~~~~~~~~^
(unpack_collections(e, _return_collections=False) for e in expr), 2
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
)
^
File "/home/aaravind/Documents/Projects/PyProjects/scraps/.venv/lib/python3.13/site-packages/dask/utils.py", line 2298, in unzip
out = list(zip(*ls))
~~~^^^^^
File "/home/aaravind/Documents/Projects/PyProjects/scraps/.venv/lib/python3.13/site-packages/dask/delayed.py", line 207, in
(unpack_collections(e, _return_collections=False) for e in expr), 2
~~~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/aaravind/Documents/Projects/PyProjects/scraps/.venv/lib/python3.13/site-packages/dask/delayed.py", line 193, in unpack_collections
(name,) = expr.__dask_keys__()
~~~~~~~~~~~~~~~~~~^^
File "/home/aaravind/Documents/Projects/PyProjects/scraps/.venv/lib/python3.13/site-packages/dask/dataframe/dask_expr/_expr.py", line 94, in __dask_keys__
return [(self._name, i) for i in range(self.npartitions)]
^^^^^^^^^^^^^^^^
File "/home/aaravind/Documents/Projects/PyProjects/scraps/.venv/lib/python3.13/site-packages/dask/dataframe/dask_expr/_expr.py", line 446, in npartitions
return len(self.divisions) - 1
^^^^^^^^^^^^^^
File "/home/aaravind/.local/share/uv/python/cpython-3.13.3-linux-x86_64-gnu/lib/python3.13/functools.py", line 1026, in __get__
val = self.func(instance)
File "/home/aaravind/Documents/Projects/PyProjects/scraps/.venv/lib/python3.13/site-packages/dask/dataframe/dask_expr/_expr.py", line 431, in divisions
return tuple(self._divisions())
~~~~~~~~~~~~~~~^^
File "/home/aaravind/Documents/Projects/PyProjects/scraps/.venv/lib/python3.13/site-packages/dask/dataframe/dask_expr/_expr.py", line 434, in _divisions
raise NotImplementedError()
NotImplementedError
```

**Environment**:

- Dask version: 2025.4.0+
- Python version: 3.13.3
- Operating System: Ubuntu 24.04.2 LTS (WSL)
- Install method (conda, pip, source): pip(uv)

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.