code: def test_non_cached_dataset_run_twice(self, use_tf_compat_v1):
      if not use_tf_compat_v1:
          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").non_cacheable()
  
      num_analyzers = 2
  
      def preprocessing_fn(inputs):
          return {
              "x_min": tft.min(inputs["x"], name="x")
              + tf.zeros_like(inputs["x"]),
              "s_integerized": tft.compute_and_apply_vocabulary(
                  inputs["s"], vocab_filename="vocab"
              ),
          }
  
      feature_spec = {
          "x": tf.io.FixedLenFeature([], tf.float32),
          "s": tf.io.FixedLenFeature([], tf.string),
      }
      input_data_dict = {
          span_0_key: [
              {
                  "x": -2,
                  "s": "a",
              },
              {
                  "x": 4,
                  "s": "a",
              },
              {
                  "x": 5,
                  "s": "a",
              },
              {
                  "x": 1,
                  "s": "ȟᎥ𝒋ǩľḿꞑȯ𝘱𝑞𝗋𝘴".encode("utf-8"),
              },
          ],
          span_1_key: [
              {
                  "x": 5,
                  "s": "c",
              },
              {
                  "x": 5,
                  "s": "2",
              },
          ],
      }
      expected_vocabulary_contents = np.array(
          [b"a", "ȟᎥ𝒋ǩľḿꞑȯ𝘱𝑞𝗋𝘴".encode("utf-8"), b"c", b"2"],
          dtype=object,
      )
  
      expected_transformed_data = [
          {
              "x_min": -2.0,
              "s_integerized": 2,
          },
          {
              "x_min": -2.0,
              "s_integerized": 3,
          },
      ]
  
      first_run_result = self._run_pipeline(
          feature_spec,
          input_data_dict,
          preprocessing_fn,
          datasets_to_transform=[span_1_key],
          expected_transform_data=expected_transformed_data,
          use_tf_compat_v1=use_tf_compat_v1,
      )
  
      for key in input_data_dict:
          if key.is_cached:
              self.assertIn(key, first_run_result.cache_output)
              self.assertEqual(
                  num_analyzers,
                  len(first_run_result.cache_output[key].cache_dict),
              )
          else:
              self.assertNotIn(key, first_run_result.cache_output)
  
      vocab1_path = (
          first_run_result.transform_output.vocabulary_file_by_name(
              "vocab"
          )
      )
      self.AssertVocabularyContents(
          vocab1_path, expected_vocabulary_contents
      )
  
      metrics = first_run_result.metrics
      # 6 from analyzing 2 spans, and 2 from transform.
      self.assertMetricsCounterEqual(metrics, "num_instances", 8)
      self.assertMetricsCounterEqual(
          metrics, "cache_entries_decoded", 0
      )
      # 2 analyzers, computed for just one dataset.
      self.assertMetricsCounterEqual(
          metrics, "cache_entries_encoded", num_analyzers
      )
      self.assertMetricsCounterEqual(
          metrics,
          "saved_models_created",
          _SINGLE_PHASE_NUM_SAVED_MODELS,
      )
      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_1_key],
          expected_transform_data=expected_transformed_data,
          use_tf_compat_v1=use_tf_compat_v1,
      )
  
      vocab1_path = (
          second_run_result.transform_output.vocabulary_file_by_name(
              "vocab"
          )
      )
      self.AssertVocabularyContents(
          vocab1_path, expected_vocabulary_contents
      )
  
      self.assertFalse(second_run_result.cache_output)
  
      metrics = second_run_result.metrics
      # Only 2 from transform, plus 2 from non-cache dataset.
      self.assertMetricsCounterEqual(metrics, "num_instances", 4)
      self.assertMetricsCounterEqual(
          metrics, "cache_entries_decoded", num_analyzers
      )
      self.assertMetricsCounterEqual(
          metrics, "cache_entries_encoded", 0
      )
  
      # The non cached dataset causes the same number of saved models created as
      # in the first iteration.
      self.assertMetricsCounterEqual(
          metrics,
          "saved_models_created",
          _SINGLE_PHASE_NUM_SAVED_MODELS,
      )
      # Cache coverage allowed us to avoid processing this many bytes of data.
      self.assertMetricsCounterGreater(
          metrics, "analysis_input_bytes_from_cache", 100
      )
  
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 16 for calling the CachedImplTest obj's underlying member method code with example usage: obj.test_non_cached_dataset_run_twice(use_tf_compat_v1) and returns: dict