code: def test_cache_merge(self):
      base_test_dir = os.path.join(
          os.environ.get(
              "TEST_UNDECLARED_OUTPUTS_DIR", self.get_temp_dir()
          ),
          self._testMethodName,
      )
  
      dataset_key_0 = analyzer_cache.DatasetKey("dataset_key_0")
      dataset_key_1 = analyzer_cache.DatasetKey("dataset_key_1")
      dataset_keys = (dataset_key_0, dataset_key_1)
      cache_keys = list("abcd")
  
      def read_manifests():
          return [
              analyzer_cache._ManifestFile(
                  analyzer_cache._get_dataset_cache_path(
                      base_test_dir, key
                  )
              ).read()
              for key in dataset_keys
          ]
  
      with beam.Pipeline() as p:
          cache_pcoll_dict = {
              dataset_key_0: analyzer_cache.DatasetCache(
                  {
                      "a": p | "CreateA" >> beam.Create([b"a"]),
                      "b": p | "CreateB" >> beam.Create([b"b"]),
                  },
                  None,
              ),
              dataset_key_1: analyzer_cache.DatasetCache(
                  {
                      "c": p | "CreateC" >> beam.Create([b"c"]),
                      "d": p | "CreateD" >> beam.Create([b"d"]),
                  },
                  None,
              ),
          }
          _ = cache_pcoll_dict | analyzer_cache.WriteAnalysisCacheToFS(
              p, base_test_dir, dataset_keys
          )
  
      first_manifests = read_manifests()
  
      with beam.Pipeline() as p:
          cache_pcoll_dict = {
              dataset_key_0: analyzer_cache.DatasetCache(
                  {
                      "c": p | "CreateC" >> beam.Create([b"c"]),
                      "d": p | "CreateD" >> beam.Create([b"d"]),
                  },
                  None,
              ),
              dataset_key_1: analyzer_cache.DatasetCache(
                  {
                      "a": p | "CreateA" >> beam.Create([b"a"]),
                      "b": p | "CreateB" >> beam.Create([b"b"]),
                  },
                  None,
              ),
          }
          _ = cache_pcoll_dict | analyzer_cache.WriteAnalysisCacheToFS(
              p, base_test_dir, dataset_keys
          )
  
      second_manifests = read_manifests()
      self.assertEqual(len(first_manifests), len(second_manifests))
      for manifest_a, manifest_b in zip(
          first_manifests, second_manifests
      ):
          for key_value_pair in manifest_a.items():
              self.assertIn(key_value_pair, manifest_b.items())
  
          self.assertEqual(2, len(manifest_a))
          self.assertCountEqual(
              range(len(manifest_a)), manifest_a.values()
          )
  
          self.assertEqual(4, len(manifest_b))
          self.assertCountEqual(
              range(len(manifest_b)), manifest_b.values()
          )
          self.assertCountEqual(cache_keys, manifest_b.keys())
  
file_path: transform/tensorflow_transform/beam/analyzer_cache_test.py
imports: import os
  import apache_beam
  from apache_beam.testing import util
  import numpy
  import tensorflow
  from tensorflow_transform import analyzer_nodes
  from tensorflow_transform import analyzers
  from tensorflow_transform.beam import analyzer_cache
  from tensorflow_transform import test_case
label_desc: Use image node_id 5 for calling the AnalyzerCacheTest obj's underlying member method code with example usage: obj.test_cache_merge() and returns: unknown