code: def _make_batch_ids(
      bdf: cudf.DataFrame,
      batch_size: int,
      num_workers: int,
      partition_info: Optional[Union[dict, str]] = None,
  ):
      # Required by dask; need to skip dummy partitions.
      if partition_info is None:
          return cudf.DataFrame(
              {
                  "batch": cudf.Series(dtype="int32"),
                  "start": cudf.Series(dtype="int64"),
              }
          )
  
      partition = partition_info["number"]
      if partition is None:
          raise ValueError("division is absent")
  
      num_batches = int(ceil(len(bdf) / batch_size))
  
      batch_ids = cupy.repeat(
          cupy.arange(
              num_batches * partition,
              num_batches * (partition + 1),
              dtype="int32",
          ),
          batch_size,
      )[: len(bdf)]
  
      bdf = bdf.reset_index(drop=True)
      bdf["batch"] = cudf.Series(batch_ids)
  
      return bdf
  
imports: import logging
  import warnings
  import argparse
  import traceback
  from cugraph.testing.mg_utils import generate_edgelist_rmat, get_allocation_counts_dask_lazy, sizeof_fmt, get_peak_output_ratio_across_workers, start_dask_client, stop_dask_client, enable_spilling
  from cugraph.structure.symmetrize import symmetrize
  from cugraph.gnn import BulkSampler
  import cugraph
  import json
  import re
  import os
  import gc
  from time import sleep, perf_counter
  from math import ceil
  import pandas
  import numpy
  import cupy
  import cudf
  import dask_cudf
  import dask.dataframe
  from dask.distributed import default_client
  from cugraph.dask import get_n_workers
  from typing import Optional, Union, Dict
label_desc: Use image node_id 4 for calling a global function with example usage: _make_batch_ids(bdf, batch_size, num_workers, partition_info) and returns: bdf, cudf