class_docstr: Allows previously-generated Artifacts to be used in a new pipeline run.
  
  By implementing this in a class (instead of a function), we reduce the
  number of MLMD reads when reusing the outputs of multiple nodes in the same
  pipeline run.
class_docstr_tok: ["Allows","previously-generated","Artifacts","to","be","used","in","a","new","pipeline","run",".","By","implementing","this","in","a","class","(","instead","of","a","function",")",",","we","reduce","the","number","of","MLMD","reads","when","reusing","the","outputs","of","multiple","nodes","in","the","same","pipeline","run","."]
docstr: Creates and publishes cache executions.
docstr_tok: ["Creates","and","publishes","cache","executions","."]
code: def _cache_and_publish(
      self,
      existing_executions: List[metadata_store_pb2.Execution],
      node: node_proto_view.NodeProtoView,
  ):
      """Creates and publishes cache executions."""
      if not existing_executions:
          return
  
      # Check if there are any previous attempts to cache and publish.
      node_context = self._get_node_context(node)
      cached_execution_contexts = [
          self._pipeline_context,
          node_context,
          self._new_pipeline_run_context,
      ]
      prev_cache_executions = (
          execution_lib.get_executions_associated_with_all_contexts(
              self._mlmd,
              contexts=[node_context, self._new_pipeline_run_context],
          )
      )
      if not prev_cache_executions:
          new_cached_executions = []
          for e in existing_executions:
              new_cached_executions.append(
                  execution_lib.prepare_execution(
                      metadata_handle=self._mlmd,
                      execution_type=metadata_store_pb2.ExecutionType(
                          id=e.type_id
                      ),
                      state=metadata_store_pb2.Execution.RUNNING,
                      execution_name=str(uuid.uuid4()),
                  )
              )
      else:
          new_cached_executions = [
              e
              for e in prev_cache_executions
              if e.last_known_state
              != metadata_store_pb2.Execution.CACHED
          ]
      logging.info(
          "New cached executions to be published: %s",
          new_cached_executions,
      )
      if not new_cached_executions:
          return
      if len(new_cached_executions) != len(existing_executions):
          raise RuntimeError(
              "The number of new executions is not the same as the number of"
              " existing executions."
          )
  
      output_artifacts_maps = [
          execution_lib.get_output_artifacts(self._mlmd, e.id)
          for e in existing_executions
      ]
      execution_publish_utils.publish_cached_executions(
          self._mlmd,
          contexts=cached_execution_contexts,
          executions=new_cached_executions,
          output_artifacts_maps=output_artifacts_maps,
      )
  
imports: import collections
  import concurrent.futures
  import enum
  from typing import Collection, Dict, Final, List, Mapping, Optional, OrderedDict, Sequence, Set, Tuple
  import uuid
  from absl import logging
  from tfx.dsl.compiler import compiler_utils
  from tfx.dsl.compiler import constants
  from tfx.orchestration import metadata
  from tfx.orchestration import node_proto_view
  from tfx.orchestration.portable import execution_publish_utils
  from tfx.orchestration.portable.mlmd import context_lib
  from tfx.orchestration.portable.mlmd import execution_lib
  from tfx.proto.orchestration import pipeline_pb2
  from ml_metadata.proto import metadata_store_pb2
label_desc: Use image node_id 5 for calling the _ArtifactRecycler obj's underlying member method code with example usage: obj._cache_and_publish(existing_executions, node) without return types