code: def hash_join_p2p(
      lhs: _Frame,
      left_on: IndexLabel | None,
      rhs: _Frame,
      right_on: IndexLabel | None,
      how: MergeHow = "inner",
      npartitions: int | None = None,
      suffixes: Suffixes = ("_x", "_y"),
      indicator: bool = False,
  ):
      from dask.dataframe.core import Index, new_dd_object
  
      if npartitions is None:
          npartitions = max(lhs.npartitions, rhs.npartitions)
  
      if isinstance(left_on, Index):
          _left_on = None
          left_index = True
      else:
          left_index = False
          _left_on = left_on
  
      if isinstance(right_on, Index):
          _right_on = None
          right_index = True
      else:
          right_index = False
          _right_on = right_on
      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
      # Avoid using dummy data for a collection it is empty
      _lhs_meta = lhs._meta_nonempty if len(lhs.columns) else lhs._meta
      _rhs_meta = rhs._meta_nonempty if len(rhs.columns) else rhs._meta
      meta = _lhs_meta.merge(_rhs_meta, **merge_kwargs)
      lhs = _calculate_partitions(lhs, left_on, npartitions)
      rhs = _calculate_partitions(rhs, right_on, npartitions)
      merge_name = "hash-join-" + tokenize(lhs, rhs, **merge_kwargs)
      disk: bool = dask.config.get("distributed.p2p.disk")
      join_layer = HashJoinP2PLayer(
          name=merge_name,
          name_input_left=lhs._name,
          meta_input_left=lhs._meta,
          left_on=_left_on,
          n_partitions_left=lhs.npartitions,
          name_input_right=rhs._name,
          meta_input_right=rhs._meta,
          right_on=_right_on,
          n_partitions_right=rhs.npartitions,
          meta_output=meta,
          how=how,
          npartitions=npartitions,
          suffixes=suffixes,
          indicator=indicator,
          left_index=left_index,
          right_index=right_index,
          disk=disk,
      )
      graph = HighLevelGraph.from_collections(
          merge_name, join_layer, dependencies=[lhs, rhs]
      )
      return new_dd_object(
          graph, merge_name, meta, [None] * (npartitions + 1)
      )
  
imports: from __future__ import annotations
  from collections.abc import Iterable, Sequence
  from typing import TYPE_CHECKING, Any
  import dask
  from dask.base import is_dask_collection, tokenize
  from dask.highlevelgraph import HighLevelGraph
  from dask.layers import Layer
  from distributed.shuffle._arrow import check_minimal_arrow_version
  from distributed.shuffle._core import ShuffleId, barrier_key, get_worker_plugin
  from distributed.shuffle._shuffle import shuffle_barrier, shuffle_transfer
label_desc: Use image node_id 3 for calling a global function with example usage: hash_join_p2p(lhs, left_on, rhs, right_on, how, npartitions, suffixes, indicator) and returns: new_dd_object