UserWarning: dask_expr does not support the DataFrameIOFunction protocol for column projection. To enable column projection, please ensure that the signature of `func` includes a `columns=` keyword argument instead.
c = <Client: No scheduler connected>
s = <Scheduler 'tcp://127.0.0.1:34381', workers: 0, cores: 0, tasks: 0>
a = <Worker 'tcp://127.0.0.1:43517', name: 0, status: closed, stored: 0, running: 0/1, ready: 0, comm: 0, waiting: 0>
b = <Worker 'tcp://127.0.0.1:43929', name: 1, status: closed, stored: 0, running: 0/2, ready: 0, comm: 0, waiting: 0>

    @gen_cluster(client=True, scheduler_kwargs={"dashboard": True})
    async def test_shuffling(c, s, a, b):
        pytest.importorskip("pyarrow")
        dd = pytest.importorskip("dask.dataframe")
        ss = Shuffling(s)
    
>       df = dask.datasets.timeseries(
            start="2000-01-01",
            end="2000-02-01",
            dtypes={"x": float, "y": float},
            freq="10 s",
        )

distributed/dashboard/tests/test_scheduler_bokeh.py:1342: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
../../../miniconda3/envs/dask-distributed/lib/python3.9/site-packages/dask/datasets.py:63: in timeseries
    return make_timeseries(
../../../miniconda3/envs/dask-distributed/lib/python3.9/site-packages/dask/dataframe/io/demo.py:434: in make_timeseries
    return from_map(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

func = <dask.dataframe.io.demo.MakeDataframePart object at 0x7f5ccc0eb1f0>
args = None
meta =                    x         y
timestamp                     
2000-01-01  0.249765 -0.919819
divisions = [Timestamp('2000-01-01 00:00:00'), Timestamp('2000-01-02 00:00:00'), Timestamp('2000-01-03 00:00:00'), Timestamp('2000-01-04 00:00:00'), Timestamp('2000-01-05 00:00:00'), Timestamp('2000-01-06 00:00:00'), ...]
label = 'make-timeseries', enforce_metadata = False
iterables = [[([Timestamp('2000-01-01 00:00:00'), Timestamp('2000-01-02 00:00:00')], 1828363483), ([Timestamp('2000-01-02 00:00:00...1-06 00:00:00')], 559906344), ([Timestamp('2000-01-06 00:00:00'), Timestamp('2000-01-07 00:00:00')], 1366879366), ...]]
kwargs = {}
DataFrameIOFunction = <class 'dask.dataframe.io.utils.DataFrameIOFunction'>
FromMap = <class 'dask_expr.io.io.FromMap'>
FromMapProjectable = <class 'dask_expr.io.io.FromMapProjectable'>
lengths = {31}, i = 0

    def from_map(
        func,
        *iterables,
        args=None,
        meta=no_default,
        divisions=None,
        label=None,
        enforce_metadata=False,
        **kwargs,
    ):
        """Create a dask-expr collection from a custom function map
    
        NOTE: The underlying ``Expr`` object produced by this API
        will support column projection (via ``simplify``) if
        the ``func`` argument has "columns" in its signature.
    
        """
        from dask.dataframe.io.utils import DataFrameIOFunction
    
        from dask_expr.io import FromMap, FromMapProjectable
    
        if "token" in kwargs:
            # This option doesn't really make sense in dask-expr
            raise NotImplementedError("dask_expr does not support a token argument.")
    
        lengths = set()
        iterables = list(iterables)
        for i, iterable in enumerate(iterables):
            if not isinstance(iterable, Iterable):
                raise ValueError(
                    f"All elements of `iterables` must be Iterable, got {type(iterable)}"
                )
            try:
                lengths.add(len(iterable))
            except (AttributeError, TypeError):
                iterables[i] = list(iterable)
                lengths.add(len(iterables[i]))
        if len(lengths) == 0:
            raise ValueError("`from_map` requires at least one Iterable input")
        elif len(lengths) > 1:
            raise ValueError("All `iterables` must have the same length")
        if lengths == {0}:
            raise ValueError("All `iterables` must have a non-zero length")
    
        # Check if `func` supports column projection
        allow_projection = False
        columns_arg_required = False
        if param := inspect.signature(func).parameters.get("columns", None):
            allow_projection = True
            columns_arg_required = param.default is param.empty
            if meta is no_default and columns_arg_required:
                raise TypeError(
                    "Argument `func` of `from_map` has a required `columns` "
                    " parameter and not `meta` provided."
                    "Either provide `meta` yourself or make `columns` an optional argument."
                )
        elif isinstance(func, DataFrameIOFunction):
>           warnings.warn(
                "dask_expr does not support the DataFrameIOFunction "
                "protocol for column projection. To enable column "
                "projection, please ensure that the signature of `func` "
                "includes a `columns=` keyword argument instead."
            )
E           UserWarning: dask_expr does not support the DataFrameIOFunction protocol for column projection. To enable column projection, please ensure that the signature of `func` includes a `columns=` keyword argument instead.

../../../miniconda3/envs/dask-distributed/lib/python3.9/site-packages/dask_expr/_collection.py:4999: UserWarning

AssertionError: assert 0 == 5
 +  where 0 = <distributed.protocol.tests.test_highlevelgraph.ExampleAnnotationPlugin object at 0x7f5c9f0c41c0>.retry_matches
 +  and   5 = <dask_expr.expr.Series: expr=Shuffle(416dd28)['a'] + Shuffle(416dd28)['b']>.npartitions
c = <Client: No scheduler connected>
s = <Scheduler 'tcp://127.0.0.1:38949', workers: 0, cores: 0, tasks: 0>
a = <Worker 'tcp://127.0.0.1:44805', name: 0, status: closed, stored: 0, running: 0/1, ready: 0, comm: 0, waiting: 0>
b = <Worker 'tcp://127.0.0.1:45247', name: 1, status: closed, stored: 0, running: 0/2, ready: 0, comm: 0, waiting: 0>

    @gen_cluster(client=True)
    async def test_dataframe_annotations(c, s, a, b):
        retries = 5
        plugin = ExampleAnnotationPlugin(retries=retries)
        s.add_plugin(plugin)
    
        assert plugin in s.plugins.values()
    
        df = dd.from_pandas(
            pd.DataFrame(
                {"a": np.arange(10, dtype=int), "b": np.arange(10, 0, -1, dtype=float)}
            ),
            npartitions=5,
        )
        df = df.shuffle("a", max_branch=2)
        acol = df["a"]
        bcol = df["b"]
    
        with dask.annotate(retries=retries):
            df = acol + bcol
    
        with dask.config.set(optimization__fuse__active=False):
            rdf = await c.compute(df)
    
        assert rdf.dtypes == np.float64
        assert (rdf == 10.0).all()
    
        # There is an annotation match per partition (i.e. task)
>       assert plugin.retry_matches == df.npartitions
E       AssertionError: assert 0 == 5
E        +  where 0 = <distributed.protocol.tests.test_highlevelgraph.ExampleAnnotationPlugin object at 0x7f5c9f0c41c0>.retry_matches
E        +  and   5 = <dask_expr.expr.Series: expr=Shuffle(416dd28)['a'] + Shuffle(416dd28)['b']>.npartitions

distributed/protocol/tests/test_highlevelgraph.py:186: AssertionError

UserWarning: dask_expr does not support the DataFrameIOFunction protocol for column projection. To enable column projection, please ensure that the signature of `func` includes a `columns=` keyword argument instead.
client = <Client: 'tcp://127.0.0.1:36857' processes=2 threads=2, memory=31.21 GiB>

    def test_basic(client):
>       df = dd.demo.make_timeseries(freq="15D", partition_freq="30D")

distributed/shuffle/tests/test_graph.py:20: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
../../../miniconda3/envs/dask-distributed/lib/python3.9/site-packages/dask/dataframe/io/demo.py:434: in make_timeseries
    return from_map(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

func = <dask.dataframe.io.demo.MakeDataframePart object at 0x7f5c9d6e0ac0>
args = None
meta =                name    id         x         y
timestamp                                    
2000-01-01  Norbert  1016  0.731171  0.590149
divisions = [Timestamp('2000-01-01 00:00:00'), Timestamp('2000-01-31 00:00:00'), Timestamp('2000-03-01 00:00:00'), Timestamp('2000-03-31 00:00:00'), Timestamp('2000-04-30 00:00:00'), Timestamp('2000-05-30 00:00:00'), ...]
label = 'make-timeseries', enforce_metadata = False
iterables = [[([Timestamp('2000-01-01 00:00:00'), Timestamp('2000-01-31 00:00:00')], 720267087), ([Timestamp('2000-01-31 00:00:00'...5-30 00:00:00')], 827030214), ([Timestamp('2000-05-30 00:00:00'), Timestamp('2000-06-29 00:00:00')], 1648004478), ...]]
kwargs = {}
DataFrameIOFunction = <class 'dask.dataframe.io.utils.DataFrameIOFunction'>
FromMap = <class 'dask_expr.io.io.FromMap'>
FromMapProjectable = <class 'dask_expr.io.io.FromMapProjectable'>
lengths = {12}, i = 0

    def from_map(
        func,
        *iterables,
        args=None,
        meta=no_default,
        divisions=None,
        label=None,
        enforce_metadata=False,
        **kwargs,
    ):
        """Create a dask-expr collection from a custom function map
    
        NOTE: The underlying ``Expr`` object produced by this API
        will support column projection (via ``simplify``) if
        the ``func`` argument has "columns" in its signature.
    
        """
        from dask.dataframe.io.utils import DataFrameIOFunction
    
        from dask_expr.io import FromMap, FromMapProjectable
    
        if "token" in kwargs:
            # This option doesn't really make sense in dask-expr
            raise NotImplementedError("dask_expr does not support a token argument.")
    
        lengths = set()
        iterables = list(iterables)
        for i, iterable in enumerate(iterables):
            if not isinstance(iterable, Iterable):
                raise ValueError(
                    f"All elements of `iterables` must be Iterable, got {type(iterable)}"
                )
            try:
                lengths.add(len(iterable))
            except (AttributeError, TypeError):
                iterables[i] = list(iterable)
                lengths.add(len(iterables[i]))
        if len(lengths) == 0:
            raise ValueError("`from_map` requires at least one Iterable input")
        elif len(lengths) > 1:
            raise ValueError("All `iterables` must have the same length")
        if lengths == {0}:
            raise ValueError("All `iterables` must have a non-zero length")
    
        # Check if `func` supports column projection
        allow_projection = False
        columns_arg_required = False
        if param := inspect.signature(func).parameters.get("columns", None):
            allow_projection = True
            columns_arg_required = param.default is param.empty
            if meta is no_default and columns_arg_required:
                raise TypeError(
                    "Argument `func` of `from_map` has a required `columns` "
                    " parameter and not `meta` provided."
                    "Either provide `meta` yourself or make `columns` an optional argument."
                )
        elif isinstance(func, DataFrameIOFunction):
>           warnings.warn(
                "dask_expr does not support the DataFrameIOFunction "
                "protocol for column projection. To enable column "
                "projection, please ensure that the signature of `func` "
                "includes a `columns=` keyword argument instead."
            )
E           UserWarning: dask_expr does not support the DataFrameIOFunction protocol for column projection. To enable column projection, please ensure that the signature of `func` includes a `columns=` keyword argument instead.

../../../miniconda3/envs/dask-distributed/lib/python3.9/site-packages/dask_expr/_collection.py:4999: UserWarning

Failed: DID NOT RAISE <class 'TypeError'>
dtype = 'csingle'

    @pytest.mark.parametrize("dtype", ["csingle", "cdouble", "clongdouble"])
    def test_raise_on_complex_numbers(dtype):
        df = dd.from_pandas(
            pd.DataFrame({"x": pd.array(range(10), dtype=dtype)}), npartitions=5
        )
        with pytest.raises(
            TypeError, match=f"p2p does not support data of type '{df.x.dtype}'"
        ), dask.config.set({"dataframe.shuffle.method": "p2p"}):
>           df.shuffle("x")
E           Failed: DID NOT RAISE <class 'TypeError'>

distributed/shuffle/tests/test_graph.py:43: Failed

Failed: DID NOT RAISE <class 'TypeError'>
dtype = 'cdouble'

    @pytest.mark.parametrize("dtype", ["csingle", "cdouble", "clongdouble"])
    def test_raise_on_complex_numbers(dtype):
        df = dd.from_pandas(
            pd.DataFrame({"x": pd.array(range(10), dtype=dtype)}), npartitions=5
        )
        with pytest.raises(
            TypeError, match=f"p2p does not support data of type '{df.x.dtype}'"
        ), dask.config.set({"dataframe.shuffle.method": "p2p"}):
>           df.shuffle("x")
E           Failed: DID NOT RAISE <class 'TypeError'>

distributed/shuffle/tests/test_graph.py:43: Failed

Failed: DID NOT RAISE <class 'TypeError'>
dtype = 'clongdouble'

    @pytest.mark.parametrize("dtype", ["csingle", "cdouble", "clongdouble"])
    def test_raise_on_complex_numbers(dtype):
        df = dd.from_pandas(
            pd.DataFrame({"x": pd.array(range(10), dtype=dtype)}), npartitions=5
        )
        with pytest.raises(
            TypeError, match=f"p2p does not support data of type '{df.x.dtype}'"
        ), dask.config.set({"dataframe.shuffle.method": "p2p"}):
>           df.shuffle("x")
E           Failed: DID NOT RAISE <class 'TypeError'>

distributed/shuffle/tests/test_graph.py:43: Failed

Failed: DID NOT RAISE <class 'TypeError'>
def test_raise_on_sparse_data():
        df = dd.from_pandas(
            pd.DataFrame({"x": pd.array(range(10), dtype="Sparse[float64]")}), npartitions=5
        )
        with pytest.raises(
            TypeError, match="p2p does not support sparse data"
        ), dask.config.set({"dataframe.shuffle.method": "p2p"}):
>           df.shuffle("x")
E           Failed: DID NOT RAISE <class 'TypeError'>

distributed/shuffle/tests/test_graph.py:71: Failed

Failed: DID NOT RAISE <class 'TypeError'>
def test_raise_on_non_string_column_name():
        df = dd.from_pandas(pd.DataFrame({"a": range(10), 1: range(10)}), npartitions=5)
        with pytest.raises(
            TypeError, match="p2p requires all column names to be str"
        ), dask.config.set({"dataframe.shuffle.method": "p2p"}):
>           df.shuffle("a")
E           Failed: DID NOT RAISE <class 'TypeError'>

distributed/shuffle/tests/test_graph.py:79: Failed

UserWarning: dask_expr does not support the DataFrameIOFunction protocol for column projection. To enable column projection, please ensure that the signature of `func` includes a `columns=` keyword argument instead.
c = <Client: No scheduler connected>
s = <Scheduler 'tcp://127.0.0.1:45997', workers: 0, cores: 0, tasks: 0>
workers = (<Worker 'tcp://127.0.0.1:36755', name: 0, status: closed, stored: 0, running: 0/2, ready: 0, comm: 0, waiting: 0>, <W... 0>, <Worker 'tcp://127.0.0.1:42615', name: 3, status: closed, stored: 0, running: 0/2, ready: 0, comm: 0, waiting: 0>)

    @gen_cluster([("", 2)] * 4, client=True)
    async def test_basic_state(c, s, *workers):
>       df = dd.demo.make_timeseries(freq="15D", partition_freq="30D")

distributed/shuffle/tests/test_graph.py:90: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
../../../miniconda3/envs/dask-distributed/lib/python3.9/site-packages/dask/dataframe/io/demo.py:434: in make_timeseries
    return from_map(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

func = <dask.dataframe.io.demo.MakeDataframePart object at 0x7f5c9df95130>
args = None
meta =            name   id         x         y
timestamp                               
2000-01-01  Tim  983 -0.170822  0.418217
divisions = [Timestamp('2000-01-01 00:00:00'), Timestamp('2000-01-31 00:00:00'), Timestamp('2000-03-01 00:00:00'), Timestamp('2000-03-31 00:00:00'), Timestamp('2000-04-30 00:00:00'), Timestamp('2000-05-30 00:00:00'), ...]
label = 'make-timeseries', enforce_metadata = False
iterables = [[([Timestamp('2000-01-01 00:00:00'), Timestamp('2000-01-31 00:00:00')], 1185735830), ([Timestamp('2000-01-31 00:00:00...5-30 00:00:00')], 1323701270), ([Timestamp('2000-05-30 00:00:00'), Timestamp('2000-06-29 00:00:00')], 456587806), ...]]
kwargs = {}
DataFrameIOFunction = <class 'dask.dataframe.io.utils.DataFrameIOFunction'>
FromMap = <class 'dask_expr.io.io.FromMap'>
FromMapProjectable = <class 'dask_expr.io.io.FromMapProjectable'>
lengths = {12}, i = 0

    def from_map(
        func,
        *iterables,
        args=None,
        meta=no_default,
        divisions=None,
        label=None,
        enforce_metadata=False,
        **kwargs,
    ):
        """Create a dask-expr collection from a custom function map
    
        NOTE: The underlying ``Expr`` object produced by this API
        will support column projection (via ``simplify``) if
        the ``func`` argument has "columns" in its signature.
    
        """
        from dask.dataframe.io.utils import DataFrameIOFunction
    
        from dask_expr.io import FromMap, FromMapProjectable
    
        if "token" in kwargs:
            # This option doesn't really make sense in dask-expr
            raise NotImplementedError("dask_expr does not support a token argument.")
    
        lengths = set()
        iterables = list(iterables)
        for i, iterable in enumerate(iterables):
            if not isinstance(iterable, Iterable):
                raise ValueError(
                    f"All elements of `iterables` must be Iterable, got {type(iterable)}"
                )
            try:
                lengths.add(len(iterable))
            except (AttributeError, TypeError):
                iterables[i] = list(iterable)
                lengths.add(len(iterables[i]))
        if len(lengths) == 0:
            raise ValueError("`from_map` requires at least one Iterable input")
        elif len(lengths) > 1:
            raise ValueError("All `iterables` must have the same length")
        if lengths == {0}:
            raise ValueError("All `iterables` must have a non-zero length")
    
        # Check if `func` supports column projection
        allow_projection = False
        columns_arg_required = False
        if param := inspect.signature(func).parameters.get("columns", None):
            allow_projection = True
            columns_arg_required = param.default is param.empty
            if meta is no_default and columns_arg_required:
                raise TypeError(
                    "Argument `func` of `from_map` has a required `columns` "
                    " parameter and not `meta` provided."
                    "Either provide `meta` yourself or make `columns` an optional argument."
                )
        elif isinstance(func, DataFrameIOFunction):
>           warnings.warn(
                "dask_expr does not support the DataFrameIOFunction "
                "protocol for column projection. To enable column "
                "projection, please ensure that the signature of `func` "
                "includes a `columns=` keyword argument instead."
            )
E           UserWarning: dask_expr does not support the DataFrameIOFunction protocol for column projection. To enable column projection, please ensure that the signature of `func` includes a `columns=` keyword argument instead.

../../../miniconda3/envs/dask-distributed/lib/python3.9/site-packages/dask_expr/_collection.py:4999: UserWarning

UserWarning: dask_expr does not support the DataFrameIOFunction protocol for column projection. To enable column projection, please ensure that the signature of `func` includes a `columns=` keyword argument instead.
client = <Client: 'tcp://127.0.0.1:38997' processes=2 threads=2, memory=31.21 GiB>

    def test_multiple_linear(client):
>       df = dd.demo.make_timeseries(freq="15D", partition_freq="30D")

distributed/shuffle/tests/test_graph.py:114: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
../../../miniconda3/envs/dask-distributed/lib/python3.9/site-packages/dask/dataframe/io/demo.py:434: in make_timeseries
    return from_map(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

func = <dask.dataframe.io.demo.MakeDataframePart object at 0x7f5c9f9c2700>
args = None
meta =                name   id         x         y
timestamp                                   
2000-01-01  Charlie  988  0.213058 -0.761275
divisions = [Timestamp('2000-01-01 00:00:00'), Timestamp('2000-01-31 00:00:00'), Timestamp('2000-03-01 00:00:00'), Timestamp('2000-03-31 00:00:00'), Timestamp('2000-04-30 00:00:00'), Timestamp('2000-05-30 00:00:00'), ...]
label = 'make-timeseries', enforce_metadata = False
iterables = [[([Timestamp('2000-01-01 00:00:00'), Timestamp('2000-01-31 00:00:00')], 1124569897), ([Timestamp('2000-01-31 00:00:00...5-30 00:00:00')], 1971128239), ([Timestamp('2000-05-30 00:00:00'), Timestamp('2000-06-29 00:00:00')], 174883427), ...]]
kwargs = {}
DataFrameIOFunction = <class 'dask.dataframe.io.utils.DataFrameIOFunction'>
FromMap = <class 'dask_expr.io.io.FromMap'>
FromMapProjectable = <class 'dask_expr.io.io.FromMapProjectable'>
lengths = {12}, i = 0

    def from_map(
        func,
        *iterables,
        args=None,
        meta=no_default,
        divisions=None,
        label=None,
        enforce_metadata=False,
        **kwargs,
    ):
        """Create a dask-expr collection from a custom function map
    
        NOTE: The underlying ``Expr`` object produced by this API
        will support column projection (via ``simplify``) if
        the ``func`` argument has "columns" in its signature.
    
        """
        from dask.dataframe.io.utils import DataFrameIOFunction
    
        from dask_expr.io import FromMap, FromMapProjectable
    
        if "token" in kwargs:
            # This option doesn't really make sense in dask-expr
            raise NotImplementedError("dask_expr does not support a token argument.")
    
        lengths = set()
        iterables = list(iterables)
        for i, iterable in enumerate(iterables):
            if not isinstance(iterable, Iterable):
                raise ValueError(
                    f"All elements of `iterables` must be Iterable, got {type(iterable)}"
                )
            try:
                lengths.add(len(iterable))
            except (AttributeError, TypeError):
                iterables[i] = list(iterable)
                lengths.add(len(iterables[i]))
        if len(lengths) == 0:
            raise ValueError("`from_map` requires at least one Iterable input")
        elif len(lengths) > 1:
            raise ValueError("All `iterables` must have the same length")
        if lengths == {0}:
            raise ValueError("All `iterables` must have a non-zero length")
    
        # Check if `func` supports column projection
        allow_projection = False
        columns_arg_required = False
        if param := inspect.signature(func).parameters.get("columns", None):
            allow_projection = True
            columns_arg_required = param.default is param.empty
            if meta is no_default and columns_arg_required:
                raise TypeError(
                    "Argument `func` of `from_map` has a required `columns` "
                    " parameter and not `meta` provided."
                    "Either provide `meta` yourself or make `columns` an optional argument."
                )
        elif isinstance(func, DataFrameIOFunction):
>           warnings.warn(
                "dask_expr does not support the DataFrameIOFunction "
                "protocol for column projection. To enable column "
                "projection, please ensure that the signature of `func` "
                "includes a `columns=` keyword argument instead."
            )
E           UserWarning: dask_expr does not support the DataFrameIOFunction protocol for column projection. To enable column projection, please ensure that the signature of `func` includes a `columns=` keyword argument instead.

../../../miniconda3/envs/dask-distributed/lib/python3.9/site-packages/dask_expr/_collection.py:4999: UserWarning

AttributeError: 'dict' object has no attribute 'layers'
c = <Client: No scheduler connected>
s = <Scheduler 'tcp://127.0.0.1:38493', workers: 0, cores: 0, tasks: 0>
a = <Worker 'tcp://127.0.0.1:42937', name: 0, status: closed, stored: 0, running: 0/1, ready: 0, comm: 0, waiting: 0>
b = <Worker 'tcp://127.0.0.1:34017', name: 1, status: closed, stored: 0, running: 0/2, ready: 0, comm: 0, waiting: 0>
df_left =      k  v1
idx       
0    0   0
0    1   1
0    2   2
1    0   3
1    1   4
1    2   5
1    3   6
2    0   7
2    1  ...    0  37
9    1  38
9    2  39
9    3  40
9    4  41
9    5  42
9    6  43
10   0  44
10   1  45
10   2  46
10   3  47
df_right =      k  v1
idx       
0    0   0
0    1   1
0    2   2
0    3   3
1    0   4
1    1   5
2    0   6
2    1   7
2    2  ...    1  42
9    2  43
9    3  44
10   0  45
10   1  46
10   2  47
10   3  48
10   4  49
10   5  50
10   6  51
10   7  52
ddf_left_unknown = <dask_expr.expr.DataFrame: expr=ClearDivisions(frame=df)>
ddf_right_unknown = <dask_expr.expr.DataFrame: expr=ClearDivisions(frame=df)>
on = 'idx', how = 'inner'

    @gen_cluster(client=True)
    async def test_merge_unknown_to_unknown(
        c,
        s,
        a,
        b,
        df_left,
        df_right,
        ddf_left_unknown,
        ddf_right_unknown,
        on,
        how,
    ):
        # Compute expected
        expected = df_left.merge(df_right, on=on, how=how)
    
        # Merge unknown to unknown
        with dask.config.set({"dataframe.shuffle.method": "p2p"}):
            result_graph = ddf_left_unknown.merge(ddf_right_unknown, on=on, how=how)
        if not any(
            isinstance(layer, (HashJoinP2PLayer, P2PShuffleLayer))
>           for layer in result_graph.dask.layers.values()
        ):
E       AttributeError: 'dict' object has no attribute 'layers'

distributed/shuffle/tests/test_merge_column_and_index.py:183: AttributeError

AttributeError: 'dict' object has no attribute 'layers'
c = <Client: No scheduler connected>
s = <Scheduler 'tcp://127.0.0.1:34469', workers: 0, cores: 0, tasks: 0>
a = <Worker 'tcp://127.0.0.1:40547', name: 0, status: closed, stored: 0, running: 0/1, ready: 0, comm: 0, waiting: 0>
b = <Worker 'tcp://127.0.0.1:34333', name: 1, status: closed, stored: 0, running: 0/2, ready: 0, comm: 0, waiting: 0>
df_left =      k  v1
idx       
0    0   0
0    1   1
0    2   2
1    0   3
1    1   4
1    2   5
1    3   6
2    0   7
2    1  ...    0  37
9    1  38
9    2  39
9    3  40
9    4  41
9    5  42
9    6  43
10   0  44
10   1  45
10   2  46
10   3  47
df_right =      k  v1
idx       
0    0   0
0    1   1
0    2   2
0    3   3
1    0   4
1    1   5
2    0   6
2    1   7
2    2  ...    1  42
9    2  43
9    3  44
10   0  45
10   1  46
10   2  47
10   3  48
10   4  49
10   5  50
10   6  51
10   7  52
ddf_left_unknown = <dask_expr.expr.DataFrame: expr=ClearDivisions(frame=df)>
ddf_right_unknown = <dask_expr.expr.DataFrame: expr=ClearDivisions(frame=df)>
on = 'idx', how = 'left'

    @gen_cluster(client=True)
    async def test_merge_unknown_to_unknown(
        c,
        s,
        a,
        b,
        df_left,
        df_right,
        ddf_left_unknown,
        ddf_right_unknown,
        on,
        how,
    ):
        # Compute expected
        expected = df_left.merge(df_right, on=on, how=how)
    
        # Merge unknown to unknown
        with dask.config.set({"dataframe.shuffle.method": "p2p"}):
            result_graph = ddf_left_unknown.merge(ddf_right_unknown, on=on, how=how)
        if not any(
            isinstance(layer, (HashJoinP2PLayer, P2PShuffleLayer))
>           for layer in result_graph.dask.layers.values()
        ):
E       AttributeError: 'dict' object has no attribute 'layers'

distributed/shuffle/tests/test_merge_column_and_index.py:183: AttributeError

AttributeError: 'dict' object has no attribute 'layers'
c = <Client: No scheduler connected>
s = <Scheduler 'tcp://127.0.0.1:35841', workers: 0, cores: 0, tasks: 0>
a = <Worker 'tcp://127.0.0.1:45441', name: 0, status: closed, stored: 0, running: 0/1, ready: 0, comm: 0, waiting: 0>
b = <Worker 'tcp://127.0.0.1:40239', name: 1, status: closed, stored: 0, running: 0/2, ready: 0, comm: 0, waiting: 0>
df_left =      k  v1
idx       
0    0   0
0    1   1
0    2   2
1    0   3
1    1   4
1    2   5
1    3   6
2    0   7
2    1  ...    0  37
9    1  38
9    2  39
9    3  40
9    4  41
9    5  42
9    6  43
10   0  44
10   1  45
10   2  46
10   3  47
df_right =      k  v1
idx       
0    0   0
0    1   1
0    2   2
0    3   3
1    0   4
1    1   5
2    0   6
2    1   7
2    2  ...    1  42
9    2  43
9    3  44
10   0  45
10   1  46
10   2  47
10   3  48
10   4  49
10   5  50
10   6  51
10   7  52
ddf_left_unknown = <dask_expr.expr.DataFrame: expr=ClearDivisions(frame=df)>
ddf_right_unknown = <dask_expr.expr.DataFrame: expr=ClearDivisions(frame=df)>
on = 'idx', how = 'right'

    @gen_cluster(client=True)
    async def test_merge_unknown_to_unknown(
        c,
        s,
        a,
        b,
        df_left,
        df_right,
        ddf_left_unknown,
        ddf_right_unknown,
        on,
        how,
    ):
        # Compute expected
        expected = df_left.merge(df_right, on=on, how=how)
    
        # Merge unknown to unknown
        with dask.config.set({"dataframe.shuffle.method": "p2p"}):
            result_graph = ddf_left_unknown.merge(ddf_right_unknown, on=on, how=how)
        if not any(
            isinstance(layer, (HashJoinP2PLayer, P2PShuffleLayer))
>           for layer in result_graph.dask.layers.values()
        ):
E       AttributeError: 'dict' object has no attribute 'layers'

distributed/shuffle/tests/test_merge_column_and_index.py:183: AttributeError

AttributeError: 'dict' object has no attribute 'layers'
c = <Client: No scheduler connected>
s = <Scheduler 'tcp://127.0.0.1:38929', workers: 0, cores: 0, tasks: 0>
a = <Worker 'tcp://127.0.0.1:35857', name: 0, status: closed, stored: 0, running: 0/1, ready: 0, comm: 0, waiting: 0>
b = <Worker 'tcp://127.0.0.1:46239', name: 1, status: closed, stored: 0, running: 0/2, ready: 0, comm: 0, waiting: 0>
df_left =      k  v1
idx       
0    0   0
0    1   1
0    2   2
1    0   3
1    1   4
1    2   5
1    3   6
2    0   7
2    1  ...    0  37
9    1  38
9    2  39
9    3  40
9    4  41
9    5  42
9    6  43
10   0  44
10   1  45
10   2  46
10   3  47
df_right =      k  v1
idx       
0    0   0
0    1   1
0    2   2
0    3   3
1    0   4
1    1   5
2    0   6
2    1   7
2    2  ...    1  42
9    2  43
9    3  44
10   0  45
10   1  46
10   2  47
10   3  48
10   4  49
10   5  50
10   6  51
10   7  52
ddf_left_unknown = <dask_expr.expr.DataFrame: expr=ClearDivisions(frame=df)>
ddf_right_unknown = <dask_expr.expr.DataFrame: expr=ClearDivisions(frame=df)>
on = 'idx', how = 'outer'

    @gen_cluster(client=True)
    async def test_merge_unknown_to_unknown(
        c,
        s,
        a,
        b,
        df_left,
        df_right,
        ddf_left_unknown,
        ddf_right_unknown,
        on,
        how,
    ):
        # Compute expected
        expected = df_left.merge(df_right, on=on, how=how)
    
        # Merge unknown to unknown
        with dask.config.set({"dataframe.shuffle.method": "p2p"}):
            result_graph = ddf_left_unknown.merge(ddf_right_unknown, on=on, how=how)
        if not any(
            isinstance(layer, (HashJoinP2PLayer, P2PShuffleLayer))
>           for layer in result_graph.dask.layers.values()
        ):
E       AttributeError: 'dict' object has no attribute 'layers'

distributed/shuffle/tests/test_merge_column_and_index.py:183: AttributeError

AttributeError: 'dict' object has no attribute 'layers'
c = <Client: No scheduler connected>
s = <Scheduler 'tcp://127.0.0.1:33715', workers: 0, cores: 0, tasks: 0>
a = <Worker 'tcp://127.0.0.1:43907', name: 0, status: closed, stored: 0, running: 0/1, ready: 0, comm: 0, waiting: 0>
b = <Worker 'tcp://127.0.0.1:37979', name: 1, status: closed, stored: 0, running: 0/2, ready: 0, comm: 0, waiting: 0>
df_left =      k  v1
idx       
0    0   0
0    1   1
0    2   2
1    0   3
1    1   4
1    2   5
1    3   6
2    0   7
2    1  ...    0  37
9    1  38
9    2  39
9    3  40
9    4  41
9    5  42
9    6  43
10   0  44
10   1  45
10   2  46
10   3  47
df_right =      k  v1
idx       
0    0   0
0    1   1
0    2   2
0    3   3
1    0   4
1    1   5
2    0   6
2    1   7
2    2  ...    1  42
9    2  43
9    3  44
10   0  45
10   1  46
10   2  47
10   3  48
10   4  49
10   5  50
10   6  51
10   7  52
ddf_left_unknown = <dask_expr.expr.DataFrame: expr=ClearDivisions(frame=df)>
ddf_right_unknown = <dask_expr.expr.DataFrame: expr=ClearDivisions(frame=df)>
on = ['idx'], how = 'inner'

    @gen_cluster(client=True)
    async def test_merge_unknown_to_unknown(
        c,
        s,
        a,
        b,
        df_left,
        df_right,
        ddf_left_unknown,
        ddf_right_unknown,
        on,
        how,
    ):
        # Compute expected
        expected = df_left.merge(df_right, on=on, how=how)
    
        # Merge unknown to unknown
        with dask.config.set({"dataframe.shuffle.method": "p2p"}):
            result_graph = ddf_left_unknown.merge(ddf_right_unknown, on=on, how=how)
        if not any(
            isinstance(layer, (HashJoinP2PLayer, P2PShuffleLayer))
>           for layer in result_graph.dask.layers.values()
        ):
E       AttributeError: 'dict' object has no attribute 'layers'

distributed/shuffle/tests/test_merge_column_and_index.py:183: AttributeError

AttributeError: 'dict' object has no attribute 'layers'
c = <Client: No scheduler connected>
s = <Scheduler 'tcp://127.0.0.1:41225', workers: 0, cores: 0, tasks: 0>
a = <Worker 'tcp://127.0.0.1:40497', name: 0, status: closed, stored: 0, running: 0/1, ready: 0, comm: 0, waiting: 0>
b = <Worker 'tcp://127.0.0.1:41293', name: 1, status: closed, stored: 0, running: 0/2, ready: 0, comm: 0, waiting: 0>
df_left =      k  v1
idx       
0    0   0
0    1   1
0    2   2
1    0   3
1    1   4
1    2   5
1    3   6
2    0   7
2    1  ...    0  37
9    1  38
9    2  39
9    3  40
9    4  41
9    5  42
9    6  43
10   0  44
10   1  45
10   2  46
10   3  47
df_right =      k  v1
idx       
0    0   0
0    1   1
0    2   2
0    3   3
1    0   4
1    1   5
2    0   6
2    1   7
2    2  ...    1  42
9    2  43
9    3  44
10   0  45
10   1  46
10   2  47
10   3  48
10   4  49
10   5  50
10   6  51
10   7  52
ddf_left_unknown = <dask_expr.expr.DataFrame: expr=ClearDivisions(frame=df)>
ddf_right_unknown = <dask_expr.expr.DataFrame: expr=ClearDivisions(frame=df)>
on = ['idx'], how = 'left'

    @gen_cluster(client=True)
    async def test_merge_unknown_to_unknown(
        c,
        s,
        a,
        b,
        df_left,
        df_right,
        ddf_left_unknown,
        ddf_right_unknown,
        on,
        how,
    ):
        # Compute expected
        expected = df_left.merge(df_right, on=on, how=how)
    
        # Merge unknown to unknown
        with dask.config.set({"dataframe.shuffle.method": "p2p"}):
            result_graph = ddf_left_unknown.merge(ddf_right_unknown, on=on, how=how)
        if not any(
            isinstance(layer, (HashJoinP2PLayer, P2PShuffleLayer))
>           for layer in result_graph.dask.layers.values()
        ):
E       AttributeError: 'dict' object has no attribute 'layers'

distributed/shuffle/tests/test_merge_column_and_index.py:183: AttributeError

AttributeError: 'dict' object has no attribute 'layers'
c = <Client: No scheduler connected>
s = <Scheduler 'tcp://127.0.0.1:38441', workers: 0, cores: 0, tasks: 0>
a = <Worker 'tcp://127.0.0.1:33473', name: 0, status: closed, stored: 0, running: 0/1, ready: 0, comm: 0, waiting: 0>
b = <Worker 'tcp://127.0.0.1:41751', name: 1, status: closed, stored: 0, running: 0/2, ready: 0, comm: 0, waiting: 0>
df_left =      k  v1
idx       
0    0   0
0    1   1
0    2   2
1    0   3
1    1   4
1    2   5
1    3   6
2    0   7
2    1  ...    0  37
9    1  38
9    2  39
9    3  40
9    4  41
9    5  42
9    6  43
10   0  44
10   1  45
10   2  46
10   3  47
df_right =      k  v1
idx       
0    0   0
0    1   1
0    2   2
0    3   3
1    0   4
1    1   5
2    0   6
2    1   7
2    2  ...    1  42
9    2  43
9    3  44
10   0  45
10   1  46
10   2  47
10   3  48
10   4  49
10   5  50
10   6  51
10   7  52
ddf_left_unknown = <dask_expr.expr.DataFrame: expr=ClearDivisions(frame=df)>
ddf_right_unknown = <dask_expr.expr.DataFrame: expr=ClearDivisions(frame=df)>
on = ['idx'], how = 'right'

    @gen_cluster(client=True)
    async def test_merge_unknown_to_unknown(
        c,
        s,
        a,
        b,
        df_left,
        df_right,
        ddf_left_unknown,
        ddf_right_unknown,
        on,
        how,
    ):
        # Compute expected
        expected = df_left.merge(df_right, on=on, how=how)
    
        # Merge unknown to unknown
        with dask.config.set({"dataframe.shuffle.method": "p2p"}):
            result_graph = ddf_left_unknown.merge(ddf_right_unknown, on=on, how=how)
        if not any(
            isinstance(layer, (HashJoinP2PLayer, P2PShuffleLayer))
>           for layer in result_graph.dask.layers.values()
        ):
E       AttributeError: 'dict' object has no attribute 'layers'

distributed/shuffle/tests/test_merge_column_and_index.py:183: AttributeError

AttributeError: 'dict' object has no attribute 'layers'
c = <Client: No scheduler connected>
s = <Scheduler 'tcp://127.0.0.1:35629', workers: 0, cores: 0, tasks: 0>
a = <Worker 'tcp://127.0.0.1:41603', name: 0, status: closed, stored: 0, running: 0/1, ready: 0, comm: 0, waiting: 0>
b = <Worker 'tcp://127.0.0.1:45729', name: 1, status: closed, stored: 0, running: 0/2, ready: 0, comm: 0, waiting: 0>
df_left =      k  v1
idx       
0    0   0
0    1   1
0    2   2
1    0   3
1    1   4
1    2   5
1    3   6
2    0   7
2    1  ...    0  37
9    1  38
9    2  39
9    3  40
9    4  41
9    5  42
9    6  43
10   0  44
10   1  45
10   2  46
10   3  47
df_right =      k  v1
idx       
0    0   0
0    1   1
0    2   2
0    3   3
1    0   4
1    1   5
2    0   6
2    1   7
2    2  ...    1  42
9    2  43
9    3  44
10   0  45
10   1  46
10   2  47
10   3  48
10   4  49
10   5  50
10   6  51
10   7  52
ddf_left_unknown = <dask_expr.expr.DataFrame: expr=ClearDivisions(frame=df)>
ddf_right_unknown = <dask_expr.expr.DataFrame: expr=ClearDivisions(frame=df)>
on = ['idx'], how = 'outer'

    @gen_cluster(client=True)
    async def test_merge_unknown_to_unknown(
        c,
        s,
        a,
        b,
        df_left,
        df_right,
        ddf_left_unknown,
        ddf_right_unknown,
        on,
        how,
    ):
        # Compute expected
        expected = df_left.merge(df_right, on=on, how=how)
    
        # Merge unknown to unknown
        with dask.config.set({"dataframe.shuffle.method": "p2p"}):
            result_graph = ddf_left_unknown.merge(ddf_right_unknown, on=on, how=how)
        if not any(
            isinstance(layer, (HashJoinP2PLayer, P2PShuffleLayer))
>           for layer in result_graph.dask.layers.values()
        ):
E       AttributeError: 'dict' object has no attribute 'layers'

distributed/shuffle/tests/test_merge_column_and_index.py:183: AttributeError

AttributeError: 'dict' object has no attribute 'layers'
c = <Client: No scheduler connected>
s = <Scheduler 'tcp://127.0.0.1:44619', workers: 0, cores: 0, tasks: 0>
a = <Worker 'tcp://127.0.0.1:37617', name: 0, status: closed, stored: 0, running: 0/1, ready: 0, comm: 0, waiting: 0>
b = <Worker 'tcp://127.0.0.1:40373', name: 1, status: closed, stored: 0, running: 0/2, ready: 0, comm: 0, waiting: 0>
df_left =      k  v1
idx       
0    0   0
0    1   1
0    2   2
1    0   3
1    1   4
1    2   5
1    3   6
2    0   7
2    1  ...    0  37
9    1  38
9    2  39
9    3  40
9    4  41
9    5  42
9    6  43
10   0  44
10   1  45
10   2  46
10   3  47
df_right =      k  v1
idx       
0    0   0
0    1   1
0    2   2
0    3   3
1    0   4
1    1   5
2    0   6
2    1   7
2    2  ...    1  42
9    2  43
9    3  44
10   0  45
10   1  46
10   2  47
10   3  48
10   4  49
10   5  50
10   6  51
10   7  52
ddf_left_unknown = <dask_expr.expr.DataFrame: expr=ClearDivisions(frame=df)>
ddf_right_unknown = <dask_expr.expr.DataFrame: expr=ClearDivisions(frame=df)>
on = ['idx', 'k'], how = 'inner'

    @gen_cluster(client=True)
    async def test_merge_unknown_to_unknown(
        c,
        s,
        a,
        b,
        df_left,
        df_right,
        ddf_left_unknown,
        ddf_right_unknown,
        on,
        how,
    ):
        # Compute expected
        expected = df_left.merge(df_right, on=on, how=how)
    
        # Merge unknown to unknown
        with dask.config.set({"dataframe.shuffle.method": "p2p"}):
            result_graph = ddf_left_unknown.merge(ddf_right_unknown, on=on, how=how)
        if not any(
            isinstance(layer, (HashJoinP2PLayer, P2PShuffleLayer))
>           for layer in result_graph.dask.layers.values()
        ):
E       AttributeError: 'dict' object has no attribute 'layers'

distributed/shuffle/tests/test_merge_column_and_index.py:183: AttributeError

AttributeError: 'dict' object has no attribute 'layers'
c = <Client: No scheduler connected>
s = <Scheduler 'tcp://127.0.0.1:33221', workers: 0, cores: 0, tasks: 0>
a = <Worker 'tcp://127.0.0.1:44507', name: 0, status: closed, stored: 0, running: 0/1, ready: 0, comm: 0, waiting: 0>
b = <Worker 'tcp://127.0.0.1:43007', name: 1, status: closed, stored: 0, running: 0/2, ready: 0, comm: 0, waiting: 0>
df_left =      k  v1
idx       
0    0   0
0    1   1
0    2   2
1    0   3
1    1   4
1    2   5
1    3   6
2    0   7
2    1  ...    0  37
9    1  38
9    2  39
9    3  40
9    4  41
9    5  42
9    6  43
10   0  44
10   1  45
10   2  46
10   3  47
df_right =      k  v1
idx       
0    0   0
0    1   1
0    2   2
0    3   3
1    0   4
1    1   5
2    0   6
2    1   7
2    2  ...    1  42
9    2  43
9    3  44
10   0  45
10   1  46
10   2  47
10   3  48
10   4  49
10   5  50
10   6  51
10   7  52
ddf_left_unknown = <dask_expr.expr.DataFrame: expr=ClearDivisions(frame=df)>
ddf_right_unknown = <dask_expr.expr.DataFrame: expr=ClearDivisions(frame=df)>
on = ['idx', 'k'], how = 'left'

    @gen_cluster(client=True)
    async def test_merge_unknown_to_unknown(
        c,
        s,
        a,
        b,
        df_left,
        df_right,
        ddf_left_unknown,
        ddf_right_unknown,
        on,
        how,
    ):
        # Compute expected
        expected = df_left.merge(df_right, on=on, how=how)
    
        # Merge unknown to unknown
        with dask.config.set({"dataframe.shuffle.method": "p2p"}):
            result_graph = ddf_left_unknown.merge(ddf_right_unknown, on=on, how=how)
        if not any(
            isinstance(layer, (HashJoinP2PLayer, P2PShuffleLayer))
>           for layer in result_graph.dask.layers.values()
        ):
E       AttributeError: 'dict' object has no attribute 'layers'

distributed/shuffle/tests/test_merge_column_and_index.py:183: AttributeError

AttributeError: 'dict' object has no attribute 'layers'
c = <Client: No scheduler connected>
s = <Scheduler 'tcp://127.0.0.1:35909', workers: 0, cores: 0, tasks: 0>
a = <Worker 'tcp://127.0.0.1:34027', name: 0, status: closed, stored: 0, running: 0/1, ready: 0, comm: 0, waiting: 0>
b = <Worker 'tcp://127.0.0.1:46455', name: 1, status: closed, stored: 0, running: 0/2, ready: 0, comm: 0, waiting: 0>
df_left =      k  v1
idx       
0    0   0
0    1   1
0    2   2
1    0   3
1    1   4
1    2   5
1    3   6
2    0   7
2    1  ...    0  37
9    1  38
9    2  39
9    3  40
9    4  41
9    5  42
9    6  43
10   0  44
10   1  45
10   2  46
10   3  47
df_right =      k  v1
idx       
0    0   0
0    1   1
0    2   2
0    3   3
1    0   4
1    1   5
2    0   6
2    1   7
2    2  ...    1  42
9    2  43
9    3  44
10   0  45
10   1  46
10   2  47
10   3  48
10   4  49
10   5  50
10   6  51
10   7  52
ddf_left_unknown = <dask_expr.expr.DataFrame: expr=ClearDivisions(frame=df)>
ddf_right_unknown = <dask_expr.expr.DataFrame: expr=ClearDivisions(frame=df)>
on = ['idx', 'k'], how = 'right'

    @gen_cluster(client=True)
    async def test_merge_unknown_to_unknown(
        c,
        s,
        a,
        b,
        df_left,
        df_right,
        ddf_left_unknown,
        ddf_right_unknown,
        on,
        how,
    ):
        # Compute expected
        expected = df_left.merge(df_right, on=on, how=how)
    
        # Merge unknown to unknown
        with dask.config.set({"dataframe.shuffle.method": "p2p"}):
            result_graph = ddf_left_unknown.merge(ddf_right_unknown, on=on, how=how)
        if not any(
            isinstance(layer, (HashJoinP2PLayer, P2PShuffleLayer))
>           for layer in result_graph.dask.layers.values()
        ):
E       AttributeError: 'dict' object has no attribute 'layers'

distributed/shuffle/tests/test_merge_column_and_index.py:183: AttributeError

AttributeError: 'dict' object has no attribute 'layers'
c = <Client: No scheduler connected>
s = <Scheduler 'tcp://127.0.0.1:39581', workers: 0, cores: 0, tasks: 0>
a = <Worker 'tcp://127.0.0.1:44197', name: 0, status: closed, stored: 0, running: 0/1, ready: 0, comm: 0, waiting: 0>
b = <Worker 'tcp://127.0.0.1:33515', name: 1, status: closed, stored: 0, running: 0/2, ready: 0, comm: 0, waiting: 0>
df_left =      k  v1
idx       
0    0   0
0    1   1
0    2   2
1    0   3
1    1   4
1    2   5
1    3   6
2    0   7
2    1  ...    0  37
9    1  38
9    2  39
9    3  40
9    4  41
9    5  42
9    6  43
10   0  44
10   1  45
10   2  46
10   3  47
df_right =      k  v1
idx       
0    0   0
0    1   1
0    2   2
0    3   3
1    0   4
1    1   5
2    0   6
2    1   7
2    2  ...    1  42
9    2  43
9    3  44
10   0  45
10   1  46
10   2  47
10   3  48
10   4  49
10   5  50
10   6  51
10   7  52
ddf_left_unknown = <dask_expr.expr.DataFrame: expr=ClearDivisions(frame=df)>
ddf_right_unknown = <dask_expr.expr.DataFrame: expr=ClearDivisions(frame=df)>
on = ['idx', 'k'], how = 'outer'

    @gen_cluster(client=True)
    async def test_merge_unknown_to_unknown(
        c,
        s,
        a,
        b,
        df_left,
        df_right,
        ddf_left_unknown,
        ddf_right_unknown,
        on,
        how,
    ):
        # Compute expected
        expected = df_left.merge(df_right, on=on, how=how)
    
        # Merge unknown to unknown
        with dask.config.set({"dataframe.shuffle.method": "p2p"}):
            result_graph = ddf_left_unknown.merge(ddf_right_unknown, on=on, how=how)
        if not any(
            isinstance(layer, (HashJoinP2PLayer, P2PShuffleLayer))
>           for layer in result_graph.dask.layers.values()
        ):
E       AttributeError: 'dict' object has no attribute 'layers'

distributed/shuffle/tests/test_merge_column_and_index.py:183: AttributeError

AttributeError: 'dict' object has no attribute 'layers'
c = <Client: No scheduler connected>
s = <Scheduler 'tcp://127.0.0.1:46221', workers: 0, cores: 0, tasks: 0>
a = <Worker 'tcp://127.0.0.1:40317', name: 0, status: closed, stored: 0, running: 0/1, ready: 0, comm: 0, waiting: 0>
b = <Worker 'tcp://127.0.0.1:36801', name: 1, status: closed, stored: 0, running: 0/2, ready: 0, comm: 0, waiting: 0>
df_left =      k  v1
idx       
0    0   0
0    1   1
0    2   2
1    0   3
1    1   4
1    2   5
1    3   6
2    0   7
2    1  ...    0  37
9    1  38
9    2  39
9    3  40
9    4  41
9    5  42
9    6  43
10   0  44
10   1  45
10   2  46
10   3  47
df_right =      k  v1
idx       
0    0   0
0    1   1
0    2   2
0    3   3
1    0   4
1    1   5
2    0   6
2    1   7
2    2  ...    1  42
9    2  43
9    3  44
10   0  45
10   1  46
10   2  47
10   3  48
10   4  49
10   5  50
10   6  51
10   7  52
ddf_left_unknown = <dask_expr.expr.DataFrame: expr=ClearDivisions(frame=df)>
ddf_right_unknown = <dask_expr.expr.DataFrame: expr=ClearDivisions(frame=df)>
on = ['k', 'idx'], how = 'inner'

    @gen_cluster(client=True)
    async def test_merge_unknown_to_unknown(
        c,
        s,
        a,
        b,
        df_left,
        df_right,
        ddf_left_unknown,
        ddf_right_unknown,
        on,
        how,
    ):
        # Compute expected
        expected = df_left.merge(df_right, on=on, how=how)
    
        # Merge unknown to unknown
        with dask.config.set({"dataframe.shuffle.method": "p2p"}):
            result_graph = ddf_left_unknown.merge(ddf_right_unknown, on=on, how=how)
        if not any(
            isinstance(layer, (HashJoinP2PLayer, P2PShuffleLayer))
>           for layer in result_graph.dask.layers.values()
        ):
E       AttributeError: 'dict' object has no attribute 'layers'

distributed/shuffle/tests/test_merge_column_and_index.py:183: AttributeError

AttributeError: 'dict' object has no attribute 'layers'
c = <Client: No scheduler connected>
s = <Scheduler 'tcp://127.0.0.1:45737', workers: 0, cores: 0, tasks: 0>
a = <Worker 'tcp://127.0.0.1:46741', name: 0, status: closed, stored: 0, running: 0/1, ready: 0, comm: 0, waiting: 0>
b = <Worker 'tcp://127.0.0.1:42193', name: 1, status: closed, stored: 0, running: 0/2, ready: 0, comm: 0, waiting: 0>
df_left =      k  v1
idx       
0    0   0
0    1   1
0    2   2
1    0   3
1    1   4
1    2   5
1    3   6
2    0   7
2    1  ...    0  37
9    1  38
9    2  39
9    3  40
9    4  41
9    5  42
9    6  43
10   0  44
10   1  45
10   2  46
10   3  47
df_right =      k  v1
idx       
0    0   0
0    1   1
0    2   2
0    3   3
1    0   4
1    1   5
2    0   6
2    1   7
2    2  ...    1  42
9    2  43
9    3  44
10   0  45
10   1  46
10   2  47
10   3  48
10   4  49
10   5  50
10   6  51
10   7  52
ddf_left_unknown = <dask_expr.expr.DataFrame: expr=ClearDivisions(frame=df)>
ddf_right_unknown = <dask_expr.expr.DataFrame: expr=ClearDivisions(frame=df)>
on = ['k', 'idx'], how = 'left'

    @gen_cluster(client=True)
    async def test_merge_unknown_to_unknown(
        c,
        s,
        a,
        b,
        df_left,
        df_right,
        ddf_left_unknown,
        ddf_right_unknown,
        on,
        how,
    ):
        # Compute expected
        expected = df_left.merge(df_right, on=on, how=how)
    
        # Merge unknown to unknown
        with dask.config.set({"dataframe.shuffle.method": "p2p"}):
            result_graph = ddf_left_unknown.merge(ddf_right_unknown, on=on, how=how)
        if not any(
            isinstance(layer, (HashJoinP2PLayer, P2PShuffleLayer))
>           for layer in result_graph.dask.layers.values()
        ):
E       AttributeError: 'dict' object has no attribute 'layers'

distributed/shuffle/tests/test_merge_column_and_index.py:183: AttributeError

AttributeError: 'dict' object has no attribute 'layers'
c = <Client: No scheduler connected>
s = <Scheduler 'tcp://127.0.0.1:46091', workers: 0, cores: 0, tasks: 0>
a = <Worker 'tcp://127.0.0.1:39643', name: 0, status: closed, stored: 0, running: 0/1, ready: 0, comm: 0, waiting: 0>
b = <Worker 'tcp://127.0.0.1:38845', name: 1, status: closed, stored: 0, running: 0/2, ready: 0, comm: 0, waiting: 0>
df_left =      k  v1
idx       
0    0   0
0    1   1
0    2   2
1    0   3
1    1   4
1    2   5
1    3   6
2    0   7
2    1  ...    0  37
9    1  38
9    2  39
9    3  40
9    4  41
9    5  42
9    6  43
10   0  44
10   1  45
10   2  46
10   3  47
df_right =      k  v1
idx       
0    0   0
0    1   1
0    2   2
0    3   3
1    0   4
1    1   5
2    0   6
2    1   7
2    2  ...    1  42
9    2  43
9    3  44
10   0  45
10   1  46
10   2  47
10   3  48
10   4  49
10   5  50
10   6  51
10   7  52
ddf_left_unknown = <dask_expr.expr.DataFrame: expr=ClearDivisions(frame=df)>
ddf_right_unknown = <dask_expr.expr.DataFrame: expr=ClearDivisions(frame=df)>
on = ['k', 'idx'], how = 'right'

    @gen_cluster(client=True)
    async def test_merge_unknown_to_unknown(
        c,
        s,
        a,
        b,
        df_left,
        df_right,
        ddf_left_unknown,
        ddf_right_unknown,
        on,
        how,
    ):
        # Compute expected
        expected = df_left.merge(df_right, on=on, how=how)
    
        # Merge unknown to unknown
        with dask.config.set({"dataframe.shuffle.method": "p2p"}):
            result_graph = ddf_left_unknown.merge(ddf_right_unknown, on=on, how=how)
        if not any(
            isinstance(layer, (HashJoinP2PLayer, P2PShuffleLayer))
>           for layer in result_graph.dask.layers.values()
        ):
E       AttributeError: 'dict' object has no attribute 'layers'

distributed/shuffle/tests/test_merge_column_and_index.py:183: AttributeError