code: def _get_validated_new_run_id(
      pipeline: pipeline_pb2.Pipeline, new_run_id: Optional[str] = None
  ) -> str:
      """Attempts to obtain a unique new_run_id.
  
      Args:
        pipeline: The pipeline IR, whose runtime parameters are already resolved.
        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 `pipeline`.
          If not found, it would use the provided `new_run_id`. If found, and
          `new_run_id` is provided, it would verify that it equals the inferred run
          id, and raise an error if they are not equal.
  
      Returns:
        The validated pipeline_run_id.
  
      Raises:
        ValueError: If `pipeline` does not contain a pipeline run id, and
          `new_run_id` is not provided.
        ValueError: If `pipeline` does contain a pipeline run id, and
          `new_run_id` is provided, but they are not equal.
      """
      inferred_new_run_id = None
      run_id_value = pipeline.runtime_spec.pipeline_run_id
      if run_id_value.HasField("field_value"):
          inferred_new_run_id = run_id_value.field_value.string_value
  
      if not (inferred_new_run_id or new_run_id):
          raise ValueError(
              "Unable to infer new pipeline run id. Either resolve the "
              "pipeline_run_id RuntimeParameter in `filtered_pipeline` first, or "
              "provide a `new_run_id` explicitly."
          )
  
      if (
          new_run_id
          and inferred_new_run_id
          and new_run_id != inferred_new_run_id
      ):
          raise ValueError(
              "Conflicting new pipeline run ids found. pipeline_run_id="
              f"{inferred_new_run_id} was inferred from `full_pipeline`, while "
              f"new_run_id={new_run_id} was explicitly provided. "
              "Consider omitting `new_run_id`, and simply use the pipeline_run_id "
              "inferred from `full_pipeline` as the new_run_id."
          )
  
      # The following OR expression will never evaluate to None, because we have
      # already checked above. However, pytype doesn't know that, so we need to cast
      # to the expression to str so that the return type is str.
      return str(inferred_new_run_id or new_run_id)
  
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 18 for calling a global function with example usage: _get_validated_new_run_id(pipeline, new_run_id) and returns: str