code: def _mg_call_plc_replicate(
      client: dask.distributed.client.Client,
      sID: bytes,
      dask_object: dict,
      input_type: str,
      col_names: list,
  ) -> Union[dask_cudf.DataFrame, dask_cudf.Series]:
      if input_type == "dataframe":
          result = [
              client.submit(
                  _call_plc_replicate_dataframe,
                  sID,
                  edata,
                  workers=[w],
                  allow_other_workers=False,
                  pure=False,
              )
              for w, edata in dask_object.items()
          ]
      elif input_type == "dataframe":
          result = [
              client.submit(
                  _call_plc_replicate_series,
                  sID,
                  edata,
                  workers=[w],
                  allow_other_workers=False,
                  pure=False,
              )
              for w, edata in dask_object.items()
          ]
      elif input_type == "edgelist":
          result = [
              client.submit(
                  _call_plc_replicate_edgelist,
                  sID,
                  edata,
                  col_names,
                  workers=[w],
                  allow_other_workers=False,
                  pure=False,
              )
              for w, edata in dask_object.items()
          ]
  
      ddf = dask_cudf.from_delayed(result, verify_meta=False).persist()
      wait(ddf)
      wait([r.release() for r in result])
      return ddf
  
file_path: cugraph/python/cugraph/cugraph/structure/replicate_edgelist.py
imports: import dask_cudf
  import cudf
  from dask.distributed import wait, default_client
  import numpy
  from pylibcugraph import ResourceHandle, replicate_edgelist
  from cugraph.dask.common.part_utils import get_persisted_df_worker_map, persist_dask_df_equal_parts_per_worker
  import dask
  import cupy
  import cugraph.dask.comms.comms
  from typing import Union, Tuple
label_desc: Use image node_id 5 for calling a global function with example usage: _mg_call_plc_replicate(client, sID, dask_object, input_type, col_names) and returns: ddf