code: def test_df_caching(spark_test_ctx):
      df1 = spark_test_ctx.spark.range(10)
      df2 = spark_test_ctx.spark.range(10)
      df3 = spark_test_ctx.spark.range(20)
  
      # Test caching for the dataframes with the same logical plan
      converter1 = make_spark_converter(df1)
      converter2 = make_spark_converter(df2)
      assert converter1.cache_dir_url == converter2.cache_dir_url
  
      # Test no caching for different dataframes
      converter3 = make_spark_converter(df3)
      assert converter1.cache_dir_url != converter3.cache_dir_url
  
      # Test no caching for the same dataframe with different row group size
      converter11 = make_spark_converter(
          df1, parquet_row_group_size_bytes=8 * 1024 * 1024
      )
      converter21 = make_spark_converter(
          df1, parquet_row_group_size_bytes=16 * 1024 * 1024
      )
      assert converter11.cache_dir_url != converter21.cache_dir_url
  
      # Test no caching for the same dataframe with different compression_codec
      converter12 = make_spark_converter(df1, compression_codec=None)
      converter22 = make_spark_converter(
          df1, compression_codec="snappy"
      )
      assert converter12.cache_dir_url != converter22.cache_dir_url
  
      ori_temp_url = spark_test_ctx.spark.conf.get(
          SparkDatasetConverter.PARENT_CACHE_DIR_URL_CONF
      )
      tempdir = tempfile.mkdtemp("_spark_converter_test1")
      new_temp_url = "file://" + tempdir.replace(os.sep, "/")
      try:
          # Test no caching for the same dataframe with different parent cache dirs
          spark_test_ctx.spark.conf.set(
              SparkDatasetConverter.PARENT_CACHE_DIR_URL_CONF,
              new_temp_url,
          )
          assert ori_temp_url != new_temp_url
          converter13 = make_spark_converter(df1)
          assert converter1.cache_dir_url != converter13.cache_dir_url
  
          # Test caching for the same dataframe with different parent cache dirs
          # that could be normalized to the same parent cache dir
          new_temp_url_2 = new_temp_url + os.sep
          spark_test_ctx.spark.conf.set(
              SparkDatasetConverter.PARENT_CACHE_DIR_URL_CONF,
              new_temp_url_2,
          )
          assert new_temp_url != new_temp_url_2
          converter14 = make_spark_converter(df1)
          assert converter13.cache_dir_url == converter14.cache_dir_url
      finally:
          spark_test_ctx.spark.conf.set(
              SparkDatasetConverter.PARENT_CACHE_DIR_URL_CONF,
              ori_temp_url,
          )
  
imports: import os
  import subprocess
  import sys
  import tempfile
  import threading
  import time
  from distutils.version import LooseVersion
  import numpy
  import pyspark
  import pytest
  import py4j
  import tensorflow.compat.v1
  from pyspark.sql.functions import pandas_udf
  from pyspark.sql.types import ArrayType, BinaryType, BooleanType, ByteType, DoubleType, FloatType, IntegerType, LongType, ShortType, StringType, StructField, StructType
  from six.moves.urllib.parse import urlparse
  from petastorm import make_batch_reader
  from petastorm.fs_utils import FilesystemResolver
  from petastorm.spark import SparkDatasetConverter, make_spark_converter, spark_dataset_converter
  from petastorm.spark.spark_dataset_converter import _check_dataset_file_median_size, _check_parent_cache_dir_url, _check_rank_and_size_consistent_with_horovod, _check_url, _get_horovod_rank_and_size, _get_spark_session, _make_sub_dir_url, register_delete_dir_handler, _wait_file_available, _normalize_databricks_dbfs_url
  from unittest import mock
  from petastorm.tests.test_tf_utils import create_tf_graph
label_desc: Use image node_id 8 for calling a global function with example usage: test_df_caching(spark_test_ctx) without return types