code: def _mg_call_plc_uniform_neighbor_sample(
      client,
      session_id,
      input_graph,
      ddf,
      keep_batches_together,
      min_batch_id,
      max_batch_id,
      fanout_vals,
      with_replacement,
      weight_t,
      indices_t,
      with_edge_properties,
      random_state,
      return_offsets=False,
      return_hops=True,
      prior_sources_behavior=None,
      deduplicate_sources=False,
      renumber=False,
      use_legacy_names=True,
      include_hop_column=True,
      compress_per_hop=False,
      compression="COO",
  ):
      n_workers = None
      if keep_batches_together:
          n_workers = get_n_workers()
  
          if hasattr(min_batch_id, "compute"):
              min_batch_id = min_batch_id.compute()
          if hasattr(max_batch_id, "compute"):
              max_batch_id = max_batch_id.compute()
  
      result = [
          client.submit(
              _call_plc_uniform_neighbor_sample,
              session_id,
              input_graph._plc_graph[w],
              starts,
              keep_batches_together,
              n_workers,
              min_batch_id,
              max_batch_id,
              fanout_vals,
              with_replacement,
              weight_t=weight_t,
              with_edge_properties=with_edge_properties,
              # FIXME accept and properly transmute a numpy/cupy random state.
              random_state=hash((random_state, w)),
              return_offsets=return_offsets,
              return_hops=return_hops,
              prior_sources_behavior=prior_sources_behavior,
              deduplicate_sources=deduplicate_sources,
              renumber=renumber,
              use_legacy_names=use_legacy_names,  # remove in 23.12
              include_hop_column=include_hop_column,  # remove in 23.12
              compress_per_hop=compress_per_hop,
              compression=compression,
              allow_other_workers=False,
              pure=False,
          )
          for w, starts in ddf.items()
      ]
      del ddf
  
      empty_df = (
          create_empty_df_with_edge_props(
              indices_t,
              weight_t,
              return_offsets=return_offsets,
              renumber=renumber,
              use_legacy_names=use_legacy_names,
              compression=compression,
              include_hop_column=include_hop_column,
          )
          if with_edge_properties
          else create_empty_df(indices_t, weight_t)
      )
      if not isinstance(empty_df, (list, tuple)):
          empty_df = [empty_df]
  
      wait(result)
  
      nout = 1
      if return_offsets:
          nout += 1
      if renumber:
          nout += 1
  
      result_split = [
          delayed(lambda x: x, nout=nout)(r) for r in result
      ]
  
      ddf = dask_cudf.from_delayed(
          [r[0] for r in result_split],
          meta=empty_df[0],
          verify_meta=False,
      ).persist()
      return_dfs = [ddf]
  
      if return_offsets:
          ddf_offsets = dask_cudf.from_delayed(
              [r[1] for r in result_split],
              meta=empty_df[1],
              verify_meta=False,
          ).persist()
          return_dfs.append(ddf_offsets)
  
      if renumber:
          ddf_renumber = dask_cudf.from_delayed(
              [r[-1] for r in result_split],
              meta=empty_df[-1],
              verify_meta=False,
          ).persist()
          return_dfs.append(ddf_renumber)
  
      wait(return_dfs)
      wait([r.release() for r in result_split])
      wait([r.release() for r in result])
      del result
  
      if len(return_dfs) == 1:
          return return_dfs[0]
      else:
          return tuple(return_dfs)
  
imports: from __future__ import annotations
  import warnings
  import numpy
  from dask import delayed
  from dask.distributed import Lock, get_client, wait
  import dask_cudf
  import cudf
  import cupy
  from pylibcugraph import ResourceHandle
  from pylibcugraph import uniform_neighbor_sample
  from cugraph.dask.comms import comms
  from cugraph.dask import get_n_workers
  from typing import Sequence, List, Union, Tuple
  from typing import TYPE_CHECKING
  from cugraph.dask.common.part_utils import get_persisted_df_worker_map, persist_dask_df_equal_parts_per_worker
label_desc: Use image node_id 5 for calling a global function with example usage: _mg_call_plc_uniform_neighbor_sample(client, session_id, input_graph, ddf, keep_batches_together, min_batch_id, max_batch_id, fanout_vals, with_replacement, weight_t, indices_t, with_edge_properties, random_state, return_offsets, return_hops, prior_sources_behavior, deduplicate_sources, renumber, use_legacy_names, include_hop_column, compress_per_hop, compression) and returns: return_dfs, tuple