code: def test_dataframe_broadcast_merge(c, on, broadcast):
      # See: https://github.com/dask/dask/issues/9870
      pd = pytest.importorskip("pandas")
      dd = pytest.importorskip("dask.dataframe")
  
      pdfl = pd.DataFrame({"a": [1, 2] * 2, "b_left": range(4)})
      pdfr = pd.DataFrame({"a": [2, 1], "b_right": range(2)})
      dfl = dd.from_pandas(pdfl, npartitions=4)
      dfr = dd.from_pandas(pdfr, npartitions=2)
  
      ddfm = dd.merge(
          dfl, dfr, on=on, broadcast=broadcast, shuffle_method="tasks"
      )
      dfm = ddfm.compute()
      dd.utils.assert_eq(
          dfm.sort_values("a"),
          pd.merge(pdfl, pdfr, on=on).sort_values("a"),
          check_index=False,
      )
  
imports: from __future__ import annotations
  import pytest
  import asyncio
  import os
  import sys
  from functools import partial
  from operator import add
  from distributed import Client, SchedulerPlugin, WorkerPlugin
  from distributed.utils_test import cleanup
  from distributed.utils_test import client
  from distributed.utils_test import cluster, cluster_fixture, gen_cluster, loop, loop_in_thread, popen, varying
  import dask
  import dask.bag
  from dask import compute, delayed, persist
  from dask.base import compute_as_if_collection, get_scheduler
  from dask.blockwise import Blockwise
  from dask.delayed import Delayed
  from dask.distributed import futures_of, wait
  from dask.layers import ShuffleLayer, SimpleShuffleLayer
  from dask.utils import get_named_args, get_scheduler_lock, tmpdir, tmpfile
  from dask.utils_test import inc
label_desc: Use image node_id 7 for calling a global function with example usage: test_dataframe_broadcast_merge(c, on, broadcast) without return types