KeyError: 'disk'
c = <Client: No scheduler connected>
s = <Scheduler 'tcp://127.0.0.1:53376', workers: 0, cores: 0, tasks: 0>
a = Dask DataFrame Structure:
                   x      y
npartitions=2              
0              int64  int64
4                ...    ...
5                ...    ...
Dask Name: repartition-dataframe, 1 graph layer
b = Dask DataFrame Structure:
                   y      z
npartitions=2              
0              int64  int64
2                ...    ...
5                ...    ...
Dask Name: repartition-dataframe, 1 graph layer

    @gen_cluster(client=True)
    async def test_minimal_version(c, s, a, b):
        no_pyarrow_ctx = (
            mock.patch.dict("sys.modules", {"pyarrow": None})
            if pa is not None
            else contextlib.nullcontext()
        )
        with no_pyarrow_ctx:
            A = pd.DataFrame({"x": [1, 2, 3, 4, 5, 6], "y": [1, 1, 2, 2, 3, 4]})
            a = dd.repartition(A, [0, 4, 5])
    
            B = pd.DataFrame({"y": [1, 3, 4, 4, 5, 6], "z": [6, 5, 4, 3, 2, 1]})
            b = dd.repartition(B, [0, 2, 5])
    
            with pytest.raises(
                ModuleNotFoundError, match="requires pyarrow"
            ), dask.config.set({"dataframe.shuffle.method": "p2p"}):
>               await c.compute(dd.merge(a, b, left_on="x", right_on="z"))

distributed\shuffle\tests\test_merge.py:72: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\utils.py:259: in wrapper
    return func(*args, **kwargs)
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\dataframe\multi.py:745: in merge
    return hash_join(
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\utils.py:259: in wrapper
    return func(*args, **kwargs)
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\dataframe\multi.py:356: in hash_join
    return hash_join_p2p(
distributed\shuffle\_merge.py:110: in hash_join_p2p
    disk: bool = dask.config.get("distributed.p2p.disk")
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

key = 'distributed.p2p.disk', default = <no_default>
config = {'admin': {'traceback': {'shorten': ['concurrent[\\\\\\/]futures[\\\\\\/]', 'dask[\\\\\\/](base|core|local|multiproces...comm': {'compression': False, 'default-scheme': 'tcp', 'offload': '10MiB', 'require-encryption': None, ...}, ...}, ...}
override_with = None

    def get(
        key: str,
        default: Any = no_default,
        config: dict = config,
        override_with: Any = None,
    ) -> Any:
        """
        Get elements from global config
    
        If ``override_with`` is not None this value will be passed straight back.
        Useful for getting kwarg defaults from Dask config.
    
        Use '.' for nested access
    
        Examples
        --------
        >>> from dask import config
        >>> config.get('foo')  # doctest: +SKIP
        {'x': 1, 'y': 2}
    
        >>> config.get('foo.x')  # doctest: +SKIP
        1
    
        >>> config.get('foo.x.y', default=123)  # doctest: +SKIP
        123
    
        >>> config.get('foo.y', override_with=None)  # doctest: +SKIP
        2
    
        >>> config.get('foo.y', override_with=3)  # doctest: +SKIP
        3
    
        See Also
        --------
        dask.config.set
        """
        if override_with is not None:
            return override_with
        keys = key.split(".")
        result = config
        for k in keys:
            k = canonical_name(k, result)
            try:
>               result = result[k]
E               KeyError: 'disk'

C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\config.py:568: KeyError

KeyError: 'disk'
c = <Client: No scheduler connected>
s = <Scheduler 'tcp://127.0.0.1:53388', workers: 0, cores: 0, tasks: 0>
a = Dask DataFrame Structure:
                   x      y
npartitions=2              
0              int64  int64
4                ...    ...
5                ...    ...
Dask Name: repartition-dataframe, 1 graph layer
b = Dask DataFrame Structure:
                   y      z
npartitions=2              
0              int64  int64
2                ...    ...
5                ...    ...
Dask Name: repartition-dataframe, 1 graph layer
how = 'inner'

    @pytest.mark.parametrize("how", ["inner", "left", "right", "outer"])
    @gen_cluster(client=True)
    async def test_basic_merge(c, s, a, b, how):
        A = pd.DataFrame({"x": [1, 2, 3, 4, 5, 6], "y": [1, 1, 2, 2, 3, 4]})
        a = dd.repartition(A, [0, 4, 5])
    
        B = pd.DataFrame({"y": [1, 3, 4, 4, 5, 6], "z": [6, 5, 4, 3, 2, 1]})
        b = dd.repartition(B, [0, 2, 5])
    
>       joined = hash_join(a, "y", b, "y", how)

distributed\shuffle\tests\test_merge.py:84: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
distributed\shuffle\_merge.py:110: in hash_join_p2p
    disk: bool = dask.config.get("distributed.p2p.disk")
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

key = 'distributed.p2p.disk', default = <no_default>
config = {'admin': {'traceback': {'shorten': ['concurrent[\\\\\\/]futures[\\\\\\/]', 'dask[\\\\\\/](base|core|local|multiproces...comm': {'compression': False, 'default-scheme': 'tcp', 'offload': '10MiB', 'require-encryption': None, ...}, ...}, ...}
override_with = None

    def get(
        key: str,
        default: Any = no_default,
        config: dict = config,
        override_with: Any = None,
    ) -> Any:
        """
        Get elements from global config
    
        If ``override_with`` is not None this value will be passed straight back.
        Useful for getting kwarg defaults from Dask config.
    
        Use '.' for nested access
    
        Examples
        --------
        >>> from dask import config
        >>> config.get('foo')  # doctest: +SKIP
        {'x': 1, 'y': 2}
    
        >>> config.get('foo.x')  # doctest: +SKIP
        1
    
        >>> config.get('foo.x.y', default=123)  # doctest: +SKIP
        123
    
        >>> config.get('foo.y', override_with=None)  # doctest: +SKIP
        2
    
        >>> config.get('foo.y', override_with=3)  # doctest: +SKIP
        3
    
        See Also
        --------
        dask.config.set
        """
        if override_with is not None:
            return override_with
        keys = key.split(".")
        result = config
        for k in keys:
            k = canonical_name(k, result)
            try:
>               result = result[k]
E               KeyError: 'disk'

C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\config.py:568: KeyError

KeyError: 'disk'
c = <Client: No scheduler connected>
s = <Scheduler 'tcp://127.0.0.1:53400', workers: 0, cores: 0, tasks: 0>
a = Dask DataFrame Structure:
                   x      y
npartitions=2              
0              int64  int64
4                ...    ...
5                ...    ...
Dask Name: repartition-dataframe, 1 graph layer
b = Dask DataFrame Structure:
                   y      z
npartitions=2              
0              int64  int64
2                ...    ...
5                ...    ...
Dask Name: repartition-dataframe, 1 graph layer
how = 'left'

    @pytest.mark.parametrize("how", ["inner", "left", "right", "outer"])
    @gen_cluster(client=True)
    async def test_basic_merge(c, s, a, b, how):
        A = pd.DataFrame({"x": [1, 2, 3, 4, 5, 6], "y": [1, 1, 2, 2, 3, 4]})
        a = dd.repartition(A, [0, 4, 5])
    
        B = pd.DataFrame({"y": [1, 3, 4, 4, 5, 6], "z": [6, 5, 4, 3, 2, 1]})
        b = dd.repartition(B, [0, 2, 5])
    
>       joined = hash_join(a, "y", b, "y", how)

distributed\shuffle\tests\test_merge.py:84: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
distributed\shuffle\_merge.py:110: in hash_join_p2p
    disk: bool = dask.config.get("distributed.p2p.disk")
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

key = 'distributed.p2p.disk', default = <no_default>
config = {'admin': {'traceback': {'shorten': ['concurrent[\\\\\\/]futures[\\\\\\/]', 'dask[\\\\\\/](base|core|local|multiproces...comm': {'compression': False, 'default-scheme': 'tcp', 'offload': '10MiB', 'require-encryption': None, ...}, ...}, ...}
override_with = None

    def get(
        key: str,
        default: Any = no_default,
        config: dict = config,
        override_with: Any = None,
    ) -> Any:
        """
        Get elements from global config
    
        If ``override_with`` is not None this value will be passed straight back.
        Useful for getting kwarg defaults from Dask config.
    
        Use '.' for nested access
    
        Examples
        --------
        >>> from dask import config
        >>> config.get('foo')  # doctest: +SKIP
        {'x': 1, 'y': 2}
    
        >>> config.get('foo.x')  # doctest: +SKIP
        1
    
        >>> config.get('foo.x.y', default=123)  # doctest: +SKIP
        123
    
        >>> config.get('foo.y', override_with=None)  # doctest: +SKIP
        2
    
        >>> config.get('foo.y', override_with=3)  # doctest: +SKIP
        3
    
        See Also
        --------
        dask.config.set
        """
        if override_with is not None:
            return override_with
        keys = key.split(".")
        result = config
        for k in keys:
            k = canonical_name(k, result)
            try:
>               result = result[k]
E               KeyError: 'disk'

C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\config.py:568: KeyError

KeyError: 'disk'
c = <Client: No scheduler connected>
s = <Scheduler 'tcp://127.0.0.1:53412', workers: 0, cores: 0, tasks: 0>
a = Dask DataFrame Structure:
                   x      y
npartitions=2              
0              int64  int64
4                ...    ...
5                ...    ...
Dask Name: repartition-dataframe, 1 graph layer
b = Dask DataFrame Structure:
                   y      z
npartitions=2              
0              int64  int64
2                ...    ...
5                ...    ...
Dask Name: repartition-dataframe, 1 graph layer
how = 'right'

    @pytest.mark.parametrize("how", ["inner", "left", "right", "outer"])
    @gen_cluster(client=True)
    async def test_basic_merge(c, s, a, b, how):
        A = pd.DataFrame({"x": [1, 2, 3, 4, 5, 6], "y": [1, 1, 2, 2, 3, 4]})
        a = dd.repartition(A, [0, 4, 5])
    
        B = pd.DataFrame({"y": [1, 3, 4, 4, 5, 6], "z": [6, 5, 4, 3, 2, 1]})
        b = dd.repartition(B, [0, 2, 5])
    
>       joined = hash_join(a, "y", b, "y", how)

distributed\shuffle\tests\test_merge.py:84: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
distributed\shuffle\_merge.py:110: in hash_join_p2p
    disk: bool = dask.config.get("distributed.p2p.disk")
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

key = 'distributed.p2p.disk', default = <no_default>
config = {'admin': {'traceback': {'shorten': ['concurrent[\\\\\\/]futures[\\\\\\/]', 'dask[\\\\\\/](base|core|local|multiproces...comm': {'compression': False, 'default-scheme': 'tcp', 'offload': '10MiB', 'require-encryption': None, ...}, ...}, ...}
override_with = None

    def get(
        key: str,
        default: Any = no_default,
        config: dict = config,
        override_with: Any = None,
    ) -> Any:
        """
        Get elements from global config
    
        If ``override_with`` is not None this value will be passed straight back.
        Useful for getting kwarg defaults from Dask config.
    
        Use '.' for nested access
    
        Examples
        --------
        >>> from dask import config
        >>> config.get('foo')  # doctest: +SKIP
        {'x': 1, 'y': 2}
    
        >>> config.get('foo.x')  # doctest: +SKIP
        1
    
        >>> config.get('foo.x.y', default=123)  # doctest: +SKIP
        123
    
        >>> config.get('foo.y', override_with=None)  # doctest: +SKIP
        2
    
        >>> config.get('foo.y', override_with=3)  # doctest: +SKIP
        3
    
        See Also
        --------
        dask.config.set
        """
        if override_with is not None:
            return override_with
        keys = key.split(".")
        result = config
        for k in keys:
            k = canonical_name(k, result)
            try:
>               result = result[k]
E               KeyError: 'disk'

C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\config.py:568: KeyError

KeyError: 'disk'
c = <Client: No scheduler connected>
s = <Scheduler 'tcp://127.0.0.1:53424', workers: 0, cores: 0, tasks: 0>
a = Dask DataFrame Structure:
                   x      y
npartitions=2              
0              int64  int64
4                ...    ...
5                ...    ...
Dask Name: repartition-dataframe, 1 graph layer
b = Dask DataFrame Structure:
                   y      z
npartitions=2              
0              int64  int64
2                ...    ...
5                ...    ...
Dask Name: repartition-dataframe, 1 graph layer
how = 'outer'

    @pytest.mark.parametrize("how", ["inner", "left", "right", "outer"])
    @gen_cluster(client=True)
    async def test_basic_merge(c, s, a, b, how):
        A = pd.DataFrame({"x": [1, 2, 3, 4, 5, 6], "y": [1, 1, 2, 2, 3, 4]})
        a = dd.repartition(A, [0, 4, 5])
    
        B = pd.DataFrame({"y": [1, 3, 4, 4, 5, 6], "z": [6, 5, 4, 3, 2, 1]})
        b = dd.repartition(B, [0, 2, 5])
    
>       joined = hash_join(a, "y", b, "y", how)

distributed\shuffle\tests\test_merge.py:84: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
distributed\shuffle\_merge.py:110: in hash_join_p2p
    disk: bool = dask.config.get("distributed.p2p.disk")
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

key = 'distributed.p2p.disk', default = <no_default>
config = {'admin': {'traceback': {'shorten': ['concurrent[\\\\\\/]futures[\\\\\\/]', 'dask[\\\\\\/](base|core|local|multiproces...comm': {'compression': False, 'default-scheme': 'tcp', 'offload': '10MiB', 'require-encryption': None, ...}, ...}, ...}
override_with = None

    def get(
        key: str,
        default: Any = no_default,
        config: dict = config,
        override_with: Any = None,
    ) -> Any:
        """
        Get elements from global config
    
        If ``override_with`` is not None this value will be passed straight back.
        Useful for getting kwarg defaults from Dask config.
    
        Use '.' for nested access
    
        Examples
        --------
        >>> from dask import config
        >>> config.get('foo')  # doctest: +SKIP
        {'x': 1, 'y': 2}
    
        >>> config.get('foo.x')  # doctest: +SKIP
        1
    
        >>> config.get('foo.x.y', default=123)  # doctest: +SKIP
        123
    
        >>> config.get('foo.y', override_with=None)  # doctest: +SKIP
        2
    
        >>> config.get('foo.y', override_with=3)  # doctest: +SKIP
        3
    
        See Also
        --------
        dask.config.set
        """
        if override_with is not None:
            return override_with
        keys = key.split(".")
        result = config
        for k in keys:
            k = canonical_name(k, result)
            try:
>               result = result[k]
E               KeyError: 'disk'

C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\config.py:568: KeyError

KeyError: 'disk'
c = <Client: No scheduler connected>
s = <Scheduler 'tcp://127.0.0.1:53436', workers: 0, cores: 0, tasks: 0>
a = <Worker 'tcp://127.0.0.1:53437', name: 0, status: closed, stored: 0, running: 0/1, ready: 0, comm: 0, waiting: 0>
b = <Worker 'tcp://127.0.0.1:53440', name: 1, status: closed, stored: 0, running: 0/2, ready: 0, comm: 0, waiting: 0>

    @gen_cluster(client=True)
    async def test_merge_p2p_shuffle_reused_dataframe_with_different_parameters(c, s, a, b):
        pdf1 = pd.DataFrame({"a": range(100), "b": range(0, 200, 2)})
        pdf2 = pd.DataFrame({"x": range(200), "y": [1, 2, 3, 4] * 50})
        ddf1 = dd.from_pandas(pdf1, npartitions=5)
        ddf2 = dd.from_pandas(pdf2, npartitions=10)
    
        with dask.config.set({"dataframe.shuffle.method": "p2p"}):
            out = (
>               ddf1.merge(ddf2, left_on="a", right_on="x")
                # Vary the number of output partitions for the shuffles of dd2
                .repartition(npartitions=20).merge(ddf2, left_on="b", right_on="x")
            )

distributed\shuffle\tests\test_merge.py:120: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\utils.py:259: in wrapper
    return func(*args, **kwargs)
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\dataframe\core.py:6010: in merge
    return merge(
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\utils.py:259: in wrapper
    return func(*args, **kwargs)
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\dataframe\multi.py:745: in merge
    return hash_join(
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\utils.py:259: in wrapper
    return func(*args, **kwargs)
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\dataframe\multi.py:356: in hash_join
    return hash_join_p2p(
distributed\shuffle\_merge.py:110: in hash_join_p2p
    disk: bool = dask.config.get("distributed.p2p.disk")
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

key = 'distributed.p2p.disk', default = <no_default>
config = {'admin': {'traceback': {'shorten': ['concurrent[\\\\\\/]futures[\\\\\\/]', 'dask[\\\\\\/](base|core|local|multiproces...comm': {'compression': False, 'default-scheme': 'tcp', 'offload': '10MiB', 'require-encryption': None, ...}, ...}, ...}
override_with = None

    def get(
        key: str,
        default: Any = no_default,
        config: dict = config,
        override_with: Any = None,
    ) -> Any:
        """
        Get elements from global config
    
        If ``override_with`` is not None this value will be passed straight back.
        Useful for getting kwarg defaults from Dask config.
    
        Use '.' for nested access
    
        Examples
        --------
        >>> from dask import config
        >>> config.get('foo')  # doctest: +SKIP
        {'x': 1, 'y': 2}
    
        >>> config.get('foo.x')  # doctest: +SKIP
        1
    
        >>> config.get('foo.x.y', default=123)  # doctest: +SKIP
        123
    
        >>> config.get('foo.y', override_with=None)  # doctest: +SKIP
        2
    
        >>> config.get('foo.y', override_with=3)  # doctest: +SKIP
        3
    
        See Also
        --------
        dask.config.set
        """
        if override_with is not None:
            return override_with
        keys = key.split(".")
        result = config
        for k in keys:
            k = canonical_name(k, result)
            try:
>               result = result[k]
E               KeyError: 'disk'

C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\config.py:568: KeyError

KeyError: 'disk'
c = <Client: No scheduler connected>
s = <Scheduler 'tcp://127.0.0.1:53448', workers: 0, cores: 0, tasks: 0>
a = <Worker 'tcp://127.0.0.1:53449', name: 0, status: closed, stored: 0, running: 0/1, ready: 0, comm: 0, waiting: 0>
b = <Worker 'tcp://127.0.0.1:53452', name: 1, status: closed, stored: 0, running: 0/2, ready: 0, comm: 0, waiting: 0>

    @gen_cluster(client=True)
    async def test_merge_p2p_shuffle_reused_dataframe_with_same_parameters(c, s, a, b):
        pdf1 = pd.DataFrame({"a": range(100), "b": range(0, 200, 2)})
        pdf2 = pd.DataFrame({"x": range(200), "y": [1, 2, 3, 4] * 50})
        ddf1 = dd.from_pandas(pdf1, npartitions=5)
        ddf2 = dd.from_pandas(pdf2, npartitions=10)
    
        # This performs two shuffles:
        #   * ddf1 is shuffled on `a`
        #   * ddf2 is shuffled on `x`
        with dask.config.set({"dataframe.shuffle.method": "p2p"}):
>           ddf3 = ddf1.merge(
                ddf2,
                left_on="a",
                right_on="x",
            )

distributed\shuffle\tests\test_merge.py:146: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\utils.py:259: in wrapper
    return func(*args, **kwargs)
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\dataframe\core.py:6010: in merge
    return merge(
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\utils.py:259: in wrapper
    return func(*args, **kwargs)
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\dataframe\multi.py:745: in merge
    return hash_join(
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\utils.py:259: in wrapper
    return func(*args, **kwargs)
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\dataframe\multi.py:356: in hash_join
    return hash_join_p2p(
distributed\shuffle\_merge.py:110: in hash_join_p2p
    disk: bool = dask.config.get("distributed.p2p.disk")
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

key = 'distributed.p2p.disk', default = <no_default>
config = {'admin': {'traceback': {'shorten': ['concurrent[\\\\\\/]futures[\\\\\\/]', 'dask[\\\\\\/](base|core|local|multiproces...comm': {'compression': False, 'default-scheme': 'tcp', 'offload': '10MiB', 'require-encryption': None, ...}, ...}, ...}
override_with = None

    def get(
        key: str,
        default: Any = no_default,
        config: dict = config,
        override_with: Any = None,
    ) -> Any:
        """
        Get elements from global config
    
        If ``override_with`` is not None this value will be passed straight back.
        Useful for getting kwarg defaults from Dask config.
    
        Use '.' for nested access
    
        Examples
        --------
        >>> from dask import config
        >>> config.get('foo')  # doctest: +SKIP
        {'x': 1, 'y': 2}
    
        >>> config.get('foo.x')  # doctest: +SKIP
        1
    
        >>> config.get('foo.x.y', default=123)  # doctest: +SKIP
        123
    
        >>> config.get('foo.y', override_with=None)  # doctest: +SKIP
        2
    
        >>> config.get('foo.y', override_with=3)  # doctest: +SKIP
        3
    
        See Also
        --------
        dask.config.set
        """
        if override_with is not None:
            return override_with
        keys = key.split(".")
        result = config
        for k in keys:
            k = canonical_name(k, result)
            try:
>               result = result[k]
E               KeyError: 'disk'

C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\config.py:568: KeyError

KeyError: 'disk'
c = <Client: No scheduler connected>
s = <Scheduler 'tcp://127.0.0.1:53460', workers: 0, cores: 0, tasks: 0>
a = Dask DataFrame Structure:
                   x      y
npartitions=2              
0              int64  int64
4                ...    ...
5                ...    ...
Dask Name: repartition-dataframe, 1 graph layer
b = Dask DataFrame Structure:
                   y      z
npartitions=2              
0              int64  int64
2                ...    ...
5                ...    ...
Dask Name: repartition-dataframe, 1 graph layer
how = 'inner', disk = True

    @pytest.mark.parametrize("how", ["inner", "outer", "left", "right"])
    @pytest.mark.parametrize("disk", [True, False])
    @gen_cluster(client=True)
    async def test_merge(c, s, a, b, how, disk):
        A = pd.DataFrame({"x": [1, 2, 3, 4, 5, 6], "y": [1, 1, 2, 2, 3, 4]})
        a = dd.repartition(A, [0, 4, 5])
    
        B = pd.DataFrame({"y": [1, 3, 4, 4, 5, 6], "z": [6, 5, 4, 3, 2, 1]})
        b = dd.repartition(B, [0, 2, 5])
    
        with dask.config.set({"dataframe.shuffle.method": "p2p"}):
            with dask.config.set({"distributed.p2p.storage.disk": disk}):
                joined = dd.merge(a, b, left_index=True, right_index=True, how=how)
            res = await c.compute(joined)
            assert_eq(
                res,
                pd.merge(A, B, left_index=True, right_index=True, how=how),
            )
>           joined = dd.merge(a, b, on="y", how=how)

distributed\shuffle\tests\test_merge.py:190: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\utils.py:259: in wrapper
    return func(*args, **kwargs)
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\dataframe\multi.py:745: in merge
    return hash_join(
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\utils.py:259: in wrapper
    return func(*args, **kwargs)
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\dataframe\multi.py:356: in hash_join
    return hash_join_p2p(
distributed\shuffle\_merge.py:110: in hash_join_p2p
    disk: bool = dask.config.get("distributed.p2p.disk")
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

key = 'distributed.p2p.disk', default = <no_default>
config = {'admin': {'traceback': {'shorten': ['concurrent[\\\\\\/]futures[\\\\\\/]', 'dask[\\\\\\/](base|core|local|multiproces...comm': {'compression': False, 'default-scheme': 'tcp', 'offload': '10MiB', 'require-encryption': None, ...}, ...}, ...}
override_with = None

    def get(
        key: str,
        default: Any = no_default,
        config: dict = config,
        override_with: Any = None,
    ) -> Any:
        """
        Get elements from global config
    
        If ``override_with`` is not None this value will be passed straight back.
        Useful for getting kwarg defaults from Dask config.
    
        Use '.' for nested access
    
        Examples
        --------
        >>> from dask import config
        >>> config.get('foo')  # doctest: +SKIP
        {'x': 1, 'y': 2}
    
        >>> config.get('foo.x')  # doctest: +SKIP
        1
    
        >>> config.get('foo.x.y', default=123)  # doctest: +SKIP
        123
    
        >>> config.get('foo.y', override_with=None)  # doctest: +SKIP
        2
    
        >>> config.get('foo.y', override_with=3)  # doctest: +SKIP
        3
    
        See Also
        --------
        dask.config.set
        """
        if override_with is not None:
            return override_with
        keys = key.split(".")
        result = config
        for k in keys:
            k = canonical_name(k, result)
            try:
>               result = result[k]
E               KeyError: 'disk'

C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\config.py:568: KeyError

KeyError: 'disk'
c = <Client: No scheduler connected>
s = <Scheduler 'tcp://127.0.0.1:53477', workers: 0, cores: 0, tasks: 0>
a = Dask DataFrame Structure:
                   x      y
npartitions=2              
0              int64  int64
4                ...    ...
5                ...    ...
Dask Name: repartition-dataframe, 1 graph layer
b = Dask DataFrame Structure:
                   y      z
npartitions=2              
0              int64  int64
2                ...    ...
5                ...    ...
Dask Name: repartition-dataframe, 1 graph layer
how = 'outer', disk = True

    @pytest.mark.parametrize("how", ["inner", "outer", "left", "right"])
    @pytest.mark.parametrize("disk", [True, False])
    @gen_cluster(client=True)
    async def test_merge(c, s, a, b, how, disk):
        A = pd.DataFrame({"x": [1, 2, 3, 4, 5, 6], "y": [1, 1, 2, 2, 3, 4]})
        a = dd.repartition(A, [0, 4, 5])
    
        B = pd.DataFrame({"y": [1, 3, 4, 4, 5, 6], "z": [6, 5, 4, 3, 2, 1]})
        b = dd.repartition(B, [0, 2, 5])
    
        with dask.config.set({"dataframe.shuffle.method": "p2p"}):
            with dask.config.set({"distributed.p2p.storage.disk": disk}):
                joined = dd.merge(a, b, left_index=True, right_index=True, how=how)
            res = await c.compute(joined)
            assert_eq(
                res,
                pd.merge(A, B, left_index=True, right_index=True, how=how),
            )
>           joined = dd.merge(a, b, on="y", how=how)

distributed\shuffle\tests\test_merge.py:190: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\utils.py:259: in wrapper
    return func(*args, **kwargs)
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\dataframe\multi.py:745: in merge
    return hash_join(
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\utils.py:259: in wrapper
    return func(*args, **kwargs)
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\dataframe\multi.py:356: in hash_join
    return hash_join_p2p(
distributed\shuffle\_merge.py:110: in hash_join_p2p
    disk: bool = dask.config.get("distributed.p2p.disk")
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

key = 'distributed.p2p.disk', default = <no_default>
config = {'admin': {'traceback': {'shorten': ['concurrent[\\\\\\/]futures[\\\\\\/]', 'dask[\\\\\\/](base|core|local|multiproces...comm': {'compression': False, 'default-scheme': 'tcp', 'offload': '10MiB', 'require-encryption': None, ...}, ...}, ...}
override_with = None

    def get(
        key: str,
        default: Any = no_default,
        config: dict = config,
        override_with: Any = None,
    ) -> Any:
        """
        Get elements from global config
    
        If ``override_with`` is not None this value will be passed straight back.
        Useful for getting kwarg defaults from Dask config.
    
        Use '.' for nested access
    
        Examples
        --------
        >>> from dask import config
        >>> config.get('foo')  # doctest: +SKIP
        {'x': 1, 'y': 2}
    
        >>> config.get('foo.x')  # doctest: +SKIP
        1
    
        >>> config.get('foo.x.y', default=123)  # doctest: +SKIP
        123
    
        >>> config.get('foo.y', override_with=None)  # doctest: +SKIP
        2
    
        >>> config.get('foo.y', override_with=3)  # doctest: +SKIP
        3
    
        See Also
        --------
        dask.config.set
        """
        if override_with is not None:
            return override_with
        keys = key.split(".")
        result = config
        for k in keys:
            k = canonical_name(k, result)
            try:
>               result = result[k]
E               KeyError: 'disk'

C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\config.py:568: KeyError

KeyError: 'disk'
c = <Client: No scheduler connected>
s = <Scheduler 'tcp://127.0.0.1:53494', workers: 0, cores: 0, tasks: 0>
a = Dask DataFrame Structure:
                   x      y
npartitions=2              
0              int64  int64
4                ...    ...
5                ...    ...
Dask Name: repartition-dataframe, 1 graph layer
b = Dask DataFrame Structure:
                   y      z
npartitions=2              
0              int64  int64
2                ...    ...
5                ...    ...
Dask Name: repartition-dataframe, 1 graph layer
how = 'left', disk = True

    @pytest.mark.parametrize("how", ["inner", "outer", "left", "right"])
    @pytest.mark.parametrize("disk", [True, False])
    @gen_cluster(client=True)
    async def test_merge(c, s, a, b, how, disk):
        A = pd.DataFrame({"x": [1, 2, 3, 4, 5, 6], "y": [1, 1, 2, 2, 3, 4]})
        a = dd.repartition(A, [0, 4, 5])
    
        B = pd.DataFrame({"y": [1, 3, 4, 4, 5, 6], "z": [6, 5, 4, 3, 2, 1]})
        b = dd.repartition(B, [0, 2, 5])
    
        with dask.config.set({"dataframe.shuffle.method": "p2p"}):
            with dask.config.set({"distributed.p2p.storage.disk": disk}):
                joined = dd.merge(a, b, left_index=True, right_index=True, how=how)
            res = await c.compute(joined)
            assert_eq(
                res,
                pd.merge(A, B, left_index=True, right_index=True, how=how),
            )
>           joined = dd.merge(a, b, on="y", how=how)

distributed\shuffle\tests\test_merge.py:190: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\utils.py:259: in wrapper
    return func(*args, **kwargs)
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\dataframe\multi.py:745: in merge
    return hash_join(
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\utils.py:259: in wrapper
    return func(*args, **kwargs)
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\dataframe\multi.py:356: in hash_join
    return hash_join_p2p(
distributed\shuffle\_merge.py:110: in hash_join_p2p
    disk: bool = dask.config.get("distributed.p2p.disk")
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

key = 'distributed.p2p.disk', default = <no_default>
config = {'admin': {'traceback': {'shorten': ['concurrent[\\\\\\/]futures[\\\\\\/]', 'dask[\\\\\\/](base|core|local|multiproces...comm': {'compression': False, 'default-scheme': 'tcp', 'offload': '10MiB', 'require-encryption': None, ...}, ...}, ...}
override_with = None

    def get(
        key: str,
        default: Any = no_default,
        config: dict = config,
        override_with: Any = None,
    ) -> Any:
        """
        Get elements from global config
    
        If ``override_with`` is not None this value will be passed straight back.
        Useful for getting kwarg defaults from Dask config.
    
        Use '.' for nested access
    
        Examples
        --------
        >>> from dask import config
        >>> config.get('foo')  # doctest: +SKIP
        {'x': 1, 'y': 2}
    
        >>> config.get('foo.x')  # doctest: +SKIP
        1
    
        >>> config.get('foo.x.y', default=123)  # doctest: +SKIP
        123
    
        >>> config.get('foo.y', override_with=None)  # doctest: +SKIP
        2
    
        >>> config.get('foo.y', override_with=3)  # doctest: +SKIP
        3
    
        See Also
        --------
        dask.config.set
        """
        if override_with is not None:
            return override_with
        keys = key.split(".")
        result = config
        for k in keys:
            k = canonical_name(k, result)
            try:
>               result = result[k]
E               KeyError: 'disk'

C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\config.py:568: KeyError

KeyError: 'disk'
c = <Client: No scheduler connected>
s = <Scheduler 'tcp://127.0.0.1:53511', workers: 0, cores: 0, tasks: 0>
a = Dask DataFrame Structure:
                   x      y
npartitions=2              
0              int64  int64
4                ...    ...
5                ...    ...
Dask Name: repartition-dataframe, 1 graph layer
b = Dask DataFrame Structure:
                   y      z
npartitions=2              
0              int64  int64
2                ...    ...
5                ...    ...
Dask Name: repartition-dataframe, 1 graph layer
how = 'right', disk = True

    @pytest.mark.parametrize("how", ["inner", "outer", "left", "right"])
    @pytest.mark.parametrize("disk", [True, False])
    @gen_cluster(client=True)
    async def test_merge(c, s, a, b, how, disk):
        A = pd.DataFrame({"x": [1, 2, 3, 4, 5, 6], "y": [1, 1, 2, 2, 3, 4]})
        a = dd.repartition(A, [0, 4, 5])
    
        B = pd.DataFrame({"y": [1, 3, 4, 4, 5, 6], "z": [6, 5, 4, 3, 2, 1]})
        b = dd.repartition(B, [0, 2, 5])
    
        with dask.config.set({"dataframe.shuffle.method": "p2p"}):
            with dask.config.set({"distributed.p2p.storage.disk": disk}):
                joined = dd.merge(a, b, left_index=True, right_index=True, how=how)
            res = await c.compute(joined)
            assert_eq(
                res,
                pd.merge(A, B, left_index=True, right_index=True, how=how),
            )
>           joined = dd.merge(a, b, on="y", how=how)

distributed\shuffle\tests\test_merge.py:190: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\utils.py:259: in wrapper
    return func(*args, **kwargs)
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\dataframe\multi.py:745: in merge
    return hash_join(
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\utils.py:259: in wrapper
    return func(*args, **kwargs)
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\dataframe\multi.py:356: in hash_join
    return hash_join_p2p(
distributed\shuffle\_merge.py:110: in hash_join_p2p
    disk: bool = dask.config.get("distributed.p2p.disk")
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

key = 'distributed.p2p.disk', default = <no_default>
config = {'admin': {'traceback': {'shorten': ['concurrent[\\\\\\/]futures[\\\\\\/]', 'dask[\\\\\\/](base|core|local|multiproces...comm': {'compression': False, 'default-scheme': 'tcp', 'offload': '10MiB', 'require-encryption': None, ...}, ...}, ...}
override_with = None

    def get(
        key: str,
        default: Any = no_default,
        config: dict = config,
        override_with: Any = None,
    ) -> Any:
        """
        Get elements from global config
    
        If ``override_with`` is not None this value will be passed straight back.
        Useful for getting kwarg defaults from Dask config.
    
        Use '.' for nested access
    
        Examples
        --------
        >>> from dask import config
        >>> config.get('foo')  # doctest: +SKIP
        {'x': 1, 'y': 2}
    
        >>> config.get('foo.x')  # doctest: +SKIP
        1
    
        >>> config.get('foo.x.y', default=123)  # doctest: +SKIP
        123
    
        >>> config.get('foo.y', override_with=None)  # doctest: +SKIP
        2
    
        >>> config.get('foo.y', override_with=3)  # doctest: +SKIP
        3
    
        See Also
        --------
        dask.config.set
        """
        if override_with is not None:
            return override_with
        keys = key.split(".")
        result = config
        for k in keys:
            k = canonical_name(k, result)
            try:
>               result = result[k]
E               KeyError: 'disk'

C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\config.py:568: KeyError

KeyError: 'disk'
c = <Client: No scheduler connected>
s = <Scheduler 'tcp://127.0.0.1:53528', workers: 0, cores: 0, tasks: 0>
a = Dask DataFrame Structure:
                   x      y
npartitions=2              
0              int64  int64
4                ...    ...
5                ...    ...
Dask Name: repartition-dataframe, 1 graph layer
b = Dask DataFrame Structure:
                   y      z
npartitions=2              
0              int64  int64
2                ...    ...
5                ...    ...
Dask Name: repartition-dataframe, 1 graph layer
how = 'inner', disk = False

    @pytest.mark.parametrize("how", ["inner", "outer", "left", "right"])
    @pytest.mark.parametrize("disk", [True, False])
    @gen_cluster(client=True)
    async def test_merge(c, s, a, b, how, disk):
        A = pd.DataFrame({"x": [1, 2, 3, 4, 5, 6], "y": [1, 1, 2, 2, 3, 4]})
        a = dd.repartition(A, [0, 4, 5])
    
        B = pd.DataFrame({"y": [1, 3, 4, 4, 5, 6], "z": [6, 5, 4, 3, 2, 1]})
        b = dd.repartition(B, [0, 2, 5])
    
        with dask.config.set({"dataframe.shuffle.method": "p2p"}):
            with dask.config.set({"distributed.p2p.storage.disk": disk}):
                joined = dd.merge(a, b, left_index=True, right_index=True, how=how)
            res = await c.compute(joined)
            assert_eq(
                res,
                pd.merge(A, B, left_index=True, right_index=True, how=how),
            )
>           joined = dd.merge(a, b, on="y", how=how)

distributed\shuffle\tests\test_merge.py:190: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\utils.py:259: in wrapper
    return func(*args, **kwargs)
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\dataframe\multi.py:745: in merge
    return hash_join(
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\utils.py:259: in wrapper
    return func(*args, **kwargs)
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\dataframe\multi.py:356: in hash_join
    return hash_join_p2p(
distributed\shuffle\_merge.py:110: in hash_join_p2p
    disk: bool = dask.config.get("distributed.p2p.disk")
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

key = 'distributed.p2p.disk', default = <no_default>
config = {'admin': {'traceback': {'shorten': ['concurrent[\\\\\\/]futures[\\\\\\/]', 'dask[\\\\\\/](base|core|local|multiproces...comm': {'compression': False, 'default-scheme': 'tcp', 'offload': '10MiB', 'require-encryption': None, ...}, ...}, ...}
override_with = None

    def get(
        key: str,
        default: Any = no_default,
        config: dict = config,
        override_with: Any = None,
    ) -> Any:
        """
        Get elements from global config
    
        If ``override_with`` is not None this value will be passed straight back.
        Useful for getting kwarg defaults from Dask config.
    
        Use '.' for nested access
    
        Examples
        --------
        >>> from dask import config
        >>> config.get('foo')  # doctest: +SKIP
        {'x': 1, 'y': 2}
    
        >>> config.get('foo.x')  # doctest: +SKIP
        1
    
        >>> config.get('foo.x.y', default=123)  # doctest: +SKIP
        123
    
        >>> config.get('foo.y', override_with=None)  # doctest: +SKIP
        2
    
        >>> config.get('foo.y', override_with=3)  # doctest: +SKIP
        3
    
        See Also
        --------
        dask.config.set
        """
        if override_with is not None:
            return override_with
        keys = key.split(".")
        result = config
        for k in keys:
            k = canonical_name(k, result)
            try:
>               result = result[k]
E               KeyError: 'disk'

C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\config.py:568: KeyError

KeyError: 'disk'
c = <Client: No scheduler connected>
s = <Scheduler 'tcp://127.0.0.1:53545', workers: 0, cores: 0, tasks: 0>
a = Dask DataFrame Structure:
                   x      y
npartitions=2              
0              int64  int64
4                ...    ...
5                ...    ...
Dask Name: repartition-dataframe, 1 graph layer
b = Dask DataFrame Structure:
                   y      z
npartitions=2              
0              int64  int64
2                ...    ...
5                ...    ...
Dask Name: repartition-dataframe, 1 graph layer
how = 'outer', disk = False

    @pytest.mark.parametrize("how", ["inner", "outer", "left", "right"])
    @pytest.mark.parametrize("disk", [True, False])
    @gen_cluster(client=True)
    async def test_merge(c, s, a, b, how, disk):
        A = pd.DataFrame({"x": [1, 2, 3, 4, 5, 6], "y": [1, 1, 2, 2, 3, 4]})
        a = dd.repartition(A, [0, 4, 5])
    
        B = pd.DataFrame({"y": [1, 3, 4, 4, 5, 6], "z": [6, 5, 4, 3, 2, 1]})
        b = dd.repartition(B, [0, 2, 5])
    
        with dask.config.set({"dataframe.shuffle.method": "p2p"}):
            with dask.config.set({"distributed.p2p.storage.disk": disk}):
                joined = dd.merge(a, b, left_index=True, right_index=True, how=how)
            res = await c.compute(joined)
            assert_eq(
                res,
                pd.merge(A, B, left_index=True, right_index=True, how=how),
            )
>           joined = dd.merge(a, b, on="y", how=how)

distributed\shuffle\tests\test_merge.py:190: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\utils.py:259: in wrapper
    return func(*args, **kwargs)
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\dataframe\multi.py:745: in merge
    return hash_join(
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\utils.py:259: in wrapper
    return func(*args, **kwargs)
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\dataframe\multi.py:356: in hash_join
    return hash_join_p2p(
distributed\shuffle\_merge.py:110: in hash_join_p2p
    disk: bool = dask.config.get("distributed.p2p.disk")
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

key = 'distributed.p2p.disk', default = <no_default>
config = {'admin': {'traceback': {'shorten': ['concurrent[\\\\\\/]futures[\\\\\\/]', 'dask[\\\\\\/](base|core|local|multiproces...comm': {'compression': False, 'default-scheme': 'tcp', 'offload': '10MiB', 'require-encryption': None, ...}, ...}, ...}
override_with = None

    def get(
        key: str,
        default: Any = no_default,
        config: dict = config,
        override_with: Any = None,
    ) -> Any:
        """
        Get elements from global config
    
        If ``override_with`` is not None this value will be passed straight back.
        Useful for getting kwarg defaults from Dask config.
    
        Use '.' for nested access
    
        Examples
        --------
        >>> from dask import config
        >>> config.get('foo')  # doctest: +SKIP
        {'x': 1, 'y': 2}
    
        >>> config.get('foo.x')  # doctest: +SKIP
        1
    
        >>> config.get('foo.x.y', default=123)  # doctest: +SKIP
        123
    
        >>> config.get('foo.y', override_with=None)  # doctest: +SKIP
        2
    
        >>> config.get('foo.y', override_with=3)  # doctest: +SKIP
        3
    
        See Also
        --------
        dask.config.set
        """
        if override_with is not None:
            return override_with
        keys = key.split(".")
        result = config
        for k in keys:
            k = canonical_name(k, result)
            try:
>               result = result[k]
E               KeyError: 'disk'

C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\config.py:568: KeyError

KeyError: 'disk'
c = <Client: No scheduler connected>
s = <Scheduler 'tcp://127.0.0.1:53562', workers: 0, cores: 0, tasks: 0>
a = Dask DataFrame Structure:
                   x      y
npartitions=2              
0              int64  int64
4                ...    ...
5                ...    ...
Dask Name: repartition-dataframe, 1 graph layer
b = Dask DataFrame Structure:
                   y      z
npartitions=2              
0              int64  int64
2                ...    ...
5                ...    ...
Dask Name: repartition-dataframe, 1 graph layer
how = 'left', disk = False

    @pytest.mark.parametrize("how", ["inner", "outer", "left", "right"])
    @pytest.mark.parametrize("disk", [True, False])
    @gen_cluster(client=True)
    async def test_merge(c, s, a, b, how, disk):
        A = pd.DataFrame({"x": [1, 2, 3, 4, 5, 6], "y": [1, 1, 2, 2, 3, 4]})
        a = dd.repartition(A, [0, 4, 5])
    
        B = pd.DataFrame({"y": [1, 3, 4, 4, 5, 6], "z": [6, 5, 4, 3, 2, 1]})
        b = dd.repartition(B, [0, 2, 5])
    
        with dask.config.set({"dataframe.shuffle.method": "p2p"}):
            with dask.config.set({"distributed.p2p.storage.disk": disk}):
                joined = dd.merge(a, b, left_index=True, right_index=True, how=how)
            res = await c.compute(joined)
            assert_eq(
                res,
                pd.merge(A, B, left_index=True, right_index=True, how=how),
            )
>           joined = dd.merge(a, b, on="y", how=how)

distributed\shuffle\tests\test_merge.py:190: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\utils.py:259: in wrapper
    return func(*args, **kwargs)
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\dataframe\multi.py:745: in merge
    return hash_join(
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\utils.py:259: in wrapper
    return func(*args, **kwargs)
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\dataframe\multi.py:356: in hash_join
    return hash_join_p2p(
distributed\shuffle\_merge.py:110: in hash_join_p2p
    disk: bool = dask.config.get("distributed.p2p.disk")
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

key = 'distributed.p2p.disk', default = <no_default>
config = {'admin': {'traceback': {'shorten': ['concurrent[\\\\\\/]futures[\\\\\\/]', 'dask[\\\\\\/](base|core|local|multiproces...comm': {'compression': False, 'default-scheme': 'tcp', 'offload': '10MiB', 'require-encryption': None, ...}, ...}, ...}
override_with = None

    def get(
        key: str,
        default: Any = no_default,
        config: dict = config,
        override_with: Any = None,
    ) -> Any:
        """
        Get elements from global config
    
        If ``override_with`` is not None this value will be passed straight back.
        Useful for getting kwarg defaults from Dask config.
    
        Use '.' for nested access
    
        Examples
        --------
        >>> from dask import config
        >>> config.get('foo')  # doctest: +SKIP
        {'x': 1, 'y': 2}
    
        >>> config.get('foo.x')  # doctest: +SKIP
        1
    
        >>> config.get('foo.x.y', default=123)  # doctest: +SKIP
        123
    
        >>> config.get('foo.y', override_with=None)  # doctest: +SKIP
        2
    
        >>> config.get('foo.y', override_with=3)  # doctest: +SKIP
        3
    
        See Also
        --------
        dask.config.set
        """
        if override_with is not None:
            return override_with
        keys = key.split(".")
        result = config
        for k in keys:
            k = canonical_name(k, result)
            try:
>               result = result[k]
E               KeyError: 'disk'

C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\config.py:568: KeyError

KeyError: 'disk'
c = <Client: No scheduler connected>
s = <Scheduler 'tcp://127.0.0.1:53580', workers: 0, cores: 0, tasks: 0>
a = Dask DataFrame Structure:
                   x      y
npartitions=2              
0              int64  int64
4                ...    ...
5                ...    ...
Dask Name: repartition-dataframe, 1 graph layer
b = Dask DataFrame Structure:
                   y      z
npartitions=2              
0              int64  int64
2                ...    ...
5                ...    ...
Dask Name: repartition-dataframe, 1 graph layer
how = 'right', disk = False

    @pytest.mark.parametrize("how", ["inner", "outer", "left", "right"])
    @pytest.mark.parametrize("disk", [True, False])
    @gen_cluster(client=True)
    async def test_merge(c, s, a, b, how, disk):
        A = pd.DataFrame({"x": [1, 2, 3, 4, 5, 6], "y": [1, 1, 2, 2, 3, 4]})
        a = dd.repartition(A, [0, 4, 5])
    
        B = pd.DataFrame({"y": [1, 3, 4, 4, 5, 6], "z": [6, 5, 4, 3, 2, 1]})
        b = dd.repartition(B, [0, 2, 5])
    
        with dask.config.set({"dataframe.shuffle.method": "p2p"}):
            with dask.config.set({"distributed.p2p.storage.disk": disk}):
                joined = dd.merge(a, b, left_index=True, right_index=True, how=how)
            res = await c.compute(joined)
            assert_eq(
                res,
                pd.merge(A, B, left_index=True, right_index=True, how=how),
            )
>           joined = dd.merge(a, b, on="y", how=how)

distributed\shuffle\tests\test_merge.py:190: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\utils.py:259: in wrapper
    return func(*args, **kwargs)
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\dataframe\multi.py:745: in merge
    return hash_join(
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\utils.py:259: in wrapper
    return func(*args, **kwargs)
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\dataframe\multi.py:356: in hash_join
    return hash_join_p2p(
distributed\shuffle\_merge.py:110: in hash_join_p2p
    disk: bool = dask.config.get("distributed.p2p.disk")
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

key = 'distributed.p2p.disk', default = <no_default>
config = {'admin': {'traceback': {'shorten': ['concurrent[\\\\\\/]futures[\\\\\\/]', 'dask[\\\\\\/](base|core|local|multiproces...comm': {'compression': False, 'default-scheme': 'tcp', 'offload': '10MiB', 'require-encryption': None, ...}, ...}, ...}
override_with = None

    def get(
        key: str,
        default: Any = no_default,
        config: dict = config,
        override_with: Any = None,
    ) -> Any:
        """
        Get elements from global config
    
        If ``override_with`` is not None this value will be passed straight back.
        Useful for getting kwarg defaults from Dask config.
    
        Use '.' for nested access
    
        Examples
        --------
        >>> from dask import config
        >>> config.get('foo')  # doctest: +SKIP
        {'x': 1, 'y': 2}
    
        >>> config.get('foo.x')  # doctest: +SKIP
        1
    
        >>> config.get('foo.x.y', default=123)  # doctest: +SKIP
        123
    
        >>> config.get('foo.y', override_with=None)  # doctest: +SKIP
        2
    
        >>> config.get('foo.y', override_with=3)  # doctest: +SKIP
        3
    
        See Also
        --------
        dask.config.set
        """
        if override_with is not None:
            return override_with
        keys = key.split(".")
        result = config
        for k in keys:
            k = canonical_name(k, result)
            try:
>               result = result[k]
E               KeyError: 'disk'

C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\config.py:568: KeyError

KeyError: 'disk'
c = <Client: No scheduler connected>
s = <Scheduler 'tcp://127.0.0.1:53597', workers: 0, cores: 0, tasks: 0>
a = <Worker 'tcp://127.0.0.1:53598', name: 0, status: closed, stored: 0, running: 0/1, ready: 0, comm: 0, waiting: 0>
b = <Worker 'tcp://127.0.0.1:53601', name: 1, status: closed, stored: 0, running: 0/2, ready: 0, comm: 0, waiting: 0>
how = 'inner'

    @pytest.mark.slow
    @gen_cluster(client=True, timeout=120)
    @pytest.mark.parametrize("how", ["inner", "outer", "left", "right"])
    async def test_merge_by_multiple_columns(c, s, a, b, how):
        # warnings here from pandas
        pdf1l = pd.DataFrame(
            {
                "a": list("abcdefghij"),
                "b": list("abcdefghij"),
                "c": [1, 2, 3, 4, 5, 6, 7, 8, 9, 10],
            },
            index=list("abcdefghij"),
        )
        pdf1r = pd.DataFrame(
            {
                "d": list("abcdefghij"),
                "e": list("abcdefghij"),
                "f": [10, 9, 8, 7, 6, 5, 4, 3, 2, 1],
            },
            index=list("abcdefghij"),
        )
    
        pdf2l = pd.DataFrame(
            {
                "a": list("abcdeabcde"),
                "b": list("abcabcabca"),
                "c": [1, 2, 3, 4, 5, 6, 7, 8, 9, 10],
            },
            index=list("abcdefghij"),
        )
        pdf2r = pd.DataFrame(
            {
                "d": list("edcbaedcba"),
                "e": list("aaabbbcccd"),
                "f": [10, 9, 8, 7, 6, 5, 4, 3, 2, 1],
            },
            index=list("fghijklmno"),
        )
    
        pdf3l = pd.DataFrame(
            {
                "a": list("aaaaaaaaaa"),
                "b": list("aaaaaaaaaa"),
                "c": [1, 2, 3, 4, 5, 6, 7, 8, 9, 10],
            },
            index=list("abcdefghij"),
        )
        pdf3r = pd.DataFrame(
            {
                "d": list("aaabbbccaa"),
                "e": list("abbbbbbbbb"),
                "f": [10, 9, 8, 7, 6, 5, 4, 3, 2, 1],
            },
            index=list("ABCDEFGHIJ"),
        )
    
        for pdl, pdr in [(pdf1l, pdf1r), (pdf2l, pdf2r), (pdf3l, pdf3r)]:
            for lpart, rpart in [(2, 2), (3, 2), (2, 3)]:
                ddl = dd.from_pandas(pdl, lpart)
                ddr = dd.from_pandas(pdr, rpart)
    
                with dask.config.set({"dataframe.shuffle.method": "p2p"}):
                    expected = pdl.join(pdr, how=how)
                    assert_eq(
                        await c.compute(ddl.join(ddr, how=how)),
                        expected,
                        # FIXME: There's an discrepancy with an empty index for
                        # pandas=2.0 (xref https://github.com/dask/dask/issues/9957).
                        # Temporarily avoid index check until the discrepancy is fixed.
                        check_index=not (PANDAS_GE_200 and expected.index.empty),
                    )
    
                    expected = pdr.join(pdl, how=how)
                    assert_eq(
                        await c.compute(ddr.join(ddl, how=how)),
                        expected,
                        # FIXME: There's an discrepancy with an empty index for
                        # pandas=2.0 (xref https://github.com/dask/dask/issues/9957).
                        # Temporarily avoid index check until the discrepancy is fixed.
                        check_index=not (PANDAS_GE_200 and expected.index.empty),
                    )
    
                    expected = pd.merge(
                        pdl, pdr, how=how, left_index=True, right_index=True
                    )
                    assert_eq(
                        await c.compute(
                            dd.merge(
                                ddl,
                                ddr,
                                how=how,
                                left_index=True,
                                right_index=True,
                            )
                        ),
                        expected,
                        # FIXME: There's an discrepancy with an empty index for
                        # pandas=2.0 (xref https://github.com/dask/dask/issues/9957).
                        # Temporarily avoid index check until the discrepancy is fixed.
                        check_index=not (PANDAS_GE_200 and expected.index.empty),
                    )
    
                    expected = pd.merge(
                        pdr, pdl, how=how, left_index=True, right_index=True
                    )
                    assert_eq(
                        await c.compute(
                            dd.merge(
                                ddr,
                                ddl,
                                how=how,
                                left_index=True,
                                right_index=True,
                            )
                        ),
                        expected,
                        # FIXME: There's an discrepancy with an empty index for
                        # pandas=2.0 (xref https://github.com/dask/dask/issues/9957).
                        # Temporarily avoid index check until the discrepancy is fixed.
                        check_index=not (PANDAS_GE_200 and expected.index.empty),
                    )
    
                    # hash join
                    list_eq(
                        await c.compute(
>                           dd.merge(
                                ddl,
                                ddr,
                                how=how,
                                left_on="a",
                                right_on="d",
                            )
                        ),
                        pd.merge(pdl, pdr, how=how, left_on="a", right_on="d"),
                    )

distributed\shuffle\tests\test_merge.py:396: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\utils.py:259: in wrapper
    return func(*args, **kwargs)
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\dataframe\multi.py:745: in merge
    return hash_join(
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\utils.py:259: in wrapper
    return func(*args, **kwargs)
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\dataframe\multi.py:356: in hash_join
    return hash_join_p2p(
distributed\shuffle\_merge.py:110: in hash_join_p2p
    disk: bool = dask.config.get("distributed.p2p.disk")
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

key = 'distributed.p2p.disk', default = <no_default>
config = {'admin': {'traceback': {'shorten': ['concurrent[\\\\\\/]futures[\\\\\\/]', 'dask[\\\\\\/](base|core|local|multiproces...comm': {'compression': False, 'default-scheme': 'tcp', 'offload': '10MiB', 'require-encryption': None, ...}, ...}, ...}
override_with = None

    def get(
        key: str,
        default: Any = no_default,
        config: dict = config,
        override_with: Any = None,
    ) -> Any:
        """
        Get elements from global config
    
        If ``override_with`` is not None this value will be passed straight back.
        Useful for getting kwarg defaults from Dask config.
    
        Use '.' for nested access
    
        Examples
        --------
        >>> from dask import config
        >>> config.get('foo')  # doctest: +SKIP
        {'x': 1, 'y': 2}
    
        >>> config.get('foo.x')  # doctest: +SKIP
        1
    
        >>> config.get('foo.x.y', default=123)  # doctest: +SKIP
        123
    
        >>> config.get('foo.y', override_with=None)  # doctest: +SKIP
        2
    
        >>> config.get('foo.y', override_with=3)  # doctest: +SKIP
        3
    
        See Also
        --------
        dask.config.set
        """
        if override_with is not None:
            return override_with
        keys = key.split(".")
        result = config
        for k in keys:
            k = canonical_name(k, result)
            try:
>               result = result[k]
E               KeyError: 'disk'

C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\config.py:568: KeyError

KeyError: 'disk'
c = <Client: No scheduler connected>
s = <Scheduler 'tcp://127.0.0.1:53615', workers: 0, cores: 0, tasks: 0>
a = <Worker 'tcp://127.0.0.1:53616', name: 0, status: closed, stored: 0, running: 0/1, ready: 0, comm: 0, waiting: 0>
b = <Worker 'tcp://127.0.0.1:53619', name: 1, status: closed, stored: 0, running: 0/2, ready: 0, comm: 0, waiting: 0>
how = 'outer'

    @pytest.mark.slow
    @gen_cluster(client=True, timeout=120)
    @pytest.mark.parametrize("how", ["inner", "outer", "left", "right"])
    async def test_merge_by_multiple_columns(c, s, a, b, how):
        # warnings here from pandas
        pdf1l = pd.DataFrame(
            {
                "a": list("abcdefghij"),
                "b": list("abcdefghij"),
                "c": [1, 2, 3, 4, 5, 6, 7, 8, 9, 10],
            },
            index=list("abcdefghij"),
        )
        pdf1r = pd.DataFrame(
            {
                "d": list("abcdefghij"),
                "e": list("abcdefghij"),
                "f": [10, 9, 8, 7, 6, 5, 4, 3, 2, 1],
            },
            index=list("abcdefghij"),
        )
    
        pdf2l = pd.DataFrame(
            {
                "a": list("abcdeabcde"),
                "b": list("abcabcabca"),
                "c": [1, 2, 3, 4, 5, 6, 7, 8, 9, 10],
            },
            index=list("abcdefghij"),
        )
        pdf2r = pd.DataFrame(
            {
                "d": list("edcbaedcba"),
                "e": list("aaabbbcccd"),
                "f": [10, 9, 8, 7, 6, 5, 4, 3, 2, 1],
            },
            index=list("fghijklmno"),
        )
    
        pdf3l = pd.DataFrame(
            {
                "a": list("aaaaaaaaaa"),
                "b": list("aaaaaaaaaa"),
                "c": [1, 2, 3, 4, 5, 6, 7, 8, 9, 10],
            },
            index=list("abcdefghij"),
        )
        pdf3r = pd.DataFrame(
            {
                "d": list("aaabbbccaa"),
                "e": list("abbbbbbbbb"),
                "f": [10, 9, 8, 7, 6, 5, 4, 3, 2, 1],
            },
            index=list("ABCDEFGHIJ"),
        )
    
        for pdl, pdr in [(pdf1l, pdf1r), (pdf2l, pdf2r), (pdf3l, pdf3r)]:
            for lpart, rpart in [(2, 2), (3, 2), (2, 3)]:
                ddl = dd.from_pandas(pdl, lpart)
                ddr = dd.from_pandas(pdr, rpart)
    
                with dask.config.set({"dataframe.shuffle.method": "p2p"}):
                    expected = pdl.join(pdr, how=how)
                    assert_eq(
                        await c.compute(ddl.join(ddr, how=how)),
                        expected,
                        # FIXME: There's an discrepancy with an empty index for
                        # pandas=2.0 (xref https://github.com/dask/dask/issues/9957).
                        # Temporarily avoid index check until the discrepancy is fixed.
                        check_index=not (PANDAS_GE_200 and expected.index.empty),
                    )
    
                    expected = pdr.join(pdl, how=how)
                    assert_eq(
                        await c.compute(ddr.join(ddl, how=how)),
                        expected,
                        # FIXME: There's an discrepancy with an empty index for
                        # pandas=2.0 (xref https://github.com/dask/dask/issues/9957).
                        # Temporarily avoid index check until the discrepancy is fixed.
                        check_index=not (PANDAS_GE_200 and expected.index.empty),
                    )
    
                    expected = pd.merge(
                        pdl, pdr, how=how, left_index=True, right_index=True
                    )
                    assert_eq(
                        await c.compute(
                            dd.merge(
                                ddl,
                                ddr,
                                how=how,
                                left_index=True,
                                right_index=True,
                            )
                        ),
                        expected,
                        # FIXME: There's an discrepancy with an empty index for
                        # pandas=2.0 (xref https://github.com/dask/dask/issues/9957).
                        # Temporarily avoid index check until the discrepancy is fixed.
                        check_index=not (PANDAS_GE_200 and expected.index.empty),
                    )
    
                    expected = pd.merge(
                        pdr, pdl, how=how, left_index=True, right_index=True
                    )
                    assert_eq(
                        await c.compute(
                            dd.merge(
                                ddr,
                                ddl,
                                how=how,
                                left_index=True,
                                right_index=True,
                            )
                        ),
                        expected,
                        # FIXME: There's an discrepancy with an empty index for
                        # pandas=2.0 (xref https://github.com/dask/dask/issues/9957).
                        # Temporarily avoid index check until the discrepancy is fixed.
                        check_index=not (PANDAS_GE_200 and expected.index.empty),
                    )
    
                    # hash join
                    list_eq(
                        await c.compute(
>                           dd.merge(
                                ddl,
                                ddr,
                                how=how,
                                left_on="a",
                                right_on="d",
                            )
                        ),
                        pd.merge(pdl, pdr, how=how, left_on="a", right_on="d"),
                    )

distributed\shuffle\tests\test_merge.py:396: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\utils.py:259: in wrapper
    return func(*args, **kwargs)
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\dataframe\multi.py:745: in merge
    return hash_join(
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\utils.py:259: in wrapper
    return func(*args, **kwargs)
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\dataframe\multi.py:356: in hash_join
    return hash_join_p2p(
distributed\shuffle\_merge.py:110: in hash_join_p2p
    disk: bool = dask.config.get("distributed.p2p.disk")
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

key = 'distributed.p2p.disk', default = <no_default>
config = {'admin': {'traceback': {'shorten': ['concurrent[\\\\\\/]futures[\\\\\\/]', 'dask[\\\\\\/](base|core|local|multiproces...comm': {'compression': False, 'default-scheme': 'tcp', 'offload': '10MiB', 'require-encryption': None, ...}, ...}, ...}
override_with = None

    def get(
        key: str,
        default: Any = no_default,
        config: dict = config,
        override_with: Any = None,
    ) -> Any:
        """
        Get elements from global config
    
        If ``override_with`` is not None this value will be passed straight back.
        Useful for getting kwarg defaults from Dask config.
    
        Use '.' for nested access
    
        Examples
        --------
        >>> from dask import config
        >>> config.get('foo')  # doctest: +SKIP
        {'x': 1, 'y': 2}
    
        >>> config.get('foo.x')  # doctest: +SKIP
        1
    
        >>> config.get('foo.x.y', default=123)  # doctest: +SKIP
        123
    
        >>> config.get('foo.y', override_with=None)  # doctest: +SKIP
        2
    
        >>> config.get('foo.y', override_with=3)  # doctest: +SKIP
        3
    
        See Also
        --------
        dask.config.set
        """
        if override_with is not None:
            return override_with
        keys = key.split(".")
        result = config
        for k in keys:
            k = canonical_name(k, result)
            try:
>               result = result[k]
E               KeyError: 'disk'

C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\config.py:568: KeyError

KeyError: 'disk'
c = <Client: No scheduler connected>
s = <Scheduler 'tcp://127.0.0.1:53632', workers: 0, cores: 0, tasks: 0>
a = <Worker 'tcp://127.0.0.1:53633', name: 0, status: closed, stored: 0, running: 0/1, ready: 0, comm: 0, waiting: 0>
b = <Worker 'tcp://127.0.0.1:53636', name: 1, status: closed, stored: 0, running: 0/2, ready: 0, comm: 0, waiting: 0>
how = 'left'

    @pytest.mark.slow
    @gen_cluster(client=True, timeout=120)
    @pytest.mark.parametrize("how", ["inner", "outer", "left", "right"])
    async def test_merge_by_multiple_columns(c, s, a, b, how):
        # warnings here from pandas
        pdf1l = pd.DataFrame(
            {
                "a": list("abcdefghij"),
                "b": list("abcdefghij"),
                "c": [1, 2, 3, 4, 5, 6, 7, 8, 9, 10],
            },
            index=list("abcdefghij"),
        )
        pdf1r = pd.DataFrame(
            {
                "d": list("abcdefghij"),
                "e": list("abcdefghij"),
                "f": [10, 9, 8, 7, 6, 5, 4, 3, 2, 1],
            },
            index=list("abcdefghij"),
        )
    
        pdf2l = pd.DataFrame(
            {
                "a": list("abcdeabcde"),
                "b": list("abcabcabca"),
                "c": [1, 2, 3, 4, 5, 6, 7, 8, 9, 10],
            },
            index=list("abcdefghij"),
        )
        pdf2r = pd.DataFrame(
            {
                "d": list("edcbaedcba"),
                "e": list("aaabbbcccd"),
                "f": [10, 9, 8, 7, 6, 5, 4, 3, 2, 1],
            },
            index=list("fghijklmno"),
        )
    
        pdf3l = pd.DataFrame(
            {
                "a": list("aaaaaaaaaa"),
                "b": list("aaaaaaaaaa"),
                "c": [1, 2, 3, 4, 5, 6, 7, 8, 9, 10],
            },
            index=list("abcdefghij"),
        )
        pdf3r = pd.DataFrame(
            {
                "d": list("aaabbbccaa"),
                "e": list("abbbbbbbbb"),
                "f": [10, 9, 8, 7, 6, 5, 4, 3, 2, 1],
            },
            index=list("ABCDEFGHIJ"),
        )
    
        for pdl, pdr in [(pdf1l, pdf1r), (pdf2l, pdf2r), (pdf3l, pdf3r)]:
            for lpart, rpart in [(2, 2), (3, 2), (2, 3)]:
                ddl = dd.from_pandas(pdl, lpart)
                ddr = dd.from_pandas(pdr, rpart)
    
                with dask.config.set({"dataframe.shuffle.method": "p2p"}):
                    expected = pdl.join(pdr, how=how)
                    assert_eq(
                        await c.compute(ddl.join(ddr, how=how)),
                        expected,
                        # FIXME: There's an discrepancy with an empty index for
                        # pandas=2.0 (xref https://github.com/dask/dask/issues/9957).
                        # Temporarily avoid index check until the discrepancy is fixed.
                        check_index=not (PANDAS_GE_200 and expected.index.empty),
                    )
    
                    expected = pdr.join(pdl, how=how)
                    assert_eq(
                        await c.compute(ddr.join(ddl, how=how)),
                        expected,
                        # FIXME: There's an discrepancy with an empty index for
                        # pandas=2.0 (xref https://github.com/dask/dask/issues/9957).
                        # Temporarily avoid index check until the discrepancy is fixed.
                        check_index=not (PANDAS_GE_200 and expected.index.empty),
                    )
    
                    expected = pd.merge(
                        pdl, pdr, how=how, left_index=True, right_index=True
                    )
                    assert_eq(
                        await c.compute(
                            dd.merge(
                                ddl,
                                ddr,
                                how=how,
                                left_index=True,
                                right_index=True,
                            )
                        ),
                        expected,
                        # FIXME: There's an discrepancy with an empty index for
                        # pandas=2.0 (xref https://github.com/dask/dask/issues/9957).
                        # Temporarily avoid index check until the discrepancy is fixed.
                        check_index=not (PANDAS_GE_200 and expected.index.empty),
                    )
    
                    expected = pd.merge(
                        pdr, pdl, how=how, left_index=True, right_index=True
                    )
                    assert_eq(
                        await c.compute(
                            dd.merge(
                                ddr,
                                ddl,
                                how=how,
                                left_index=True,
                                right_index=True,
                            )
                        ),
                        expected,
                        # FIXME: There's an discrepancy with an empty index for
                        # pandas=2.0 (xref https://github.com/dask/dask/issues/9957).
                        # Temporarily avoid index check until the discrepancy is fixed.
                        check_index=not (PANDAS_GE_200 and expected.index.empty),
                    )
    
                    # hash join
                    list_eq(
                        await c.compute(
>                           dd.merge(
                                ddl,
                                ddr,
                                how=how,
                                left_on="a",
                                right_on="d",
                            )
                        ),
                        pd.merge(pdl, pdr, how=how, left_on="a", right_on="d"),
                    )

distributed\shuffle\tests\test_merge.py:396: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\utils.py:259: in wrapper
    return func(*args, **kwargs)
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\dataframe\multi.py:745: in merge
    return hash_join(
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\utils.py:259: in wrapper
    return func(*args, **kwargs)
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\dataframe\multi.py:356: in hash_join
    return hash_join_p2p(
distributed\shuffle\_merge.py:110: in hash_join_p2p
    disk: bool = dask.config.get("distributed.p2p.disk")
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

key = 'distributed.p2p.disk', default = <no_default>
config = {'admin': {'traceback': {'shorten': ['concurrent[\\\\\\/]futures[\\\\\\/]', 'dask[\\\\\\/](base|core|local|multiproces...comm': {'compression': False, 'default-scheme': 'tcp', 'offload': '10MiB', 'require-encryption': None, ...}, ...}, ...}
override_with = None

    def get(
        key: str,
        default: Any = no_default,
        config: dict = config,
        override_with: Any = None,
    ) -> Any:
        """
        Get elements from global config
    
        If ``override_with`` is not None this value will be passed straight back.
        Useful for getting kwarg defaults from Dask config.
    
        Use '.' for nested access
    
        Examples
        --------
        >>> from dask import config
        >>> config.get('foo')  # doctest: +SKIP
        {'x': 1, 'y': 2}
    
        >>> config.get('foo.x')  # doctest: +SKIP
        1
    
        >>> config.get('foo.x.y', default=123)  # doctest: +SKIP
        123
    
        >>> config.get('foo.y', override_with=None)  # doctest: +SKIP
        2
    
        >>> config.get('foo.y', override_with=3)  # doctest: +SKIP
        3
    
        See Also
        --------
        dask.config.set
        """
        if override_with is not None:
            return override_with
        keys = key.split(".")
        result = config
        for k in keys:
            k = canonical_name(k, result)
            try:
>               result = result[k]
E               KeyError: 'disk'

C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\config.py:568: KeyError

KeyError: 'disk'
c = <Client: No scheduler connected>
s = <Scheduler 'tcp://127.0.0.1:53649', workers: 0, cores: 0, tasks: 0>
a = <Worker 'tcp://127.0.0.1:53650', name: 0, status: closed, stored: 0, running: 0/1, ready: 0, comm: 0, waiting: 0>
b = <Worker 'tcp://127.0.0.1:53653', name: 1, status: closed, stored: 0, running: 0/2, ready: 0, comm: 0, waiting: 0>
how = 'right'

    @pytest.mark.slow
    @gen_cluster(client=True, timeout=120)
    @pytest.mark.parametrize("how", ["inner", "outer", "left", "right"])
    async def test_merge_by_multiple_columns(c, s, a, b, how):
        # warnings here from pandas
        pdf1l = pd.DataFrame(
            {
                "a": list("abcdefghij"),
                "b": list("abcdefghij"),
                "c": [1, 2, 3, 4, 5, 6, 7, 8, 9, 10],
            },
            index=list("abcdefghij"),
        )
        pdf1r = pd.DataFrame(
            {
                "d": list("abcdefghij"),
                "e": list("abcdefghij"),
                "f": [10, 9, 8, 7, 6, 5, 4, 3, 2, 1],
            },
            index=list("abcdefghij"),
        )
    
        pdf2l = pd.DataFrame(
            {
                "a": list("abcdeabcde"),
                "b": list("abcabcabca"),
                "c": [1, 2, 3, 4, 5, 6, 7, 8, 9, 10],
            },
            index=list("abcdefghij"),
        )
        pdf2r = pd.DataFrame(
            {
                "d": list("edcbaedcba"),
                "e": list("aaabbbcccd"),
                "f": [10, 9, 8, 7, 6, 5, 4, 3, 2, 1],
            },
            index=list("fghijklmno"),
        )
    
        pdf3l = pd.DataFrame(
            {
                "a": list("aaaaaaaaaa"),
                "b": list("aaaaaaaaaa"),
                "c": [1, 2, 3, 4, 5, 6, 7, 8, 9, 10],
            },
            index=list("abcdefghij"),
        )
        pdf3r = pd.DataFrame(
            {
                "d": list("aaabbbccaa"),
                "e": list("abbbbbbbbb"),
                "f": [10, 9, 8, 7, 6, 5, 4, 3, 2, 1],
            },
            index=list("ABCDEFGHIJ"),
        )
    
        for pdl, pdr in [(pdf1l, pdf1r), (pdf2l, pdf2r), (pdf3l, pdf3r)]:
            for lpart, rpart in [(2, 2), (3, 2), (2, 3)]:
                ddl = dd.from_pandas(pdl, lpart)
                ddr = dd.from_pandas(pdr, rpart)
    
                with dask.config.set({"dataframe.shuffle.method": "p2p"}):
                    expected = pdl.join(pdr, how=how)
                    assert_eq(
                        await c.compute(ddl.join(ddr, how=how)),
                        expected,
                        # FIXME: There's an discrepancy with an empty index for
                        # pandas=2.0 (xref https://github.com/dask/dask/issues/9957).
                        # Temporarily avoid index check until the discrepancy is fixed.
                        check_index=not (PANDAS_GE_200 and expected.index.empty),
                    )
    
                    expected = pdr.join(pdl, how=how)
                    assert_eq(
                        await c.compute(ddr.join(ddl, how=how)),
                        expected,
                        # FIXME: There's an discrepancy with an empty index for
                        # pandas=2.0 (xref https://github.com/dask/dask/issues/9957).
                        # Temporarily avoid index check until the discrepancy is fixed.
                        check_index=not (PANDAS_GE_200 and expected.index.empty),
                    )
    
                    expected = pd.merge(
                        pdl, pdr, how=how, left_index=True, right_index=True
                    )
                    assert_eq(
                        await c.compute(
                            dd.merge(
                                ddl,
                                ddr,
                                how=how,
                                left_index=True,
                                right_index=True,
                            )
                        ),
                        expected,
                        # FIXME: There's an discrepancy with an empty index for
                        # pandas=2.0 (xref https://github.com/dask/dask/issues/9957).
                        # Temporarily avoid index check until the discrepancy is fixed.
                        check_index=not (PANDAS_GE_200 and expected.index.empty),
                    )
    
                    expected = pd.merge(
                        pdr, pdl, how=how, left_index=True, right_index=True
                    )
                    assert_eq(
                        await c.compute(
                            dd.merge(
                                ddr,
                                ddl,
                                how=how,
                                left_index=True,
                                right_index=True,
                            )
                        ),
                        expected,
                        # FIXME: There's an discrepancy with an empty index for
                        # pandas=2.0 (xref https://github.com/dask/dask/issues/9957).
                        # Temporarily avoid index check until the discrepancy is fixed.
                        check_index=not (PANDAS_GE_200 and expected.index.empty),
                    )
    
                    # hash join
                    list_eq(
                        await c.compute(
>                           dd.merge(
                                ddl,
                                ddr,
                                how=how,
                                left_on="a",
                                right_on="d",
                            )
                        ),
                        pd.merge(pdl, pdr, how=how, left_on="a", right_on="d"),
                    )

distributed\shuffle\tests\test_merge.py:396: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\utils.py:259: in wrapper
    return func(*args, **kwargs)
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\dataframe\multi.py:745: in merge
    return hash_join(
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\utils.py:259: in wrapper
    return func(*args, **kwargs)
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\dataframe\multi.py:356: in hash_join
    return hash_join_p2p(
distributed\shuffle\_merge.py:110: in hash_join_p2p
    disk: bool = dask.config.get("distributed.p2p.disk")
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

key = 'distributed.p2p.disk', default = <no_default>
config = {'admin': {'traceback': {'shorten': ['concurrent[\\\\\\/]futures[\\\\\\/]', 'dask[\\\\\\/](base|core|local|multiproces...comm': {'compression': False, 'default-scheme': 'tcp', 'offload': '10MiB', 'require-encryption': None, ...}, ...}, ...}
override_with = None

    def get(
        key: str,
        default: Any = no_default,
        config: dict = config,
        override_with: Any = None,
    ) -> Any:
        """
        Get elements from global config
    
        If ``override_with`` is not None this value will be passed straight back.
        Useful for getting kwarg defaults from Dask config.
    
        Use '.' for nested access
    
        Examples
        --------
        >>> from dask import config
        >>> config.get('foo')  # doctest: +SKIP
        {'x': 1, 'y': 2}
    
        >>> config.get('foo.x')  # doctest: +SKIP
        1
    
        >>> config.get('foo.x.y', default=123)  # doctest: +SKIP
        123
    
        >>> config.get('foo.y', override_with=None)  # doctest: +SKIP
        2
    
        >>> config.get('foo.y', override_with=3)  # doctest: +SKIP
        3
    
        See Also
        --------
        dask.config.set
        """
        if override_with is not None:
            return override_with
        keys = key.split(".")
        result = config
        for k in keys:
            k = canonical_name(k, result)
            try:
>               result = result[k]
E               KeyError: 'disk'

C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\config.py:568: KeyError

KeyError: 'disk'
c = <Client: No scheduler connected>
s = <Scheduler 'tcp://127.0.0.1:53666', workers: 0, cores: 0, tasks: 0>
a = <Worker 'tcp://127.0.0.1:53667', name: 0, status: closed, stored: 0, running: 0/1, ready: 0, comm: 0, waiting: 0>
b = <Worker 'tcp://127.0.0.1:53670', name: 1, status: closed, stored: 0, running: 0/2, ready: 0, comm: 0, waiting: 0>
how = 'inner'

    @pytest.mark.parametrize("how", ["inner", "left", "right", "outer"])
    @gen_cluster(client=True)
    async def test_index_merge_p2p(c, s, a, b, how):
        pdf_left = pd.DataFrame({"a": [4, 2, 3] * 10, "b": 1}).set_index("a")
        pdf_right = pd.DataFrame({"a": [4, 2, 3] * 10, "c": 1})
    
        left = dd.from_pandas(pdf_left, npartitions=5, sort=False)
        right = dd.from_pandas(pdf_right, npartitions=6)
    
        with dask.config.set({"dataframe.shuffle.method": "p2p"}):
            assert_eq(
>               await c.compute(left.merge(right, how=how, left_index=True, right_on="a")),
                pdf_left.merge(pdf_right, how=how, left_index=True, right_on="a"),
            )

distributed\shuffle\tests\test_merge.py:471: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\utils.py:259: in wrapper
    return func(*args, **kwargs)
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\dataframe\core.py:6010: in merge
    return merge(
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\utils.py:259: in wrapper
    return func(*args, **kwargs)
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\dataframe\multi.py:745: in merge
    return hash_join(
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\utils.py:259: in wrapper
    return func(*args, **kwargs)
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\dataframe\multi.py:356: in hash_join
    return hash_join_p2p(
distributed\shuffle\_merge.py:110: in hash_join_p2p
    disk: bool = dask.config.get("distributed.p2p.disk")
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

key = 'distributed.p2p.disk', default = <no_default>
config = {'admin': {'traceback': {'shorten': ['concurrent[\\\\\\/]futures[\\\\\\/]', 'dask[\\\\\\/](base|core|local|multiproces...comm': {'compression': False, 'default-scheme': 'tcp', 'offload': '10MiB', 'require-encryption': None, ...}, ...}, ...}
override_with = None

    def get(
        key: str,
        default: Any = no_default,
        config: dict = config,
        override_with: Any = None,
    ) -> Any:
        """
        Get elements from global config
    
        If ``override_with`` is not None this value will be passed straight back.
        Useful for getting kwarg defaults from Dask config.
    
        Use '.' for nested access
    
        Examples
        --------
        >>> from dask import config
        >>> config.get('foo')  # doctest: +SKIP
        {'x': 1, 'y': 2}
    
        >>> config.get('foo.x')  # doctest: +SKIP
        1
    
        >>> config.get('foo.x.y', default=123)  # doctest: +SKIP
        123
    
        >>> config.get('foo.y', override_with=None)  # doctest: +SKIP
        2
    
        >>> config.get('foo.y', override_with=3)  # doctest: +SKIP
        3
    
        See Also
        --------
        dask.config.set
        """
        if override_with is not None:
            return override_with
        keys = key.split(".")
        result = config
        for k in keys:
            k = canonical_name(k, result)
            try:
>               result = result[k]
E               KeyError: 'disk'

C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\config.py:568: KeyError

KeyError: 'disk'
c = <Client: No scheduler connected>
s = <Scheduler 'tcp://127.0.0.1:53678', workers: 0, cores: 0, tasks: 0>
a = <Worker 'tcp://127.0.0.1:53679', name: 0, status: closed, stored: 0, running: 0/1, ready: 0, comm: 0, waiting: 0>
b = <Worker 'tcp://127.0.0.1:53682', name: 1, status: closed, stored: 0, running: 0/2, ready: 0, comm: 0, waiting: 0>
how = 'left'

    @pytest.mark.parametrize("how", ["inner", "left", "right", "outer"])
    @gen_cluster(client=True)
    async def test_index_merge_p2p(c, s, a, b, how):
        pdf_left = pd.DataFrame({"a": [4, 2, 3] * 10, "b": 1}).set_index("a")
        pdf_right = pd.DataFrame({"a": [4, 2, 3] * 10, "c": 1})
    
        left = dd.from_pandas(pdf_left, npartitions=5, sort=False)
        right = dd.from_pandas(pdf_right, npartitions=6)
    
        with dask.config.set({"dataframe.shuffle.method": "p2p"}):
            assert_eq(
>               await c.compute(left.merge(right, how=how, left_index=True, right_on="a")),
                pdf_left.merge(pdf_right, how=how, left_index=True, right_on="a"),
            )

distributed\shuffle\tests\test_merge.py:471: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\utils.py:259: in wrapper
    return func(*args, **kwargs)
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\dataframe\core.py:6010: in merge
    return merge(
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\utils.py:259: in wrapper
    return func(*args, **kwargs)
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\dataframe\multi.py:745: in merge
    return hash_join(
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\utils.py:259: in wrapper
    return func(*args, **kwargs)
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\dataframe\multi.py:356: in hash_join
    return hash_join_p2p(
distributed\shuffle\_merge.py:110: in hash_join_p2p
    disk: bool = dask.config.get("distributed.p2p.disk")
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

key = 'distributed.p2p.disk', default = <no_default>
config = {'admin': {'traceback': {'shorten': ['concurrent[\\\\\\/]futures[\\\\\\/]', 'dask[\\\\\\/](base|core|local|multiproces...comm': {'compression': False, 'default-scheme': 'tcp', 'offload': '10MiB', 'require-encryption': None, ...}, ...}, ...}
override_with = None

    def get(
        key: str,
        default: Any = no_default,
        config: dict = config,
        override_with: Any = None,
    ) -> Any:
        """
        Get elements from global config
    
        If ``override_with`` is not None this value will be passed straight back.
        Useful for getting kwarg defaults from Dask config.
    
        Use '.' for nested access
    
        Examples
        --------
        >>> from dask import config
        >>> config.get('foo')  # doctest: +SKIP
        {'x': 1, 'y': 2}
    
        >>> config.get('foo.x')  # doctest: +SKIP
        1
    
        >>> config.get('foo.x.y', default=123)  # doctest: +SKIP
        123
    
        >>> config.get('foo.y', override_with=None)  # doctest: +SKIP
        2
    
        >>> config.get('foo.y', override_with=3)  # doctest: +SKIP
        3
    
        See Also
        --------
        dask.config.set
        """
        if override_with is not None:
            return override_with
        keys = key.split(".")
        result = config
        for k in keys:
            k = canonical_name(k, result)
            try:
>               result = result[k]
E               KeyError: 'disk'

C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\config.py:568: KeyError

KeyError: 'disk'
c = <Client: No scheduler connected>
s = <Scheduler 'tcp://127.0.0.1:53691', workers: 0, cores: 0, tasks: 0>
a = <Worker 'tcp://127.0.0.1:53692', name: 0, status: closed, stored: 0, running: 0/1, ready: 0, comm: 0, waiting: 0>
b = <Worker 'tcp://127.0.0.1:53695', name: 1, status: closed, stored: 0, running: 0/2, ready: 0, comm: 0, waiting: 0>
how = 'right'

    @pytest.mark.parametrize("how", ["inner", "left", "right", "outer"])
    @gen_cluster(client=True)
    async def test_index_merge_p2p(c, s, a, b, how):
        pdf_left = pd.DataFrame({"a": [4, 2, 3] * 10, "b": 1}).set_index("a")
        pdf_right = pd.DataFrame({"a": [4, 2, 3] * 10, "c": 1})
    
        left = dd.from_pandas(pdf_left, npartitions=5, sort=False)
        right = dd.from_pandas(pdf_right, npartitions=6)
    
        with dask.config.set({"dataframe.shuffle.method": "p2p"}):
            assert_eq(
>               await c.compute(left.merge(right, how=how, left_index=True, right_on="a")),
                pdf_left.merge(pdf_right, how=how, left_index=True, right_on="a"),
            )

distributed\shuffle\tests\test_merge.py:471: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\utils.py:259: in wrapper
    return func(*args, **kwargs)
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\dataframe\core.py:6010: in merge
    return merge(
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\utils.py:259: in wrapper
    return func(*args, **kwargs)
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\dataframe\multi.py:745: in merge
    return hash_join(
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\utils.py:259: in wrapper
    return func(*args, **kwargs)
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\dataframe\multi.py:356: in hash_join
    return hash_join_p2p(
distributed\shuffle\_merge.py:110: in hash_join_p2p
    disk: bool = dask.config.get("distributed.p2p.disk")
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

key = 'distributed.p2p.disk', default = <no_default>
config = {'admin': {'traceback': {'shorten': ['concurrent[\\\\\\/]futures[\\\\\\/]', 'dask[\\\\\\/](base|core|local|multiproces...comm': {'compression': False, 'default-scheme': 'tcp', 'offload': '10MiB', 'require-encryption': None, ...}, ...}, ...}
override_with = None

    def get(
        key: str,
        default: Any = no_default,
        config: dict = config,
        override_with: Any = None,
    ) -> Any:
        """
        Get elements from global config
    
        If ``override_with`` is not None this value will be passed straight back.
        Useful for getting kwarg defaults from Dask config.
    
        Use '.' for nested access
    
        Examples
        --------
        >>> from dask import config
        >>> config.get('foo')  # doctest: +SKIP
        {'x': 1, 'y': 2}
    
        >>> config.get('foo.x')  # doctest: +SKIP
        1
    
        >>> config.get('foo.x.y', default=123)  # doctest: +SKIP
        123
    
        >>> config.get('foo.y', override_with=None)  # doctest: +SKIP
        2
    
        >>> config.get('foo.y', override_with=3)  # doctest: +SKIP
        3
    
        See Also
        --------
        dask.config.set
        """
        if override_with is not None:
            return override_with
        keys = key.split(".")
        result = config
        for k in keys:
            k = canonical_name(k, result)
            try:
>               result = result[k]
E               KeyError: 'disk'

C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\config.py:568: KeyError

KeyError: 'disk'
c = <Client: No scheduler connected>
s = <Scheduler 'tcp://127.0.0.1:53703', workers: 0, cores: 0, tasks: 0>
a = <Worker 'tcp://127.0.0.1:53704', name: 0, status: closed, stored: 0, running: 0/1, ready: 0, comm: 0, waiting: 0>
b = <Worker 'tcp://127.0.0.1:53707', name: 1, status: closed, stored: 0, running: 0/2, ready: 0, comm: 0, waiting: 0>
how = 'outer'

    @pytest.mark.parametrize("how", ["inner", "left", "right", "outer"])
    @gen_cluster(client=True)
    async def test_index_merge_p2p(c, s, a, b, how):
        pdf_left = pd.DataFrame({"a": [4, 2, 3] * 10, "b": 1}).set_index("a")
        pdf_right = pd.DataFrame({"a": [4, 2, 3] * 10, "c": 1})
    
        left = dd.from_pandas(pdf_left, npartitions=5, sort=False)
        right = dd.from_pandas(pdf_right, npartitions=6)
    
        with dask.config.set({"dataframe.shuffle.method": "p2p"}):
            assert_eq(
>               await c.compute(left.merge(right, how=how, left_index=True, right_on="a")),
                pdf_left.merge(pdf_right, how=how, left_index=True, right_on="a"),
            )

distributed\shuffle\tests\test_merge.py:471: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\utils.py:259: in wrapper
    return func(*args, **kwargs)
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\dataframe\core.py:6010: in merge
    return merge(
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\utils.py:259: in wrapper
    return func(*args, **kwargs)
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\dataframe\multi.py:745: in merge
    return hash_join(
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\utils.py:259: in wrapper
    return func(*args, **kwargs)
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\dataframe\multi.py:356: in hash_join
    return hash_join_p2p(
distributed\shuffle\_merge.py:110: in hash_join_p2p
    disk: bool = dask.config.get("distributed.p2p.disk")
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

key = 'distributed.p2p.disk', default = <no_default>
config = {'admin': {'traceback': {'shorten': ['concurrent[\\\\\\/]futures[\\\\\\/]', 'dask[\\\\\\/](base|core|local|multiproces...comm': {'compression': False, 'default-scheme': 'tcp', 'offload': '10MiB', 'require-encryption': None, ...}, ...}, ...}
override_with = None

    def get(
        key: str,
        default: Any = no_default,
        config: dict = config,
        override_with: Any = None,
    ) -> Any:
        """
        Get elements from global config
    
        If ``override_with`` is not None this value will be passed straight back.
        Useful for getting kwarg defaults from Dask config.
    
        Use '.' for nested access
    
        Examples
        --------
        >>> from dask import config
        >>> config.get('foo')  # doctest: +SKIP
        {'x': 1, 'y': 2}
    
        >>> config.get('foo.x')  # doctest: +SKIP
        1
    
        >>> config.get('foo.x.y', default=123)  # doctest: +SKIP
        123
    
        >>> config.get('foo.y', override_with=None)  # doctest: +SKIP
        2
    
        >>> config.get('foo.y', override_with=3)  # doctest: +SKIP
        3
    
        See Also
        --------
        dask.config.set
        """
        if override_with is not None:
            return override_with
        keys = key.split(".")
        result = config
        for k in keys:
            k = canonical_name(k, result)
            try:
>               result = result[k]
E               KeyError: 'disk'

C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\config.py:568: KeyError

KeyError: 'disk'
c = <Client: No scheduler connected>
s = <Scheduler 'tcp://127.0.0.1:53715', workers: 0, cores: 0, tasks: 0>
a = <Worker 'tcp://127.0.0.1:53716', name: 0, status: closed, stored: 0, running: 0/1, ready: 0, comm: 0, waiting: 0>

    @mock.patch(
        "distributed.shuffle._worker_plugin._ShuffleRunManager",
        LimitedGetOrCreateShuffleRunManager,
    )
    @gen_cluster(client=True, nthreads=[("", 1)])
    async def test_merge_does_not_deadlock_if_worker_joins(c, s, a):
        """Regression test for https://github.com/dask/distributed/issues/8411"""
        pdf1 = pd.DataFrame({"a": range(100), "b": range(0, 200, 2)})
        pdf2 = pd.DataFrame({"x": range(200), "y": [1, 2, 3, 4] * 50})
        df1 = dd.from_pandas(pdf1, npartitions=10)
        df2 = dd.from_pandas(pdf2, npartitions=20)
    
        run_manager_A = a.plugins["shuffle"].shuffle_runs
    
        with dask.config.set({"dataframe.shuffle.method": "p2p"}):
>           joined = dd.merge(df1, df2, left_on="a", right_on="x")

distributed\shuffle\tests\test_merge.py:516: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\utils.py:259: in wrapper
    return func(*args, **kwargs)
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\dataframe\multi.py:745: in merge
    return hash_join(
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\utils.py:259: in wrapper
    return func(*args, **kwargs)
C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\dataframe\multi.py:356: in hash_join
    return hash_join_p2p(
distributed\shuffle\_merge.py:110: in hash_join_p2p
    disk: bool = dask.config.get("distributed.p2p.disk")
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

key = 'distributed.p2p.disk', default = <no_default>
config = {'admin': {'traceback': {'shorten': ['concurrent[\\\\\\/]futures[\\\\\\/]', 'dask[\\\\\\/](base|core|local|multiproces...comm': {'compression': False, 'default-scheme': 'tcp', 'offload': '10MiB', 'require-encryption': None, ...}, ...}, ...}
override_with = None

    def get(
        key: str,
        default: Any = no_default,
        config: dict = config,
        override_with: Any = None,
    ) -> Any:
        """
        Get elements from global config
    
        If ``override_with`` is not None this value will be passed straight back.
        Useful for getting kwarg defaults from Dask config.
    
        Use '.' for nested access
    
        Examples
        --------
        >>> from dask import config
        >>> config.get('foo')  # doctest: +SKIP
        {'x': 1, 'y': 2}
    
        >>> config.get('foo.x')  # doctest: +SKIP
        1
    
        >>> config.get('foo.x.y', default=123)  # doctest: +SKIP
        123
    
        >>> config.get('foo.y', override_with=None)  # doctest: +SKIP
        2
    
        >>> config.get('foo.y', override_with=3)  # doctest: +SKIP
        3
    
        See Also
        --------
        dask.config.set
        """
        if override_with is not None:
            return override_with
        keys = key.split(".")
        result = config
        for k in keys:
            k = canonical_name(k, result)
            try:
>               result = result[k]
E               KeyError: 'disk'

C:\Users\runneradmin\miniconda3\envs\dask-distributed\Lib\site-packages\dask\config.py:568: KeyError

concurrent.futures._base.CancelledError: inc-64456584d8a2e7176e2cd177efaa15f2
s = <Scheduler 'tcp://127.0.0.1:52715', workers: 0, cores: 0, tasks: 0>
a = <Worker 'tcp://127.0.0.1:52716', name: 0, status: closed, stored: 0, running: 0/1, ready: 0, comm: 0, waiting: 0>
b = <Worker 'tcp://127.0.0.1:52719', name: 1, status: closed, stored: 0, running: 0/2, ready: 0, comm: 0, waiting: 0>

    @gen_cluster()
    async def test_mixing_clients_same_scheduler(s, a, b):
        async with Client(s.address, asynchronous=True) as c1, Client(
            s.address, asynchronous=True
        ) as c2:
            future = c1.submit(inc, 1)
>           assert await c2.submit(inc, future) == 3

distributed\tests\test_client.py:6232: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

self = <Future: cancelled, key: inc-64456584d8a2e7176e2cd177efaa15f2>
raiseit = True

    async def _result(self, raiseit=True):
        await self._state.wait()
        if self.status == "error":
            exc = clean_exception(self._state.exception, self._state.traceback)
            if raiseit:
                typ, exc, tb = exc
                raise exc.with_traceback(tb)
            else:
                return exc
        elif self.status == "cancelled":
            exception = CancelledError(self.key)
            if raiseit:
>               raise exception
E               concurrent.futures._base.CancelledError: inc-64456584d8a2e7176e2cd177efaa15f2

distributed\client.py:336: CancelledError