code: def _reuse_pipeline_run_artifacts(
      metadata_handle: metadata.Metadata,
      marked_pipeline: pipeline_pb2.Pipeline,
      base_run_id: Optional[str] = None,
      new_run_id: Optional[str] = None,
  ):
      """Reuses the output Artifacts from a previous pipeline run.
  
      This computes the maximal set of nodes whose outputs can be associated with
      the new pipeline_run without creating any inconsistencies, and reuses their
      node outputs (similar to repeatedly calling `reuse_node_outputs`). It also
      puts a ParentContext into MLMD, with the `base_run_id` being the parent
      context, and the new run_id (provided by the user, or inferred from
      `pipeline`) as the child context.
  
      Args:
        metadata_handle: A handler to access MLMD store.
        marked_pipeline: The output of mark_pipeline function.
        base_run_id: The pipeline_run_id where the output artifacts were produced.
          Defaults to the latest previous pipeline run to use as base_run_id.
        new_run_id: The pipeline_run_id to associate those output artifacts with.
          This function will always attempt to infer the new run id from
          `full_pipeline`'s IR. If not found, it would use the provided
          `new_run_id`. If found, and `new_run_id` is provided, it would verify that
          it is the same as the inferred run id, and raise an error if they are not
          the same.
  
      Raises:
        ValueError: If `marked_pipeline` does not contain a pipeline run id, and
          `new_run_id` is not provided.
        ValueError: If `marked_pipeline` does contain a pipeline run id, and
          `new_run_id` is provided, but they are not the same.
      """
      validated_new_run_id = _get_validated_new_run_id(
          marked_pipeline, new_run_id
      )
      artifact_recycler = _ArtifactRecycler(
          metadata_handle,
          pipeline_name=marked_pipeline.pipeline_info.id,
          new_run_id=validated_new_run_id,
          base_run_id=base_run_id,
      )
  
      reuse_nodes = [
          node
          for node in node_proto_view.get_view_for_all_in(
              marked_pipeline
          )
          if _should_attempt_to_reuse_artifact(node.execution_options)
      ]
      logging.info(
          "Reusing nodes: %s", [n.node_info.id for n in reuse_nodes]
      )
      with concurrent.futures.ThreadPoolExecutor() as executor:
          futures = {}
          for node in reuse_nodes:
              node_id = node.node_info.id
              futures[node_id] = (
                  node,
                  executor.submit(
                      artifact_recycler.reuse_node_outputs, node=node
                  ),
              )
  
          for node_id, (node, future) in futures.items():
              try:
                  future.result()
              except Exception as err:  # pylint: disable=broad-except
                  if _reuse_artifact_required(node.execution_options):
                      # Raise error only if failed to reuse artifacts required.
                      raise
                  err_str = str(err)
                  if (
                      "No previous successful exe" in err_str
                      or "node context" in err_str
                  ):
                      # This is mostly due to no previous execution of the node, so the
                      # error is safe to suppress.
                      logging.info(err_str)
                  else:
                      logging.warning(
                          "Failed to reuse artifacts for node %s. Due to %s",
                          node_id,
                          err_str,
                      )
      artifact_recycler.put_parent_context()
  
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 21 for calling a global function with example usage: _reuse_pipeline_run_artifacts(metadata_handle, marked_pipeline, base_run_id, new_run_id) without return types