Index join with MultiIndex columns fails on the shuffle path: "columns overlap but no suffix specified: MultiIndex([('_partitions', '')])"

Describe the issue

Joining two dask DataFrames on the index raises

ValueError: columns overlap but no suffix specified: MultiIndex([('_partitions', '')])

when both frames have MultiIndex columns and the right side has two or more partitions with unknown divisions. That combination lowers the join to the hash-shuffle merge path (RearrangeByColumn). The shuffle assigns a helper "_partitions" column, but with MultiIndex columns pandas stores that label as the tuple ("_partitions", ""). The drop-back step compares labels against the plain string, so the helper column is never removed, reaches the final per-partition pandas.merge on both sides, and collides.

The same join succeeds with single-level columns, and with MultiIndex columns when the right side has one partition (broadcast path), so the failure is specific to the shuffle path plus MultiIndex columns.

The real-world trigger was groupby(...).agg(["median"], split_out=2), which produces MultiIndex columns, multiple partitions, and unknown divisions at once, followed by a join onto another aggregated frame.

Minimal Complete Verifiable Example

import numpy as np
import pandas as pd
import dask.dataframe as dd

mi_l = pd.MultiIndex.from_tuples([("a", "median"), ("b", "median")])
mi_r = pd.MultiIndex.from_tuples([("c", "median"), ("d", "median")])
idx = pd.RangeIndex(6, name="id")

left = dd.from_pandas(
    pd.DataFrame(np.arange(12).reshape(6, 2), index=idx, columns=mi_l),
    npartitions=1,
)
# Two partitions with unknown divisions force the shuffle-merge path.
right = dd.from_pandas(
    pd.DataFrame(np.arange(12).reshape(6, 2), index=idx, columns=mi_r),
    npartitions=2,
).clear_divisions()

left.join(right).compute()  # ValueError

Controls that pass with the same shapes:

# single-level columns, 2 partitions, unknown divisions
dd.from_pandas(pd.DataFrame(np.arange(12).reshape(6, 2), index=idx, columns=["a", "b"]), npartitions=1).join(
    dd.from_pandas(pd.DataFrame(np.arange(12).reshape(6, 2), index=idx, columns=["c", "d"]), npartitions=2).clear_divisions()
).compute()

# MultiIndex columns, right side has 1 partition
left.join(right.repartition(npartitions=1)).compute()

Anything else we need to know?

The string "_partitions" is compared against column labels in two places in dask/dataframe/dask_expr/_shuffle.py, and both miss the tuple label that MultiIndex columns produce:

1. RearrangeByColumn._lower, the drop-back after the shuffle:
   python复制

   return shuffled[
       [c for c in shuffled.columns if c not in ["_partitions", "_partitions_0"]]
   ]

2. ShuffleBase._simplify_up, the projection push-down, which keeps a column only if col in partitioning_index or col in projection. Fixing only the drop-back is not enough: the push-down then drops the helper before the shuffle runs and _shuffle_group fails with KeyError: '_partitions'.

A fix that treats a tuple label as a match when its first level is one of the helper names, applied in both places, makes the MCVE pass along with three-level columns, both sides multi-partition, a real groupby(...).agg(["median"], split_out=2) source, and a column-key merge between MultiIndex frames, and leaves test_shuffle.py and test_merge.py green. I am happy to open a PR with that change and a MultiIndex regression test.

Unrelated but noticed while testing: a real user column literally named _partitions is silently dropped by any shuffle, before and after such a fix.

The code on main today is unchanged from 2026.8.0 in this file, so the bug is present there as well. Searched for the error message together with _partitions and MultiIndex and found no existing report.

Environment

• Dask version: 2026.8.0 (also reproduces against current main)
• pandas version: 3.0.5
• numpy version: 2.5.3
• Python version: 3.12.14
• Operating System: Linux
• Install method: conda-forge
• Scheduler: default threaded scheduler, task-based shuffle