class_docstr: This is an implementation of a Distributed Lookup Service to provide the following
  services to its users. Map 1) global node-ids to partition-ids, and 2) global node-ids
  to shuffle global node-ids (contiguous, within each node for a give node_type and across
  all the partitions)
  
  This services initializes itself with the node-id to partition-id mappings, which are inputs
  to this service. The node-id to partition-id  mappings are assumed to be in one file for each
  node type. These node-id-to-partition-id mappings are split within the service processes so that
  each process ends up with a contiguous chunk. It first divides the no of mappings (node-id to
  partition-id) for each node type into equal chunks across all the service processes. So each
  service process will be thse owner of a set of node-id-to-partition-id mappings. This class
  has two functions which are as follows:
  
  1) `get_partition_ids` function which returns the node-id to partition-id mappings to the user
  2) `get_shuffle_nids` function which returns the node-id to shuffle-node-id mapping to the user
  
  Parameters:
  -----------
  input_dir : string
      string representing the input directory where the node-type partition-id
      files are located
  ntype_names : list of strings
      list of strings which are used to read files located within the input_dir
      directory and these files contents are partition-id's for the node-ids which
      are of a particular node type
  id_map : dgl.distributed.id_map instance
      this id_map is used to retrieve ntype-ids, node type ids, and type_nids, per type
      node ids, for any given global node id
  rank : integer
      integer indicating the rank of a given process
  world_size : integer
      integer indicating the total no. of processes
  num_parts : integer
      interger representing the no. of partitions
code: def __init__(
      self, input_dir, ntype_names, rank, world_size, num_parts
  ):
      assert os.path.isdir(input_dir)
      assert ntype_names is not None
      assert len(ntype_names) > 0
  
      # These lists are indexed by ntype_ids.
      type_nid_begin = []
      type_nid_end = []
      partid_list = []
      ntype_count = []
      ntypes = []
  
      # Iterate over the node types and extract the partition id mappings.
      for ntype in ntype_names:
          filename = f"{ntype}.txt"
          logging.debug(
              f"[Rank: {rank}] Reading file: {os.path.join(input_dir, filename)}"
          )
  
          read_options = pyarrow.csv.ReadOptions(
              use_threads=True,
              block_size=4096,
              autogenerate_column_names=True,
          )
          parse_options = pyarrow.csv.ParseOptions(delimiter=" ")
          ntype_partids = []
          with pyarrow.csv.open_csv(
              os.path.join(input_dir, "{}.txt".format(ntype)),
              read_options=read_options,
              parse_options=parse_options,
          ) as reader:
              for next_chunk in reader:
                  if next_chunk is None:
                      break
                  next_table = pyarrow.Table.from_batches([next_chunk])
                  ntype_partids.append(next_table["f0"].to_numpy())
  
          ntype_partids = np.concatenate(ntype_partids)
          count = len(ntype_partids)
          ntype_count.append(count)
          ntypes.append(ntype)
  
          # Each rank assumes a contiguous set of partition-ids which are equally split
          # across all the processes.
          split_size = np.ceil(count / np.int64(world_size)).astype(
              np.int64
          )
          start, end = (
              np.int64(rank) * split_size,
              np.int64(rank + 1) * split_size,
          )
          if rank == (world_size - 1):
              end = count
          type_nid_begin.append(start)
          type_nid_end.append(end)
  
          # Slice the partition-ids which belong to the current instance.
          partid_list.append(copy.deepcopy(ntype_partids[start:end]))
  
          # Explicitly release the array read from the file.
          del ntype_partids
  
      logging.debug(
          f"[Rank: {rank}] ntypeid begin - {type_nid_begin} - {type_nid_end}"
      )
  
      # Store all the information in the object instance variable.
      self.type_nid_begin = np.array(type_nid_begin, dtype=np.int64)
      self.type_nid_end = np.array(type_nid_end, dtype=np.int64)
      self.partid_list = partid_list
      self.ntype_count = np.array(ntype_count, dtype=np.int64)
      self.ntypes = ntypes
      self.rank = rank
      self.world_size = world_size
      self.num_parts = num_parts
  
file_path: dgl/tools/distpartitioning/dist_lookup.py
imports: import copy
  import logging
  import os
  import numpy
  import pyarrow
  import torch
  from gloo_wrapper import allgather_sizes, alltoallv_cpu
  from pyarrow import csv
  from utils import map_partid_rank
label_desc: Use image node_id 1 to create a new DistLookupService object  with example: obj = DistLookupService(input_dir, ntype_names, rank, world_size, num_parts)