code: def test_cache_with_missing_metadata(self):
      tft_unit.skip_if_not_tf2("Tensorflow 2.x required.")
      span_0_key = analyzer_cache.DatasetKey("span-0")
      span_1_key = analyzer_cache.DatasetKey("span-1")
      span_2_key = analyzer_cache.DatasetKey("span-2")
  
      def preprocessing_fn(inputs):
          return {
              "x_min": tft.min(inputs["x"], name="x")
              + tf.zeros_like(inputs["x"])
          }
  
      feature_spec = {"x": tf.io.FixedLenFeature([], tf.float32)}
      input_data_dict = {
          span_0_key: [],
          span_1_key: [{"x": idx} for idx in range(4)],
          span_2_key: [{"x": idx} for idx in range(2)],
      }  # pyformat: disable
  
      expected_transformed_data = [{"x_min": 0} for _ in range(2)]
  
      first_run_result = self._run_pipeline(
          feature_spec,
          input_data_dict,
          preprocessing_fn,
          should_read_cache=True,
          datasets_to_transform=[span_2_key],
          expected_transform_data=expected_transformed_data,
          use_tf_compat_v1=False,
      )
  
      # Deleting dataset cache metadata files.
      for key in input_data_dict:
          self.assertIn(key, first_run_result.cache_output)
          dataset_cache_dir = os.path.join(self._cache_dir, key.key)
          cache_metadata_files = tf.io.gfile.glob(
              os.path.join(
                  dataset_cache_dir,
                  f"{analyzer_cache._METADATA_FILE_NAME}*",
              )
          )
          self.assertLen(cache_metadata_files, 1)
          os.rename(
              cache_metadata_files[0],
              os.path.join(
                  dataset_cache_dir,
                  f"deleted_{analyzer_cache._METADATA_FILE_NAME}",
              ),
          )
  
      metrics = first_run_result.metrics
      self.assertMetricsCounterEqual(
          metrics, "cache_entries_decoded", 0
      )
      self.assertMetricsCounterEqual(
          metrics, "cache_entries_encoded", 3
      )
      self.assertMetricsCounterEqual(
          metrics, "analysis_input_bytes_from_cache", 0
      )
  
      second_run_result = self._run_pipeline(
          feature_spec,
          input_data_dict,
          preprocessing_fn,
          should_read_cache=True,
          datasets_to_transform=[span_2_key],
          expected_transform_data=expected_transformed_data,
          use_tf_compat_v1=False,
      )
  
      self.assertFalse(second_run_result.cache_output)
  
      metrics = second_run_result.metrics
      self.assertMetricsCounterEqual(
          metrics, "cache_entries_decoded", 3
      )
      self.assertMetricsCounterEqual(
          metrics, "cache_entries_encoded", 0
      )
      # Because the metadata associated with cached datasets was deleted, we
      # report 0 for this counter.
      self.assertMetricsCounterEqual(
          metrics, "analysis_input_bytes_from_cache", 0
      )
  
imports: import functools
  import os
  import struct
  from typing import Callable, Mapping, List
  import uuid
  import apache_beam
  from apache_beam.testing import util
  import numpy
  import tensorflow
  import tensorflow_transform
  from tensorflow_transform import analyzer_nodes
  from tensorflow_transform import common_types
  from tensorflow_transform import impl_helper
  from tensorflow_transform import nodes
  import tensorflow_transform.beam
  from tensorflow_transform.beam import analysis_graph_builder
  from tensorflow_transform.beam import analyzer_cache
  from tensorflow_transform.beam import tft_unit
  from tensorflow_transform.tf_metadata import dataset_metadata
  from tfx_bsl.types import tfx_namedtuple
label_desc: Use image node_id 15 for calling the CachedImplTest obj's underlying member method code with example usage: obj.test_cache_with_missing_metadata() and returns: dict