code: def test_vocab_reserved_tokens_not_cached(self, use_tf_compat_v1):
      if not use_tf_compat_v1:
          tft_unit.skip_if_not_tf2("Tensorflow 2.x required.")
  
      def preprocessing_fn(inputs):
          return {
              "s_integerized": tft.compute_and_apply_vocabulary(
                  inputs["s"],
                  vocab_filename="vocab",
                  reserved_tokens=["reserved"],
              ),
          }
  
      feature_spec = {
          "s": tf.io.FixedLenFeature([], tf.string),
      }
      input_data_dict = {
          analyzer_cache.DatasetKey("span-0"): [dict(s="a")] * 3
          + [dict(s="b")]
      }
      expected_vocabulary_contents = np.array(
          [b"reserved", b"a", b"b"], dtype=object
      )
  
      first_run_result = self._run_pipeline(
          feature_spec,
          input_data_dict,
          preprocessing_fn,
          use_tf_compat_v1=use_tf_compat_v1,
      )
  
      vocab1_path = (
          first_run_result.transform_output.vocabulary_file_by_name(
              "vocab"
          )
      )
      self.AssertVocabularyContents(
          vocab1_path, expected_vocabulary_contents
      )
  
      metrics = first_run_result.metrics
      self.assertMetricsCounterEqual(metrics, "num_instances", 4)
      self.assertMetricsCounterEqual(
          metrics, "cache_entries_decoded", 0
      )
      self.assertMetricsCounterEqual(
          metrics, "cache_entries_encoded", 1
      )
      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,
          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
      self.assertMetricsCounterEqual(metrics, "num_instances", 0)
      self.assertMetricsCounterEqual(
          metrics, "cache_entries_decoded", 1
      )
      self.assertMetricsCounterEqual(
          metrics, "cache_entries_encoded", 0
      )
      self.assertMetricsCounterGreater(
          metrics, "analysis_input_bytes_from_cache", 1
      )
  
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 18 for calling the CachedImplTest obj's underlying member method code with example usage: obj.test_vocab_reserved_tokens_not_cached(use_tf_compat_v1) and returns: dict