Broadcast join fails when passing a list of columns to merge on

[ayushdg]

Describe the issue:

The broadcast merge codepath for dataframes throws an error when passing a list of columns in the on argument with a keyError

Minimal Complete Verifiable Example:

import pandas as pd
import numpy as np
from dask import dataframe as dd
from distributed import Client, wait
c = Client()

df1 = pd.DataFrame({"a":np.arange(20),"b":np.arange(20),"c":[1,2]*10})
df2 = df1.copy(deep=True)
df1 = dd.from_pandas(df1,2)
df2 = dd.from_pandas(df2,5)

len(df1.merge(df2,on=["a"], how="inner", shuffle="tasks", broadcast=True))
# KeyError: "('a',)"
# Works with on="a"

Anything else we need to know?:
Same issue persists when joining on multiple columns

Environment:

• Dask version: 2022.12.0
• Python version: 3.9
• Operating System: ubuntu18.04
• Install method (conda, pip, source): pip

[fjetter]

I can confirm and can reproduce this. My best guess is that this is a serialization error or we're calling a stringify or smth too eagerly.

basically the list is cast to a tuple and then stringified somewhere, i.e. ["a"] becomes "('a',)". Passing a tuple directly or the literal works as expected.

If I had to guess, this looks suspicious but I don't fully understand what's going on there. Maybe @rjzamora ?

[rjzamora]

Thanks for raising @ayushdg ! I ran into this bug yesterday, but didn't get a chance to raise and issue and investigate yet. @fjetter is correct that this is likely another HLG-serialization edge case (probably having to do with msgpack not distinguishing lists/tuples). I will try to figure out a fix, but also look forward to something like dask/distributed#6028 avoiding these problems altogether :)