code: def merge(
      left,
      right,
      how="inner",
      on=None,
      left_on=None,
      right_on=None,
      left_index=False,
      right_index=False,
      suffixes=("_x", "_y"),
      indicator=False,
      npartitions=None,
      shuffle_method=None,
      max_branch=None,
      broadcast=None,
  ):
      for o in [on, left_on, right_on]:
          if isinstance(o, _Frame):
              raise NotImplementedError(
                  "Dask collections not currently allowed in merge columns"
              )
      if (
          not on
          and not left_on
          and not right_on
          and not left_index
          and not right_index
      ):
          on = [c for c in left.columns if c in right.columns]
          if not on:
              left_index = right_index = True
  
      if on and not left_on and not right_on:
          left_on = right_on = on
          on = None
  
      supported_how = (
          "left",
          "right",
          "outer",
          "inner",
          "leftanti",
          "leftsemi",
      )
      if how not in supported_how:
          raise ValueError(
              f"dask.dataframe.merge does not support how='{how}'. Options are: {supported_how}."
              f" Note that 'leftanti' and 'leftsemi' are only dask_cudf options."
          )
  
      if isinstance(left, (pd.Series, pd.DataFrame)) and isinstance(
          right, (pd.Series, pd.DataFrame)
      ):
          return pd.merge(
              left,
              right,
              how=how,
              on=on,
              left_on=left_on,
              right_on=right_on,
              left_index=left_index,
              right_index=right_index,
              suffixes=suffixes,
              indicator=indicator,
          )
  
      # Transform pandas objects into dask.dataframe objects
      if not is_dask_collection(left):
          if right_index and left_on:  # change to join on index
              left = left.set_index(left[left_on])
              left_on = None
              left_index = True
          left = from_pandas(left, npartitions=1)  # turn into DataFrame
  
      if not is_dask_collection(right):
          if left_index and right_on:  # change to join on index
              right = right.set_index(right[right_on])
              right_on = None
              right_index = True
          right = from_pandas(
              right, npartitions=1
          )  # turn into DataFrame
  
      # Both sides are now dd.DataFrame or dd.Series objects
      merge_indexed_left = (
          left_index or left._contains_index_name(left_on)
      ) and left.known_divisions
  
      merge_indexed_right = (
          right_index or right._contains_index_name(right_on)
      ) and right.known_divisions
  
      # Both sides indexed
      if merge_indexed_left and merge_indexed_right:  # Do indexed join
          return merge_indexed_dataframes(
              left,
              right,
              how=how,
              suffixes=suffixes,
              indicator=indicator,
              left_on=left_on,
              right_on=right_on,
              left_index=left_index,
              right_index=right_index,
          )
  
      # Single partition on one side
      # Note that cudf supports "leftsemi" and "leftanti" joins
      elif (
          left.npartitions == 1
          and how in allowed_right
          or right.npartitions == 1
          and how in allowed_left
      ):
          return single_partition_join(
              left,
              right,
              how=how,
              right_on=right_on,
              left_on=left_on,
              left_index=left_index,
              right_index=right_index,
              suffixes=suffixes,
              indicator=indicator,
          )
  
      # One side is indexed, the other not
      elif (
          left_index
          and left.known_divisions
          and not right_index
          or right_index
          and right.known_divisions
          and not left_index
      ):
          left_empty = left._meta_nonempty
          right_empty = right._meta_nonempty
          meta = left_empty.merge(
              right_empty,
              how=how,
              on=on,
              left_on=left_on,
              right_on=right_on,
              left_index=left_index,
              right_index=right_index,
              suffixes=suffixes,
              indicator=indicator,
          )
  
          if merge_indexed_left and left.known_divisions:
              right = rearrange_by_divisions(
                  right,
                  right_on,
                  left.divisions,
                  max_branch,
                  shuffle_method=shuffle_method,
              )
              left = left.clear_divisions()
          elif merge_indexed_right and right.known_divisions:
              left = rearrange_by_divisions(
                  left,
                  left_on,
                  right.divisions,
                  max_branch,
                  shuffle_method=shuffle_method,
              )
              right = right.clear_divisions()
          return map_partitions(
              merge_chunk,
              left,
              right,
              meta=meta,
              how=how,
              on=on,
              left_on=left_on,
              right_on=right_on,
              left_index=left_index,
              right_index=right_index,
              suffixes=suffixes,
              indicator=indicator,
              result_meta=meta,
          )
      # Catch all hash join
      else:
          if left_on and right_on:
              warn_dtype_mismatch(left, right, left_on, right_on)
  
          # Check if we should use a broadcast_join
          # See note on `broadcast_bias` below.
          broadcast_bias = 0.5
          if isinstance(broadcast, float):
              broadcast_bias = float(broadcast)
              broadcast = None
          elif (
              not isinstance(broadcast, bool) and broadcast is not None
          ):
              # Let's be strict about the `broadcast` type to
              # avoid arbitrarily casting int to float or bool.
              raise ValueError(
                  f"Optional `broadcast` argument must be float or bool."
                  f"Type={type(broadcast)} is not supported."
              )
          bcast_side = (
              "left"
              if left.npartitions < right.npartitions
              else "right"
          )
          n_small = min(left.npartitions, right.npartitions)
          n_big = max(left.npartitions, right.npartitions)
          if (
              shuffle_method in ("tasks", "p2p", None)
              and how in ("inner", "left", "right")
              and how != bcast_side
              and broadcast is not False
          ):
              # Note on `broadcast_bias`:
              # We can expect the broadcast merge to be competitive with
              # the shuffle merge when the number of partitions in the
              # smaller collection is less than the logarithm of the number
              # of partitions in the larger collection.  By default, we add
              # a small preference for the shuffle-based merge by multiplying
              # the log result by a 0.5 scaling factor.  We call this factor
              # the `broadcast_bias`, because a larger number will make Dask
              # more likely to select the `broadcast_join` code path.  If
              # the user specifies a floating-point value for the `broadcast`
              # kwarg, that value will be used as the `broadcast_bias`.
  
              # FIXME: We never evaluated how P2P compares against broadcast and
              # where a suitable cutoff point is. While scaling likely still
              # depends on number of partitions, the broadcast bias should likely
              # be different
              if broadcast or (
                  n_small < math.log2(n_big) * broadcast_bias
              ):
                  return broadcast_join(
                      left,
                      left.index if left_index else left_on,
                      right,
                      right.index if right_index else right_on,
                      how,
                      npartitions,
                      suffixes,
                      indicator=indicator,
                  )
  
          return hash_join(
              left,
              left.index if left_index else left_on,
              right,
              right.index if right_index else right_on,
              how,
              npartitions,
              suffixes,
              shuffle_method=shuffle_method,
              indicator=indicator,
              max_branch=max_branch,
          )
  
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 9 for calling a global function with example usage: merge(left, right, how, on, left_on, right_on, left_index, right_index, suffixes, indicator, npartitions, shuffle_method, max_branch, broadcast) and returns: pd, merge_indexed_dataframes, single_partition_join, map_partitions, hash_join, broadcast_join