code: def broadcast_join(
      lhs,
      left_on,
      rhs,
      right_on,
      how="inner",
      npartitions=None,
      suffixes=("_x", "_y"),
      shuffle_method=None,
      indicator=False,
      parts_out=None,
  ):
      """Join two DataFrames on particular columns by broadcasting
  
      This broadcasts the partitions of the smaller DataFrame to each
      partition of the larger DataFrame, joins each partition pair,
      and then concatenates the new data for each output partition.
      """
  
      if npartitions:
          # Repartition the larger collection before the merge
          if lhs.npartitions < rhs.npartitions:
              rhs = rhs.repartition(npartitions=npartitions)
          else:
              lhs = lhs.repartition(npartitions=npartitions)
  
      if how not in ("inner", "left", "right"):
          # Broadcast algorithm cannot handle an "outer" join
          raise ValueError(
              "Only 'inner', 'left' and 'right' broadcast joins are supported."
          )
  
      if how == "left" and lhs.npartitions < rhs.npartitions:
          # Must broadcast rhs for a "left" broadcast join
          raise ValueError(
              "'left' broadcast join requires rhs broadcast."
          )
  
      if how == "right" and rhs.npartitions <= lhs.npartitions:
          # Must broadcast lhs for a "right" broadcast join
          raise ValueError(
              "'right' broadcast join requires lhs broadcast."
          )
  
      # TODO: It *may* be beneficial to perform the hash
      # split for "inner" join as well (even if it is not
      # technically needed for correctness).  More testing
      # is needed here.
      if how != "inner":
          # Shuffle to-be-broadcasted side by hash. This
          # means that we will need to perform a local
          # shuffle and split on each partition of the
          # "other" collection (with the same hashing
          # approach) to ensure the correct rows are
          # joined by `merge_chunk`.  The local hash and
          # split of lhs is in `_split_partition`.
          if lhs.npartitions < rhs.npartitions:
              lhs2 = shuffle_func(
                  lhs,
                  left_on,
                  shuffle_method="tasks",
              )
              lhs_name = lhs2._name
              lhs_dep = lhs2
              rhs_name = rhs._name
              rhs_dep = rhs
          else:
              rhs2 = shuffle_func(
                  rhs,
                  right_on,
                  shuffle_method="tasks",
              )
              lhs_name = lhs._name
              lhs_dep = lhs
              rhs_name = rhs2._name
              rhs_dep = rhs2
      else:
          lhs_name = lhs._name
          lhs_dep = lhs
          rhs_name = rhs._name
          rhs_dep = rhs
  
      if isinstance(left_on, Index):
          left_on = None
          left_index = True
      else:
          left_index = False
  
      if isinstance(right_on, Index):
          right_on = None
          right_index = True
      else:
          right_index = False
  
      merge_kwargs = dict(
          how=how,
          left_on=left_on,
          right_on=right_on,
          left_index=left_index,
          right_index=right_index,
          suffixes=suffixes,
          indicator=indicator,
      )
  
      # dummy result
      meta = lhs._meta_nonempty.merge(
          rhs._meta_nonempty, **merge_kwargs
      )
      merge_kwargs["result_meta"] = meta
  
      # Assume the output partitions/divisions
      # should correspond to the collection that
      # is NOT broadcasted.
      if lhs.npartitions < rhs.npartitions:
          npartitions = rhs.npartitions
          divisions = rhs.divisions
          _index_names = set(rhs._meta_nonempty.index.names)
      else:
          npartitions = lhs.npartitions
          divisions = lhs.divisions
          _index_names = set(lhs._meta_nonempty.index.names)
  
      # Cannot preserve divisions if the index is lost
      if _index_names != set(meta.index.names):
          divisions = [None] * (npartitions + 1)
  
      token = tokenize(lhs, rhs, npartitions, **merge_kwargs)
      name = "bcast-join-" + token
      broadcast_join_layer = BroadcastJoinLayer(
          name,
          npartitions,
          lhs_name,
          lhs.npartitions,
          rhs_name,
          rhs.npartitions,
          parts_out=parts_out,
          **merge_kwargs,
      )
  
      graph = HighLevelGraph.from_collections(
          name,
          broadcast_join_layer,
          dependencies=[lhs_dep, rhs_dep],
      )
  
      return new_dd_object(graph, name, meta, divisions)
  
imports: from __future__ import annotations
  import math
  import pickle
  import warnings
  from functools import partial, wraps
  import numpy
  import pandas
  from pandas.api.types import is_dtype_equal
  from tlz import merge_sorted, unique
  from dask.base import is_dask_collection, tokenize
  from dask.dataframe import methods
  from dask.dataframe.core import DataFrame, Index, Series, _concat, _deprecated_kwarg, _Frame, _maybe_from_pandas, is_broadcastable, map_partitions, new_dd_object, prefix_reduction, suffix_reduction
  from dask.dataframe.dispatch import group_split_dispatch, hash_object_dispatch
  from dask.dataframe.io import from_pandas
  from dask.dataframe.shuffle import partitioning_index, rearrange_by_divisions, shuffle, shuffle_group
  from dask.dataframe.utils import asciitable, check_meta, is_dataframe_like, is_series_like, make_meta, strip_unknown_categories
  from dask.highlevelgraph import HighLevelGraph
  from dask.layers import BroadcastJoinLayer
  from dask.utils import M, apply, get_default_shuffle_method
label_desc: Use image node_id 31 for calling a global function with example usage: broadcast_join(lhs, left_on, rhs, right_on, how, npartitions, suffixes, shuffle_method, indicator, parts_out) and returns: new_dd_object