code: def mark_pipeline(
      pipeline: pipeline_pb2.Pipeline,
      from_nodes: Optional[Collection[str]] = None,
      to_nodes: Optional[Collection[str]] = None,
      skip_nodes: Optional[Collection[str]] = None,
      skip_snapshot_nodes: Optional[Collection[str]] = None,
      snapshot_settings: pipeline_pb2.SnapshotSettings = _default_snapshot_settings,
  ) -> pipeline_pb2.Pipeline:
      """Modifies the Pipeline IR in place, in preparation for partial run.
  
      This function modifies the node-level execution_options to annotate them with
      additional information needed for partial runs, such as which nodes to run,
      which nodes to skip, which node is responsible for performing the snapshot,
      as well as the snapshot settings.
  
      The set of nodes to be run is the set of nodes between from_nodes and to_nodes
      -- i.e., the set of nodes that are reachable by traversing downstream from
      `from_nodes` AND also reachable by traversing upstream from `to_nodes`.
  
      Any `reusable_nodes` will be reused in the partial run, as long as they do not
      depend on a node that is marked to run.
  
      Args:
        pipeline: A valid compiled Pipeline IR proto to be marked.
        from_nodes: The collection of nodes where the "sweep" starts (see detailed
          description). If None, selects all nodes.
        to_nodes: The collection of nodes where the "sweep" ends (see detailed
          description). If None, selects all nodes.
        skip_nodes: **MOST USERS DO NOT NEED THIS.** Use this to force-skip nodes
          that would otherwise have been marked to run. Note that if a node depends
          on nodes that cannot be skipped, then it is still marked to run for
          pipeline result correctness. If None, does not force-skip any node.
        skip_snapshot_nodes: **MOST USERS DO NOT NEED THIS.** Use this to force
          marking snapshot as OPTIONAL for a skipped node. Setting this field can
          be dangerous -- it can cause following partial runs to behave incorrectly
          with missing cached executions.
        snapshot_settings: Settings needed to perform the snapshot step. Defaults to
          using LATEST_PIPELINE_RUN strategy.
  
      Returns:
        Updated pipeline IR.
  
      Raises:
        ValueError: If pipeline's execution_mode is not SYNC.
        ValueError: If pipeline contains a sub-pipeline.
        ValueError: If pipeline was already marked for partial run.
        ValueError: If both from_nodes and to_nodes are empty.
        ValueError: If from_nodes/to_nodes contain node_ids not in the pipeline.
        ValueError: If pipeline is not topologically sorted.
      """
      nodes = node_proto_view.get_view_for_all_in(pipeline)
      _ensure_sync_pipeline(pipeline)
      _ensure_no_partial_run_marks(nodes)
      _ensure_not_full_run(from_nodes, to_nodes)
      _ensure_no_missing_nodes(nodes, from_nodes, to_nodes)
      _ensure_topologically_sorted(nodes)
  
      node_map = make_ordered_node_map(nodes)
  
      from_node_ids = from_nodes or node_map.keys()
      to_node_ids = to_nodes or node_map.keys()
      skip_node_ids = skip_nodes or []
      skip_snapshot_node_ids = set(skip_snapshot_nodes or [])
      nodes_to_run = compute_nodes_to_run(
          node_map, from_node_ids, to_node_ids, skip_node_ids
      )
  
      nodes_required_to_reuse, nodes_to_reuse = _compute_nodes_to_reuse(
          node_map, nodes_to_run, skip_snapshot_node_ids
      )
      nodes_requiring_snapshot = _compute_nodes_requiring_snapshot(
          node_map, nodes_to_run, nodes_to_reuse
      )
      snapshot_node = _pick_snapshot_node(
          node_map, nodes_to_run, nodes_to_reuse
      )
      _mark_nodes(
          node_map,
          nodes_to_run,
          nodes_required_to_reuse,
          nodes_to_reuse,
          nodes_requiring_snapshot,
          snapshot_node,
      )
      pipeline.runtime_spec.snapshot_settings.CopyFrom(
          snapshot_settings
      )
      return pipeline
  
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 4 for calling a global function with example usage: mark_pipeline(pipeline, from_nodes, to_nodes, skip_nodes, skip_snapshot_nodes, snapshot_settings) and returns: pipeline