code: def _cache_df_or_retrieve_cache_data_url(
      df,
      parent_cache_dir_url,
      parquet_row_group_size_bytes,
      compression_codec,
      dtype,
  ):
      """Check whether the df is cached.
  
      If so, return the existing cache file path.
      If not, cache the df into the cache_dir in parquet format and return the
      cache file path.
      Use atexit to delete the cache before the python interpreter exits.
      :param df: A :class:`pyspark.sql.DataFrame` object.
      :param parquet_row_group_size_bytes: An int denoting the number of bytes
          in a parquet row group.
      :param compression_codec: Specify compression codec.
      :param dtype: ``None``, 'float32' or 'float64', specifying the precision of the floating-point
          elements in the output dataset. Integer types will remain unchanged. If ``None``, all types
          will remain unchanged. Default 'float32'.
      :return: A string denoting the path of the saved parquet file.
      """
      # TODO
      #  Improve the cache list by hash table (Note we need use hash(df_plan +
      #  row_group_size)
      with _cache_df_meta_list_lock:
          df_plan = _get_df_plan(df)
          for meta in _cache_df_meta_list:
              if (
                  meta.row_group_size == parquet_row_group_size_bytes
                  and meta.compression_codec == compression_codec
                  and meta.df_plan.sameResult(df_plan)
                  and meta.dtype == dtype
                  and meta.parent_cache_dir_url == parent_cache_dir_url
              ):
                  return meta.cache_dir_url
          # do not find cached dataframe, start materializing.
          cached_df_meta = (
              CachedDataFrameMeta.create_cached_dataframe_meta(
                  df,
                  parent_cache_dir_url,
                  parquet_row_group_size_bytes,
                  compression_codec,
                  dtype,
              )
          )
          _cache_df_meta_list.append(cached_df_meta)
          return cached_df_meta.cache_dir_url
  
imports: import atexit
  import datetime
  import logging
  import os
  import shutil
  import threading
  import time
  import uuid
  from distutils.version import LooseVersion
  from multiprocessing.pool import ThreadPool
  from typing import List, Any
  import pyspark
  from pyarrow import LocalFileSystem
  from pyspark.sql.session import SparkSession
  from pyspark.sql.types import ArrayType, DoubleType, FloatType
  from six.moves.urllib.parse import urlparse
  from petastorm import make_batch_reader
  from petastorm.fs_utils import FilesystemResolver, get_filesystem_and_path_or_paths, normalize_dir_url
  from fsspec.core import strip_protocol
label_desc: Use image node_id 14 for calling a global function with example usage: _cache_df_or_retrieve_cache_data_url(df, parent_cache_dir_url, parquet_row_group_size_bytes, compression_codec, dtype) and returns: cached_df_meta, meta