code: def test_do_with_partial_cache(self, *_):
      self._exec_properties[
          standard_component_specs.MODULE_FILE_KEY
      ] = self._module_file
      self._set_up_input_artifacts(num_examples=10)
      self._input_dict[
          standard_component_specs.EXAMPLES_KEY
      ] = self._example_artifacts
      self._exec_properties[
          standard_component_specs.DISABLE_STATISTICS_KEY
      ] = True
      del self._output_dict[
          standard_component_specs.TRANSFORMED_EXAMPLES_KEY
      ]
      self._transform_executor.Do(
          self._input_dict, self._output_dict, self._exec_properties
      )
      self.assertTrue(
          fileio.exists(self._updated_analyzer_cache_artifact.uri)
      )
      # Only the latest dataset should get cached.
      self.assertLen(
          cache_uris := fileio.listdir(
              self._updated_analyzer_cache_artifact.uri
          ),
          1,
      )
      self.assertIn("example_gen9", cache_uris[0])
  
      output_data_dir = self._get_output_data_dir("2nd_run")
  
      self._make_base_do_params(self._SOURCE_DATA_DIR, output_data_dir)
  
      self._exec_properties[
          standard_component_specs.MODULE_FILE_KEY
      ] = self._module_file
      self._set_up_input_artifacts(num_examples=10)
      self._input_dict[
          standard_component_specs.EXAMPLES_KEY
      ] = self._example_artifacts
      self._exec_properties[
          standard_component_specs.DISABLE_STATISTICS_KEY
      ] = True
      del self._output_dict[
          standard_component_specs.TRANSFORMED_EXAMPLES_KEY
      ]
  
      analyzer_cache_artifact = standard_artifacts.TransformCache()
      analyzer_cache_artifact.uri = (
          self._updated_analyzer_cache_artifact.uri
      )
      self._input_dict[standard_component_specs.ANALYZER_CACHE_KEY] = [
          analyzer_cache_artifact
      ]
      self._transform_executor.Do(
          self._input_dict, self._output_dict, self._exec_properties
      )
      self.assertTrue(fileio.exists(analyzer_cache_artifact.uri))
      # Only the latest uncached dataset should get cached, plus previous cache.
      self.assertLen(
          cache_uris := fileio.listdir(analyzer_cache_artifact.uri), 2
      )
      cache_uris_spans = sum(
          [
              re.findall(r".*example_gen(\d*).*", uri)
              for uri in cache_uris
          ],
          [],
      )
      self.assertCountEqual(cache_uris_spans, ("8", "9"))
  
imports: import copy
  import json
  import os
  import re
  import tempfile
  from unittest import mock
  from absl.testing import parameterized
  import apache_beam
  import tensorflow
  import tensorflow_data_validation
  import tensorflow_transform
  from tensorflow_transform.beam import tft_unit
  from tfx import types
  from tfx.components.testdata.module_file import transform_module
  from tfx.components.transform import executor
  from tfx.dsl.io import fileio
  from tfx.proto import example_gen_pb2
  from tfx.proto import transform_pb2
  from tfx.types import artifact_utils
  from tfx.types import standard_artifacts
  from tfx.types import standard_component_specs
  from tfx.utils import io_utils
  from tfx.utils import name_utils
  from tfx.utils import proto_utils
label_desc: Use image node_id 29 for calling the ExecutorTest obj's underlying member method code with example usage: obj.test_do_with_partial_cache() without return types