class_docstr: Helper class to handle metadata I/O.
  
  Not thread-safe without external synchronisation.
docstr: Fetches cached output artifacts if any.
  
  Returns the output artifacts of a cached execution if any. An eligible
  cached execution should take the same input artifacts, execution properties
  and is associated with the same pipeline context.
  
  Args:
    input_artifacts: inputs used by the run.
    exec_properties: execution properties used by the run.
    pipeline_info: info of the current pipeline run.
    component_info: info of the current component.
  
  Returns:
    Dict of cached output artifacts if eligible cached execution is found.
    Otherwise, return None.
docstr_tok: ["Fetches","cached","output","artifacts","if","any",".","Returns","the","output","artifacts","of","a","cached","execution","if","any",".","An","eligible","cached","execution","should","take","the","same","input","artifacts",",","execution","properties","and","is","associated","with","the","same","pipeline","context",".","Args",":","input_artifacts",":","inputs","used","by","the","run",".","exec_properties",":","execution","properties","used","by","the","run",".","pipeline_info",":","info","of","the","current","pipeline","run",".","component_info",":","info","of","the","current","component",".","Returns",":","Dict","of","cached","output","artifacts","if","eligible","cached","execution","is","found",".","Otherwise",",","return","None","."]
code: def get_cached_outputs(
      self,
      input_artifacts: Dict[str, List[_Artifact]],
      exec_properties: Dict[str, Any],
      pipeline_info: data_types.PipelineInfo,
      component_info: data_types.ComponentInfo,
  ) -> Optional[Dict[str, List[_Artifact]]]:
      """Fetches cached output artifacts if any.
  
      Returns the output artifacts of a cached execution if any. An eligible
      cached execution should take the same input artifacts, execution properties
      and is associated with the same pipeline context.
  
      Args:
        input_artifacts: inputs used by the run.
        exec_properties: execution properties used by the run.
        pipeline_info: info of the current pipeline run.
        component_info: info of the current component.
  
      Returns:
        Dict of cached output artifacts if eligible cached execution is found.
        Otherwise, return None.
      """
      absl.logging.debug(
          (
              "Trying to fetch cached output artifacts with the following info: \n"
              "input_artifacts: %s \n"
              "exec_properties: %s \n"
              "component_info %s"
          )
          % (input_artifacts, exec_properties, component_info)
      )
  
      # Step 0: Finds the context of the pipeline. No context means no valid cache
      # results.
      context = self.get_pipeline_context(pipeline_info)
      if context is None:
          absl.logging.warning(
              "Pipeline context not available for %s" % pipeline_info
          )
          return None
  
      # Step 1: Finds historical executions related to the context in step 0.
      historical_executions = dict(
          (e.id, e)
          for e in self.store.get_executions_by_context(context.id)
      )
  
      # Step 2: Filters historical executions to find those that used all the
      # given inputs as input artifacts. The result of this step is a set of
      # reversely sorted execution ids.
      input_ids = collections.defaultdict(set)
      for key, input_list in input_artifacts.items():
          for single_input in input_list:
              input_ids[key].add(single_input.mlmd_artifact.id)
      artifact_to_executions = collections.defaultdict(set)
      for event in self.store.get_events_by_artifact_ids(
          list(set(itertools.chain.from_iterable(input_ids.values())))
      ):
          if event.type == metadata_store_pb2.Event.INPUT:
              artifact_to_executions[event.artifact_id].add(
                  event.execution_id
              )
      common_execution_ids = sorted(
          set.intersection(
              set(historical_executions.keys()),
              *(artifact_to_executions.values()),
          ),
          reverse=True,
      )
  
      # Step 3: Filters candidate executions further based on the followings:
      #   - Shares the given properties
      #   - Is in complete state
      # The maximum number of candidates is capped by MAX_EXECUTIONS_FOR_CACHE.
      expected_previous_execution = self._prepare_execution(
          EXECUTION_STATE_COMPLETE,
          exec_properties,
          pipeline_info=pipeline_info,
          component_info=component_info,
      )
  
      candidate_execution_ids = [
          e_id
          for e_id in common_execution_ids  # pylint: disable=g-complex-comprehension
          if self._is_eligible_previous_execution(
              copy.deepcopy(expected_previous_execution),
              copy.deepcopy(historical_executions[e_id]),
          )
      ]
      candidate_execution_ids = candidate_execution_ids[
          0 : min(
              len(candidate_execution_ids), MAX_EXECUTIONS_FOR_CACHE
          )
      ]
  
      # Step 4: Traverse all candidates, if the input artifacts of a candidate
      # match given input artifacts, return the output artifacts of that execution
      # as result. Note that this is necessary since a candidate execution might
      # use more than the given artifacts.
      candidate_execution_to_events = collections.defaultdict(list)
      for event in self.store.get_events_by_execution_ids(
          candidate_execution_ids
      ):
          candidate_execution_to_events[event.execution_id].append(
              event
          )
  
      for events in candidate_execution_to_events.values():
          # Creates the {key -> artifact id set} for the candidate execution.
          current_input_ids = collections.defaultdict(set)
          for event in events:
              if event.type == metadata_store_pb2.Event.INPUT:
                  current_input_ids[event.path.steps[0].key].add(
                      event.artifact_id
                  )
          # If all inputs match, tries to get the outputs of the execution and uses
          # as the cached outputs of the current execution.
          if current_input_ids == input_ids:
              cached_outputs = self._get_outputs_of_events(events)
              if cached_outputs is not None:
                  return cached_outputs
  
      return None
  
imports: import collections
  import copy
  import hashlib
  import itertools
  import os
  import random
  import time
  import types
  from typing import Any, Dict, List, Optional, Set, Tuple, Type, Union
  import absl
  from tfx.dsl.io import fileio
  from tfx.orchestration import data_types
  from tfx.orchestration.portable.mlmd import event_lib
  from tfx.types import artifact
  from tfx.types import artifact_utils
  import ml_metadata
  from ml_metadata.proto import metadata_store_pb2
label_desc: Use image node_id 23 for calling the Metadata obj's underlying member method code with example usage: obj.get_cached_outputs(input_artifacts, exec_properties, pipeline_info, component_info) and returns: None, None, cached_outputs