code: def get_live_output_artifacts_of_node_by_output_key(
      store: mlmd.MetadataStore,
      *,
      pipeline_id: str,
      node_id: str,
      pipeline_run_id: Optional[str] = None,
      execution_states: Optional[
          Sequence["mlmd.proto.Execution.State"]
      ] = None,
  ) -> Mapping[str, Sequence[Sequence[mlmd.proto.Artifact]]]:
      """Get LIVE output artifacts of the given node grouped by output key.
  
      The LIVE output artifacts associated with an output key are represented as a
      list of a list of artifacts.
      1. The outer list represents artifacts generated across all executions.
      2. The inner list represents artifacts generated by one execution.
      3. Elements in the outer list are returned in descending order of the creation
      time of the execution associated with them.
      4. Elements in the inner list have no order guarantee.
      5. If no LIVE output artifacts found for one execution, an empty list will be
      returned.
  
      Args:
        store: A MetadataStore object.
        pipeline_id: A pipeline ID.
        node_id: A node ID.
        pipeline_run_id: The pipeline run ID that the node belongs to. Only
          artifacts from the specified pipeline run are returned if specified.
        execution_states: The MLMD execution state(s) to pull LIVE artifacts from.
          If not specified or is empty, will consider MLMD execution states in
          [COMPLETE, CACHED].
  
      Returns:
        A mapping from output key to all output artifacts from the given node.
      """
      # Step 1: Get LIVE artifacts attributed to node with `node_id`.
      live_artifacts = _get_node_live_artifacts(
          store,
          pipeline_id=pipeline_id,
          node_id=node_id,
          pipeline_run_id=pipeline_run_id,
      )
      if not live_artifacts:
          return {}
  
      # Step 2: Get executions associated with node that created `live_artifacts`
      # ordered by execution creation time in descending order.
      # These executions should satisfy the constraint:
      # min (execution update time) >= min (artifact create time)
      min_live_artifact_create_time = min(
          [a.create_time_since_epoch for a in live_artifacts], default=0
      )
  
      # Within one transaction that updates both artifacts and execution, the
      # timestamp of execution is larger or equal than that of the artifacts.
      # Apply time skew for the artifacts created before cl/574333630 is rolled out.
      # TODO(b/275231956): Remove the following 2 lines if we are sure that there
      # are no more artifacts older than the timestamp.
      if (
          min_live_artifact_create_time
          < orchestration_constants.TIME_SKEW_DATE
      ):
          min_live_artifact_create_time -= 24 * 3600 * 1000
  
      executions_ordered_by_desc_creation_time = get_node_executions(
          store,
          pipeline_id=pipeline_id,
          node_id=node_id,
          pipeline_run_id=pipeline_run_id,
          order_by=mlmd.OrderByField.CREATE_TIME,
          is_asc=False,
          execution_states=execution_states,
          min_last_update_time_since_epoch=min_live_artifact_create_time,
      )
      if not executions_ordered_by_desc_creation_time:
          return {}
  
      # Step 3: Get output events by executions obtained in step 2.
      events_by_executions = store.get_events_by_execution_ids(
          _ids(executions_ordered_by_desc_creation_time)
      )
      output_events = [
          e
          for e in events_by_executions
          if event_lib.is_valid_output_event(e)
      ]
  
      # Step 4: Construct and return `output_artifacts_by_output_key` from events.
      #
      # Create a mapping from execution_id to an empty list first to make sure
      # iteration orders of output_events_by_execution_id and
      # output_artifacts_map_by_execution_id are both in desc order of execution's
      # creation time.
      #
      # The desc order is guaranteed by execution_ids and dict is guaranteed to be
      # iterated in the insertion order of keys.
      output_events_by_execution_id = {
          execution.id: []
          for execution in executions_ordered_by_desc_creation_time
      }
      for event in output_events:
          output_events_by_execution_id[event.execution_id].append(
              event
          )
  
      artifact_ids_by_output_key_map_by_execution_id = {}
      for exec_id, events in output_events_by_execution_id.items():
          output_artifacts_map = (
              event_lib.reconstruct_artifact_id_multimap(events)
          )
          artifact_ids_by_output_key_map_by_execution_id[
              exec_id
          ] = output_artifacts_map
  
      output_artifacts_by_output_key = collections.defaultdict(list)
  
      # Keep only LIVE output artifacts when constructing the result.
      live_artifacts_by_id = {a.id: a for a in live_artifacts}
      for (
          artifact_ids_by_output_key
      ) in artifact_ids_by_output_key_map_by_execution_id.values():
          for (
              output_key,
              artifact_ids,
          ) in artifact_ids_by_output_key.items():
              live_output_artifacts = [
                  live_artifacts_by_id[artifact_id]
                  for artifact_id in artifact_ids
                  if artifact_id in live_artifacts_by_id
              ]
              output_artifacts_by_output_key[output_key].append(
                  live_output_artifacts
              )
      return output_artifacts_by_output_key
  
file_path: tfx/tfx/orchestration/portable/mlmd/store_ext.py
imports: import collections
  import itertools
  from typing import Callable, Mapping, Optional, Sequence, Union
  from tfx.dsl.compiler import compiler_utils
  from tfx.dsl.compiler import constants
  from tfx.orchestration.experimental.core import constants
  from tfx.orchestration.portable.mlmd import event_lib
  from tfx.orchestration.portable.mlmd import filter_query_builder
  import ml_metadata
label_desc: Use image node_id 5 for calling a global function with example usage: get_live_output_artifacts_of_node_by_output_key(store) and returns: output_artifacts_by_output_key, dict, dict