code: def publish_execution_results(
      mlmd_handle: metadata.Metadata,
      executor_output: execution_result_pb2.ExecutorOutput,
      execution_info: data_types.ExecutionInfo,
      contexts: list[proto.Context],
  ) -> Optional[typing_utils.ArtifactMultiMap]:
      """Publishes execution result to MLMD for single component run."""
      if executor_output.execution_result.code != status_lib.Code.OK:
          if (
              executor_output.execution_result.code
              == status_lib.Code.CANCELLED
          ):
              execution_state = proto.Execution.CANCELED
          else:
              execution_state = proto.Execution.FAILED
          remove_temporary_task_dirs(tmp_dir=execution_info.tmp_dir)
          node_uid = task_lib.NodeUid(
              pipeline_uid=task_lib.PipelineUid.from_pipeline_id_and_run_id(
                  pipeline_id=execution_info.pipeline_info.id,
                  pipeline_run_id=execution_info.pipeline_run_id,
              ),
              node_id=execution_info.pipeline_node.node_info.id,
          )
          _update_execution_state_in_mlmd(
              mlmd_handle=mlmd_handle,
              node_uid=node_uid,
              execution_id=execution_info.execution_id,
              new_state=execution_state,
              error_code=executor_output.execution_result.code,
              error_msg=executor_output.execution_result.result_message,
              execution_result=executor_output.execution_result,
          )
          return
      remove_temporary_task_dirs(
          stateful_working_dir=execution_info.stateful_working_dir,
          tmp_dir=execution_info.tmp_dir,
      )
      # TODO(b/262040844): Instead of directly using the context manager here, we
      # should consider creating and using wrapper functions.
      with mlmd_state.evict_from_cache(execution_info.execution_id):
          (
              output_dict,
              _,
          ) = execution_publish_utils.publish_succeeded_execution(
              mlmd_handle,
              execution_id=execution_info.execution_id,
              contexts=contexts,
              output_artifacts=execution_info.output_dict,
              executor_output=executor_output,
          )
          return output_dict
  
imports: from __future__ import annotations
  from typing import Optional
  from absl import logging
  from tfx.dsl.io import fileio
  from tfx.orchestration import data_types_utils
  from tfx.orchestration import metadata
  from tfx.orchestration.experimental.core import component_generated_alert_pb2
  from tfx.orchestration.experimental.core import constants
  from tfx.orchestration.experimental.core import event_observer
  from tfx.orchestration.experimental.core import garbage_collection
  from tfx.orchestration.experimental.core import mlmd_state
  from tfx.orchestration.experimental.core import task
  from tfx.orchestration.experimental.core import task_scheduler
  from tfx.orchestration.portable import data_types
  from tfx.orchestration.portable import execution_publish_utils
  from tfx.orchestration.portable.mlmd import execution_lib
  from tfx.proto.orchestration import execution_result_pb2
  from tfx.utils import status
  from tfx.utils import typing_utils
  from ml_metadata import proto
label_desc: Use image node_id 2 for calling a global function with example usage: publish_execution_results(mlmd_handle, executor_output, execution_info, contexts) and returns: output_dict