code: def _mark_nodes(
      node_map: Mapping[str, node_proto_view.NodeProtoView],
      nodes_to_run: Set[str],
      nodes_required_to_reuse: Set[str],
      nodes_to_reuse: Set[str],
      nodes_requiring_snapshot: Set[str],
      snapshot_node: Optional[str],
  ):
      """Mark nodes."""
      for (
          node_id,
          node,
      ) in node_map.items():  # assumes topological order
          if node_id in nodes_to_run:
              node.execution_options.run.perform_snapshot = (
                  node_id == snapshot_node
              )
              node.execution_options.run.depends_on_snapshot = (
                  node_id in nodes_requiring_snapshot
              )
          elif node_id in nodes_required_to_reuse:
              node.execution_options.skip.reuse_artifacts_mode = (
                  _REUSE_ARTIFACT_REQUIRED
              )
          elif node_id in nodes_to_reuse:
              node.execution_options.skip.reuse_artifacts_mode = (
                  _REUSE_ARTIFACT_OPTIONAL
              )
          else:
              node.execution_options.skip.reuse_artifacts_mode = (
                  pipeline_pb2.NodeExecutionOptions.Skip.NEVER
              )
  
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 7 for calling a global function with example usage: _mark_nodes(node_map, nodes_to_run, nodes_required_to_reuse, nodes_to_reuse, nodes_requiring_snapshot, snapshot_node) without return types