We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
P2P with null partitions fails
I came across a use case where P2P shuffling fails when a column in a partition has all null values. Here's a minimal reproducer:

```python
import numpy as np
import pandas as pd
import dask.dataframe as dd
from dask.distributed import Client

def make_partition(i):
    """Return null column for one partition"""
    if i == 1:
        return pd.DataFrame({"a": np.random.random(10), "b": None})
    return pd.DataFrame({"a": np.random.random(10), "b": np.random.random(10)})


if __name__ == "__main__":
    with Client() as client:
        ddf = dd.from_map(make_partition, range(10))
        result = ddf.set_index("a", shuffle="p2p").compute()
        print(result)
```

which raises the following error:

```
Traceback (most recent call last):
  File "/Users/james/mambaforge/envs/dask-py39/lib/python3.9/site-packages/distributed/shuffle/_shuffle.py", line 96, in shuffle_barrier
    return _get_worker_plugin().barrier(id, run_ids)
  File "/Users/james/mambaforge/envs/dask-py39/lib/python3.9/site-packages/distributed/shuffle/_worker_plugin.py", line 925, in barrier
    result = sync(self.worker.loop, self._barrier, shuffle_id, run_ids)
  File "/Users/james/mambaforge/envs/dask-py39/lib/python3.9/site-packages/distributed/utils.py", line 426, in sync
    raise exc.with_traceback(tb)
  File "/Users/james/mambaforge/envs/dask-py39/lib/python3.9/site-packages/distributed/utils.py", line 399, in f
    result = yield future
  File "/Users/james/mambaforge/envs/dask-py39/lib/python3.9/site-packages/tornado/gen.py", line 767, in run
    value = future.result()
  File "/Users/james/mambaforge/envs/dask-py39/lib/python3.9/site-packages/distributed/shuffle/_worker_plugin.py", line 689, in _barrier
    await shuffle.barrier()
  File "/Users/james/mambaforge/envs/dask-py39/lib/python3.9/site-packages/distributed/shuffle/_worker_plugin.py", line 116, in barrier
    await self.scheduler.shuffle_barrier(id=self.id, run_id=self.run_id)
  File "/Users/james/mambaforge/envs/dask-py39/lib/python3.9/site-packages/distributed/core.py", line 1374, in send_recv_from_rpc
    return await send_recv(comm=comm, op=key, **kwargs)
  File "/Users/james/mambaforge/envs/dask-py39/lib/python3.9/site-packages/distributed/core.py", line 1158, in send_recv
    raise exc.with_traceback(tb)
  File "/Users/james/mambaforge/envs/dask-py39/lib/python3.9/site-packages/distributed/core.py", line 930, in _handle_comm
    result = await result
  File "/Users/james/mambaforge/envs/dask-py39/lib/python3.9/site-packages/distributed/shuffle/_scheduler_plugin.py", line 139, in barrier
    await self.scheduler.broadcast(
  File "/Users/james/mambaforge/envs/dask-py39/lib/python3.9/site-packages/distributed/scheduler.py", line 6169, in broadcast
    results = await All(
  File "/Users/james/mambaforge/envs/dask-py39/lib/python3.9/site-packages/distributed/utils.py", line 252, in All
    result = await tasks.next()
  File "/Users/james/mambaforge/envs/dask-py39/lib/python3.9/site-packages/distributed/scheduler.py", line 6147, in send_message
    resp = await send_recv(
  File "/Users/james/mambaforge/envs/dask-py39/lib/python3.9/site-packages/distributed/core.py", line 1160, in send_recv
    raise Exception(response["exception_text"])
Exception: ArrowInvalid('Schema at index 1 was different: \na: double\nb: null\n_partitions: int64\n__index_level_0__: int64\nvs\na: double\nb: double\n_partitions: int64\n__index_level_0__: int64')

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/Users/james/projects/dask/dask/test-p2p-shuffle.py", line 16, in <module>
    result = ddf.set_index("a", shuffle="p2p").compute()
  File "/Users/james/mambaforge/envs/dask-py39/lib/python3.9/site-packages/distributed/shuffle/_shuffle.py", line 98, in shuffle_barrier
    raise RuntimeError(f"shuffle_barrier failed during shuffle {id}") from e
RuntimeError: shuffle_barrier failed during shuffle 2b30bc4838ba6b632ee7d432b2b31dc8
```

Interestingly this snippet _usually_ fails, but I happened to notice it sometimes runs successfully which is in itself is also interesting 

```
                 b
a
0.015788  0.677673
0.019857  0.481580
0.027898  0.564877
0.031679  0.442530
0.048167  0.990417
...            ...
0.957410  0.651139
0.969251       NaN
0.976877  0.369628
0.984942       NaN
0.999345  0.926310

[100 rows x 1 columns]
```

cc @hendrikmakait for visibility 

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `distributed/shuffle/_arrow.py`
Content:
```
1 from __future__ import annotations
2 
3 from io import BytesIO
4 from typing import TYPE_CHECKING
5 
6 from packaging.version import parse
7 
8 if TYPE_CHECKING:
9     import pandas as pd
10     import pyarrow as pa
11 
12 
13 def check_dtype_support(meta_input: pd.DataFrame) -> None:
14     import pandas as pd
15 
16     for name in meta_input:
17         column = meta_input[name]
18         # FIXME: PyArrow does not support complex numbers: https://issues.apache.org/jira/browse/ARROW-638
19         if pd.api.types.is_complex_dtype(column):
20             raise TypeError(
21                 f"p2p does not support data of type '{column.dtype}' found in column '{name}'."
22             )
23         # FIXME: PyArrow does not support sparse data: https://issues.apache.org/jira/browse/ARROW-8679
24         if isinstance(column.dtype, pd.SparseDtype):
25             raise TypeError("p2p does not support sparse data found in column '{name}'")
26 
27 
28 def check_minimal_arrow_version() -> None:
29     """Verify that the the correct version of pyarrow is installed to support
30     the P2P extension.
31 
32     Raises a RuntimeError in case pyarrow is not installed or installed version
33     is not recent enough.
34     """
35     # First version to introduce Table.sort_by
36     minversion = "7.0.0"
37     try:
38         import pyarrow as pa
39     except ImportError:
40         raise RuntimeError(f"P2P shuffling requires pyarrow>={minversion}")
41 
42     if parse(pa.__version__) < parse(minversion):
43         raise RuntimeError(
44             f"P2P shuffling requires pyarrow>={minversion} but only found {pa.__version__}"
45         )
46 
47 
48 def convert_partition(data: bytes, meta: pd.DataFrame) -> pd.DataFrame:
49     import pyarrow as pa
50 
51     from dask.dataframe.dispatch import from_pyarrow_table_dispatch
52 
53     file = BytesIO(data)
54     end = len(data)
55     shards = []
56     while file.tell() < end:
57         sr = pa.RecordBatchStreamReader(file)
58         shards.append(sr.read_all())
59     table = pa.concat_tables(shards, promote=True)
60 
61     df = from_pyarrow_table_dispatch(meta, table, self_destruct=True)
62     return df.astype(meta.dtypes, copy=False)
63 
64 
65 def list_of_buffers_to_table(data: list[bytes]) -> pa.Table:
66     """Convert a list of arrow buffers and a schema to an Arrow Table"""
67     import pyarrow as pa
68 
69     return pa.concat_tables(deserialize_table(buffer) for buffer in data)
70 
71 
72 def serialize_table(table: pa.Table) -> bytes:
73     import pyarrow as pa
74 
75     stream = pa.BufferOutputStream()
76     with pa.ipc.new_stream(stream, table.schema) as writer:
77         writer.write_table(table)
78     return stream.getvalue().to_pybytes()
79 
80 
81 def deserialize_table(buffer: bytes) -> pa.Table:
82     import pyarrow as pa
83 
84     with pa.ipc.open_stream(pa.py_buffer(buffer)) as reader:
85         return reader.read_all()
86
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/distributed/shuffle/_arrow.py b/distributed/shuffle/_arrow.py
--- a/distributed/shuffle/_arrow.py
+++ b/distributed/shuffle/_arrow.py
@@ -79,7 +79,9 @@
     """Convert a list of arrow buffers and a schema to an Arrow Table"""
     import pyarrow as pa
 
-    return pa.concat_tables(deserialize_table(buffer) for buffer in data)
+    return pa.concat_tables(
+        (deserialize_table(buffer) for buffer in data), promote=True
+    )
 
 
 def serialize_table(table: pa.Table) -> bytes:

{"golden_diff": "diff --git a/distributed/shuffle/_arrow.py b/distributed/shuffle/_arrow.py\n--- a/distributed/shuffle/_arrow.py\n+++ b/distributed/shuffle/_arrow.py\n@@ -79,7 +79,9 @@\n     \"\"\"Convert a list of arrow buffers and a schema to an Arrow Table\"\"\"\n     import pyarrow as pa\n \n-    return pa.concat_tables(deserialize_table(buffer) for buffer in data)\n+    return pa.concat_tables(\n+        (deserialize_table(buffer) for buffer in data), promote=True\n+    )\n \n \n def serialize_table(table: pa.Table) -> bytes:\n", "issue": "P2P with null partitions fails\nI came across a use case where P2P shuffling fails when a column in a partition has all null values. Here's a minimal reproducer:\r\n\r\n```python\r\nimport numpy as np\r\nimport pandas as pd\r\nimport dask.dataframe as dd\r\nfrom dask.distributed import Client\r\n\r\ndef make_partition(i):\r\n    \"\"\"Return null column for one partition\"\"\"\r\n    if i == 1:\r\n        return pd.DataFrame({\"a\": np.random.random(10), \"b\": None})\r\n    return pd.DataFrame({\"a\": np.random.random(10), \"b\": np.random.random(10)})\r\n\r\n\r\nif __name__ == \"__main__\":\r\n    with Client() as client:\r\n        ddf = dd.from_map(make_partition, range(10))\r\n        result = ddf.set_index(\"a\", shuffle=\"p2p\").compute()\r\n        print(result)\r\n```\r\n\r\nwhich raises the following error:\r\n\r\n```\r\nTraceback (most recent call last):\r\n  File \"/Users/james/mambaforge/envs/dask-py39/lib/python3.9/site-packages/distributed/shuffle/_shuffle.py\", line 96, in shuffle_barrier\r\n    return _get_worker_plugin().barrier(id, run_ids)\r\n  File \"/Users/james/mambaforge/envs/dask-py39/lib/python3.9/site-packages/distributed/shuffle/_worker_plugin.py\", line 925, in barrier\r\n    result = sync(self.worker.loop, self._barrier, shuffle_id, run_ids)\r\n  File \"/Users/james/mambaforge/envs/dask-py39/lib/python3.9/site-packages/distributed/utils.py\", line 426, in sync\r\n    raise exc.with_traceback(tb)\r\n  File \"/Users/james/mambaforge/envs/dask-py39/lib/python3.9/site-packages/distributed/utils.py\", line 399, in f\r\n    result = yield future\r\n  File \"/Users/james/mambaforge/envs/dask-py39/lib/python3.9/site-packages/tornado/gen.py\", line 767, in run\r\n    value = future.result()\r\n  File \"/Users/james/mambaforge/envs/dask-py39/lib/python3.9/site-packages/distributed/shuffle/_worker_plugin.py\", line 689, in _barrier\r\n    await shuffle.barrier()\r\n  File \"/Users/james/mambaforge/envs/dask-py39/lib/python3.9/site-packages/distributed/shuffle/_worker_plugin.py\", line 116, in barrier\r\n    await self.scheduler.shuffle_barrier(id=self.id, run_id=self.run_id)\r\n  File \"/Users/james/mambaforge/envs/dask-py39/lib/python3.9/site-packages/distributed/core.py\", line 1374, in send_recv_from_rpc\r\n    return await send_recv(comm=comm, op=key, **kwargs)\r\n  File \"/Users/james/mambaforge/envs/dask-py39/lib/python3.9/site-packages/distributed/core.py\", line 1158, in send_recv\r\n    raise exc.with_traceback(tb)\r\n  File \"/Users/james/mambaforge/envs/dask-py39/lib/python3.9/site-packages/distributed/core.py\", line 930, in _handle_comm\r\n    result = await result\r\n  File \"/Users/james/mambaforge/envs/dask-py39/lib/python3.9/site-packages/distributed/shuffle/_scheduler_plugin.py\", line 139, in barrier\r\n    await self.scheduler.broadcast(\r\n  File \"/Users/james/mambaforge/envs/dask-py39/lib/python3.9/site-packages/distributed/scheduler.py\", line 6169, in broadcast\r\n    results = await All(\r\n  File \"/Users/james/mambaforge/envs/dask-py39/lib/python3.9/site-packages/distributed/utils.py\", line 252, in All\r\n    result = await tasks.next()\r\n  File \"/Users/james/mambaforge/envs/dask-py39/lib/python3.9/site-packages/distributed/scheduler.py\", line 6147, in send_message\r\n    resp = await send_recv(\r\n  File \"/Users/james/mambaforge/envs/dask-py39/lib/python3.9/site-packages/distributed/core.py\", line 1160, in send_recv\r\n    raise Exception(response[\"exception_text\"])\r\nException: ArrowInvalid('Schema at index 1 was different: \\na: double\\nb: null\\n_partitions: int64\\n__index_level_0__: int64\\nvs\\na: double\\nb: double\\n_partitions: int64\\n__index_level_0__: int64')\r\n\r\nThe above exception was the direct cause of the following exception:\r\n\r\nTraceback (most recent call last):\r\n  File \"/Users/james/projects/dask/dask/test-p2p-shuffle.py\", line 16, in <module>\r\n    result = ddf.set_index(\"a\", shuffle=\"p2p\").compute()\r\n  File \"/Users/james/mambaforge/envs/dask-py39/lib/python3.9/site-packages/distributed/shuffle/_shuffle.py\", line 98, in shuffle_barrier\r\n    raise RuntimeError(f\"shuffle_barrier failed during shuffle {id}\") from e\r\nRuntimeError: shuffle_barrier failed during shuffle 2b30bc4838ba6b632ee7d432b2b31dc8\r\n```\r\n\r\nInterestingly this snippet _usually_ fails, but I happened to notice it sometimes runs successfully which is in itself is also interesting \r\n\r\n```\r\n                 b\r\na\r\n0.015788  0.677673\r\n0.019857  0.481580\r\n0.027898  0.564877\r\n0.031679  0.442530\r\n0.048167  0.990417\r\n...            ...\r\n0.957410  0.651139\r\n0.969251       NaN\r\n0.976877  0.369628\r\n0.984942       NaN\r\n0.999345  0.926310\r\n\r\n[100 rows x 1 columns]\r\n```\r\n\r\ncc @hendrikmakait for visibility \n", "before_files": [{"content": "from __future__ import annotations\n\nfrom io import BytesIO\nfrom typing import TYPE_CHECKING\n\nfrom packaging.version import parse\n\nif TYPE_CHECKING:\n    import pandas as pd\n    import pyarrow as pa\n\n\ndef check_dtype_support(meta_input: pd.DataFrame) -> None:\n    import pandas as pd\n\n    for name in meta_input:\n        column = meta_input[name]\n        # FIXME: PyArrow does not support complex numbers: https://issues.apache.org/jira/browse/ARROW-638\n        if pd.api.types.is_complex_dtype(column):\n            raise TypeError(\n                f\"p2p does not support data of type '{column.dtype}' found in column '{name}'.\"\n            )\n        # FIXME: PyArrow does not support sparse data: https://issues.apache.org/jira/browse/ARROW-8679\n        if isinstance(column.dtype, pd.SparseDtype):\n            raise TypeError(\"p2p does not support sparse data found in column '{name}'\")\n\n\ndef check_minimal_arrow_version() -> None:\n    \"\"\"Verify that the the correct version of pyarrow is installed to support\n    the P2P extension.\n\n    Raises a RuntimeError in case pyarrow is not installed or installed version\n    is not recent enough.\n    \"\"\"\n    # First version to introduce Table.sort_by\n    minversion = \"7.0.0\"\n    try:\n        import pyarrow as pa\n    except ImportError:\n        raise RuntimeError(f\"P2P shuffling requires pyarrow>={minversion}\")\n\n    if parse(pa.__version__) < parse(minversion):\n        raise RuntimeError(\n            f\"P2P shuffling requires pyarrow>={minversion} but only found {pa.__version__}\"\n        )\n\n\ndef convert_partition(data: bytes, meta: pd.DataFrame) -> pd.DataFrame:\n    import pyarrow as pa\n\n    from dask.dataframe.dispatch import from_pyarrow_table_dispatch\n\n    file = BytesIO(data)\n    end = len(data)\n    shards = []\n    while file.tell() < end:\n        sr = pa.RecordBatchStreamReader(file)\n        shards.append(sr.read_all())\n    table = pa.concat_tables(shards, promote=True)\n\n    df = from_pyarrow_table_dispatch(meta, table, self_destruct=True)\n    return df.astype(meta.dtypes, copy=False)\n\n\ndef list_of_buffers_to_table(data: list[bytes]) -> pa.Table:\n    \"\"\"Convert a list of arrow buffers and a schema to an Arrow Table\"\"\"\n    import pyarrow as pa\n\n    return pa.concat_tables(deserialize_table(buffer) for buffer in data)\n\n\ndef serialize_table(table: pa.Table) -> bytes:\n    import pyarrow as pa\n\n    stream = pa.BufferOutputStream()\n    with pa.ipc.new_stream(stream, table.schema) as writer:\n        writer.write_table(table)\n    return stream.getvalue().to_pybytes()\n\n\ndef deserialize_table(buffer: bytes) -> pa.Table:\n    import pyarrow as pa\n\n    with pa.ipc.open_stream(pa.py_buffer(buffer)) as reader:\n        return reader.read_all()\n", "path": "distributed/shuffle/_arrow.py"}], "after_files": [{"content": "from __future__ import annotations\n\nfrom io import BytesIO\nfrom typing import TYPE_CHECKING\n\nfrom packaging.version import parse\n\nif TYPE_CHECKING:\n    import pandas as pd\n    import pyarrow as pa\n\n\ndef check_dtype_support(meta_input: pd.DataFrame) -> None:\n    import pandas as pd\n\n    for name in meta_input:\n        column = meta_input[name]\n        # FIXME: PyArrow does not support complex numbers: https://issues.apache.org/jira/browse/ARROW-638\n        if pd.api.types.is_complex_dtype(column):\n            raise TypeError(\n                f\"p2p does not support data of type '{column.dtype}' found in column '{name}'.\"\n            )\n        # FIXME: PyArrow does not support sparse data: https://issues.apache.org/jira/browse/ARROW-8679\n        if isinstance(column.dtype, pd.SparseDtype):\n            raise TypeError(\"p2p does not support sparse data found in column '{name}'\")\n\n\ndef check_minimal_arrow_version() -> None:\n    \"\"\"Verify that the the correct version of pyarrow is installed to support\n    the P2P extension.\n\n    Raises a RuntimeError in case pyarrow is not installed or installed version\n    is not recent enough.\n    \"\"\"\n    # First version to introduce Table.sort_by\n    minversion = \"7.0.0\"\n    try:\n        import pyarrow as pa\n    except ImportError:\n        raise RuntimeError(f\"P2P shuffling requires pyarrow>={minversion}\")\n\n    if parse(pa.__version__) < parse(minversion):\n        raise RuntimeError(\n            f\"P2P shuffling requires pyarrow>={minversion} but only found {pa.__version__}\"\n        )\n\n\ndef convert_partition(data: bytes, meta: pd.DataFrame) -> pd.DataFrame:\n    import pandas as pd\n    import pyarrow as pa\n\n    from dask.dataframe.dispatch import from_pyarrow_table_dispatch\n\n    file = BytesIO(data)\n    end = len(data)\n    shards = []\n    while file.tell() < end:\n        sr = pa.RecordBatchStreamReader(file)\n        shards.append(sr.read_all())\n    table = pa.concat_tables(shards, promote=True)\n\n    def default_types_mapper(pyarrow_dtype: pa.DataType) -> object:\n        # Avoid converting strings from `string[pyarrow]` to `string[python]`\n        # if we have *some* `string[pyarrow]`\n        if (\n            pyarrow_dtype in {pa.large_string(), pa.string()}\n            and pd.StringDtype(\"pyarrow\") in meta.dtypes.values\n        ):\n            return pd.StringDtype(\"pyarrow\")\n        return None\n\n    df = from_pyarrow_table_dispatch(\n        meta, table, self_destruct=True, types_mapper=default_types_mapper\n    )\n    return df.astype(meta.dtypes, copy=False)\n\n\ndef list_of_buffers_to_table(data: list[bytes]) -> pa.Table:\n    \"\"\"Convert a list of arrow buffers and a schema to an Arrow Table\"\"\"\n    import pyarrow as pa\n\n    return pa.concat_tables(\n        (deserialize_table(buffer) for buffer in data), promote=True\n    )\n\n\ndef serialize_table(table: pa.Table) -> bytes:\n    import pyarrow as pa\n\n    stream = pa.BufferOutputStream()\n    with pa.ipc.new_stream(stream, table.schema) as writer:\n        writer.write_table(table)\n    return stream.getvalue().to_pybytes()\n\n\ndef deserialize_table(buffer: bytes) -> pa.Table:\n    import pyarrow as pa\n\n    with pa.ipc.open_stream(pa.py_buffer(buffer)) as reader:\n        return reader.read_all()\n", "path": "distributed/shuffle/_arrow.py"}]}