[BUG] map_partition of a delayed DataFrame partition

[madsbk]

What happened:
Get a cancel error:

concurrent.futures._base.CancelledError: ('f-0f531032308c06167e4aa2b8add1e324', 1)

What you expected to happen:
Get the result of mapping a DataFrame on the function f

Minimal Complete Verifiable Example:

import pandas as pd
import dask
import dask.dataframe as dd
from distributed import Client, LocalCluster, wait

def inc(x):
    return x + 1

def f(df, x):
    print("f working?: ", type(x))
    return x

def main(client):
    delayed_df = dd.from_pandas(
        pd.DataFrame({"a": range(5)}), npartitions=2
    ).to_delayed()
    dl = delayed_df[0].persist()
    # Replaceing the `delayed_df` with the more simple delayed `inc`, everything works
    # dl = dask.delayed(inc)(42)
    wait(dl)

    df: dd.DataFrame = dd.from_pandas(pd.DataFrame({"a": range(5)}), npartitions=2)
    df: dd.DataFrame = df.map_partitions(f, dl, meta=df._meta)
    # print("main() - df.dask.layers: ", df.dask.layers)
    df = df.persist(optimize_graph=False)
    df.compute()

if __name__ == "__main__":
    with LocalCluster(
        scheduler_port=0, asynchronous=False, n_workers=1, nthreads=1, processes=True
    ) as cluster:
        with Client(cluster, asynchronous=False) as client:
            main(client)

Am I doing something stupid here? Doing dl = delayed_df[0].persist() to get a delayed handle to the first partition of df should be fine right?

[mrocklin]

I ran this locally and got the following:

f working?:  <class 'tuple'>
f working?:  <class 'tuple'>

If this is the correct response then maybe it's an environment issue?

[madsbk]

If this is the correct response then maybe it's an environment issue?

Maybe, are you running the main branch?

[mrocklin]

Ah, indeed, after updating things fail as you report

Traceback (most recent call last):
  File "/Users/mrocklin/foo.py", line 33, in <module>
    main(client)
  File "/Users/mrocklin/foo.py", line 26, in main
    df.compute()
  File "/Users/mrocklin/workspace/dask/dask/base.py", line 285, in compute
    (result,) = compute(self, traverse=False, **kwargs)
  File "/Users/mrocklin/workspace/dask/dask/base.py", line 567, in compute
    results = schedule(dsk, keys, **kwargs)
  File "/Users/mrocklin/workspace/distributed/distributed/client.py", line 2707, in get
    results = self.gather(packed, asynchronous=asynchronous, direct=direct)
  File "/Users/mrocklin/workspace/distributed/distributed/client.py", line 2021, in gather
    return self.sync(
  File "/Users/mrocklin/workspace/distributed/distributed/client.py", line 862, in sync
    return sync(
  File "/Users/mrocklin/workspace/distributed/distributed/utils.py", line 326, in sync
    raise exc.with_traceback(tb)
  File "/Users/mrocklin/workspace/distributed/distributed/utils.py", line 309, in f
    result[0] = yield future
  File "/Users/mrocklin/miniconda/envs/dev/lib/python3.9/site-packages/tornado/gen.py", line 762, in run
    value = future.result()
  File "/Users/mrocklin/workspace/distributed/distributed/client.py", line 1887, in _gather
    raise exc
concurrent.futures._base.CancelledError: ('f-5f1c792d74e96aae96b521f1d0f02c0d', 1)

Probably the next thing to do here is to git bisect. Is this something that you have time to do?

[jrbourbeau]

git bisect pointed me to dask/dask#7734 as the change where we started getting the CancelledError

[madsbk]

git bisect pointed me to dask/dask#7734 as the change where we started getting the CancelledError

Thanks @jrbourbeau, the problem was in that PR: dask/dask#7838

[jrbourbeau]

Closed via dask/dask#7838