code: def preprocess_ondisk_dataset(
      dataset_dir: str,
      include_original_edge_id: bool = False,
      force_preprocess: bool = None,
  ) -> str:
      """Preprocess the on-disk dataset. Parse the input config file,
      load the data, and save the data in the format that GraphBolt supports.
  
      Parameters
      ----------
      dataset_dir : str
          The path to the dataset directory.
      include_original_edge_id : bool, optional
          Whether to include the original edge id in the FusedCSCSamplingGraph.
      force_preprocess: bool, optional
          Whether to force reload the ondisk dataset.
  
      Returns
      -------
      output_config_path : str
          The path to the output config file.
      """
      # Check if the dataset path is valid.
      if not os.path.exists(dataset_dir):
          raise RuntimeError(f"Invalid dataset path: {dataset_dir}")
  
      # Check if the dataset_dir is a directory.
      if not os.path.isdir(dataset_dir):
          raise RuntimeError(
              f"The dataset must be a directory. But got {dataset_dir}"
          )
  
      # 0. Check if the dataset is already preprocessed.
      processed_dir_prefix = "preprocessed"
      preprocess_metadata_path = os.path.join(
          processed_dir_prefix, "metadata.yaml"
      )
      if os.path.exists(
          os.path.join(dataset_dir, preprocess_metadata_path)
      ):
          if force_preprocess is None:
              with open(
                  os.path.join(dataset_dir, preprocess_metadata_path),
                  "r",
              ) as f:
                  preprocess_config = yaml.safe_load(f)
              if (
                  preprocess_config.get(
                      "include_original_edge_id", None
                  )
                  == include_original_edge_id
              ):
                  force_preprocess = check_dataset_change(
                      dataset_dir, processed_dir_prefix
                  )
              else:
                  force_preprocess = True
          if force_preprocess:
              shutil.rmtree(
                  os.path.join(dataset_dir, processed_dir_prefix)
              )
              print(
                  "The on-disk dataset is re-preprocessing, so the existing "
                  + "preprocessed dataset has been removed."
              )
          else:
              print("The dataset is already preprocessed.")
              return os.path.join(dataset_dir, preprocess_metadata_path)
  
      print("Start to preprocess the on-disk dataset.")
  
      # Check if the metadata.yaml exists.
      metadata_file_path = os.path.join(dataset_dir, "metadata.yaml")
      if not os.path.exists(metadata_file_path):
          raise RuntimeError("metadata.yaml does not exist.")
  
      # Read the input config.
      with open(metadata_file_path, "r") as f:
          input_config = yaml.safe_load(f)
  
      # 1. Make `processed_dir_abs` directory if it does not exist.
      os.makedirs(
          os.path.join(dataset_dir, processed_dir_prefix), exist_ok=True
      )
      output_config = deepcopy(input_config)
  
      # 2. Load the edge data and create a DGLGraph.
      if "graph" not in input_config:
          raise RuntimeError(
              "Invalid config: does not contain graph field."
          )
      is_homogeneous = "type" not in input_config["graph"]["nodes"][0]
      if is_homogeneous:
          # Homogeneous graph.
          num_nodes = input_config["graph"]["nodes"][0]["num"]
          edge_fmt = input_config["graph"]["edges"][0]["format"]
          edge_path = input_config["graph"]["edges"][0]["path"]
          src, dst = read_edges(dataset_dir, edge_fmt, edge_path)
          g = dgl.graph((src, dst), num_nodes=num_nodes)
      else:
          # Heterogeneous graph.
          # Construct the num nodes dict.
          num_nodes_dict = {}
          for node_info in input_config["graph"]["nodes"]:
              num_nodes_dict[node_info["type"]] = node_info["num"]
          # Construct the data dict.
          data_dict = {}
          for edge_info in input_config["graph"]["edges"]:
              edge_fmt = edge_info["format"]
              edge_path = edge_info["path"]
              src, dst = read_edges(dataset_dir, edge_fmt, edge_path)
              data_dict[etype_str_to_tuple(edge_info["type"])] = (
                  src,
                  dst,
              )
          # Construct the heterograph.
          g = dgl.heterograph(data_dict, num_nodes_dict)
  
      # 3. Load the sampling related node/edge features and add them to
      # the sampling-graph.
      if input_config["graph"].get("feature_data", None):
          for graph_feature in input_config["graph"]["feature_data"]:
              in_memory = (
                  True
                  if "in_memory" not in graph_feature
                  else graph_feature["in_memory"]
              )
              if graph_feature["domain"] == "node":
                  node_data = read_data(
                      os.path.join(dataset_dir, graph_feature["path"]),
                      graph_feature["format"],
                      in_memory=in_memory,
                  )
                  if is_homogeneous:
                      g.ndata[graph_feature["name"]] = node_data
                  else:
                      g.nodes[graph_feature["type"]].data[
                          graph_feature["name"]
                      ] = node_data
              if graph_feature["domain"] == "edge":
                  edge_data = read_data(
                      os.path.join(dataset_dir, graph_feature["path"]),
                      graph_feature["format"],
                      in_memory=in_memory,
                  )
                  if is_homogeneous:
                      g.edata[graph_feature["name"]] = edge_data
                  else:
                      g.edges[
                          etype_str_to_tuple(graph_feature["type"])
                      ].data[graph_feature["name"]] = edge_data
          if not is_homogeneous:
              # For heterogenous graph, a node/edge feature must cover all
              # node/edge types.
              for feat_name, feat_data in g.ndata.items():
                  existing_types = set(feat_data.keys())
                  assert existing_types == set(g.ntypes), (
                      f"Node feature {feat_name} does not cover all node types."
                      + f"Existing types: {existing_types}."
                      + f"Expected types: {g.ntypes}."
                  )
              for feat_name, feat_data in g.edata.items():
                  existing_types = set(feat_data.keys())
                  assert existing_types == set(g.canonical_etypes), (
                      f"Edge feature {feat_name} does not cover all edge types."
                      + f"Existing types: {existing_types}."
                      + f"Expected types: {g.etypes}."
                  )
  
      # 4. Convert the DGLGraph to a FusedCSCSamplingGraph.
      fused_csc_sampling_graph = from_dglgraph(
          g, is_homogeneous, include_original_edge_id
      )
  
      # 5. Record value of include_original_edge_id.
      output_config[
          "include_original_edge_id"
      ] = include_original_edge_id
  
      # 6. Save the FusedCSCSamplingGraph and modify the output_config.
      output_config["graph_topology"] = {}
      output_config["graph_topology"]["type"] = "FusedCSCSamplingGraph"
      output_config["graph_topology"]["path"] = os.path.join(
          processed_dir_prefix, "fused_csc_sampling_graph.pt"
      )
  
      torch.save(
          fused_csc_sampling_graph,
          os.path.join(
              dataset_dir,
              output_config["graph_topology"]["path"],
          ),
      )
      del output_config["graph"]
  
      # 7. Load the node/edge features and do necessary conversion.
      if input_config.get("feature_data", None):
          has_edge_feature_data = False
          for feature, out_feature in zip(
              input_config["feature_data"],
              output_config["feature_data"],
          ):
              # Always save the feature in numpy format.
              out_feature["format"] = "numpy"
              out_feature["path"] = os.path.join(
                  processed_dir_prefix,
                  feature["path"].replace("pt", "npy"),
              )
              in_memory = (
                  True
                  if "in_memory" not in feature
                  else feature["in_memory"]
              )
              if (
                  not has_edge_feature_data
                  and feature["domain"] == "edge"
              ):
                  has_edge_feature_data = True
              copy_or_convert_data(
                  os.path.join(dataset_dir, feature["path"]),
                  os.path.join(dataset_dir, out_feature["path"]),
                  feature["format"],
                  output_format=out_feature["format"],
                  in_memory=in_memory,
                  is_feature=True,
              )
          if has_edge_feature_data and not include_original_edge_id:
              dgl_warning(
                  "Edge feature is stored, but edge IDs are not saved."
              )
  
      # 8. Save tasks and train/val/test split according to the output_config.
      if input_config.get("tasks", None):
          for input_task, output_task in zip(
              input_config["tasks"], output_config["tasks"]
          ):
              for set_name in [
                  "train_set",
                  "validation_set",
                  "test_set",
              ]:
                  if set_name not in input_task:
                      continue
                  for input_set_per_type, output_set_per_type in zip(
                      input_task[set_name], output_task[set_name]
                  ):
                      for input_data, output_data in zip(
                          input_set_per_type["data"],
                          output_set_per_type["data"],
                      ):
                          # Always save the feature in numpy format.
                          output_data["format"] = "numpy"
                          output_data["path"] = os.path.join(
                              processed_dir_prefix,
                              input_data["path"].replace("pt", "npy"),
                          )
                          copy_or_convert_data(
                              os.path.join(
                                  dataset_dir, input_data["path"]
                              ),
                              os.path.join(
                                  dataset_dir, output_data["path"]
                              ),
                              input_data["format"],
                              output_data["format"],
                          )
  
      # 9. Save the output_config.
      output_config_path = os.path.join(
          dataset_dir, preprocess_metadata_path
      )
      with open(output_config_path, "w") as f:
          yaml.dump(output_config, f)
      print("Finish preprocessing the on-disk dataset.")
  
      # 10. Calculate and save the hash value of the dataset directory.
      hash_value_file = "dataset_hash_value.txt"
      hash_value_file_path = os.path.join(
          dataset_dir, processed_dir_prefix, hash_value_file
      )
      if os.path.exists(hash_value_file_path):
          os.remove(hash_value_file_path)
      dir_hash = calculate_dir_hash(dataset_dir)
      with open(hash_value_file_path, "w") as f:
          f.write(json.dumps(dir_hash, indent=4))
  
      # 11. Return the absolute path of the preprocessing yaml file.
      return output_config_path
  
imports: import json
  import os
  import shutil
  import textwrap
  from copy import deepcopy
  from typing import Dict, List, Union
  import torch
  import yaml
  import dgl
  from ...base import dgl_warning
  from ...data.utils import download, extract_archive
  from ..base import etype_str_to_tuple
  from ..dataset import Dataset, Task
  from ..internal import calculate_dir_hash, check_dataset_change, copy_or_convert_data, get_attributes, read_data, read_edges
  from ..itemset import ItemSet, ItemSetDict
  from ..sampling_graph import SamplingGraph
  from .fused_csc_sampling_graph import from_dglgraph, FusedCSCSamplingGraph
  from .ondisk_metadata import OnDiskGraphTopology, OnDiskMetaData, OnDiskTaskData, OnDiskTVTSet
  from .torch_based_feature_store import TorchBasedFeatureStore
label_desc: Use image node_id 1 for calling a global function with example usage: preprocess_ondisk_dataset(dataset_dir, include_original_edge_id, force_preprocess) and returns: output_config_path, os