code: def run_with_executor(
      execution_info: data_types.ExecutionInfo,
      executor: base_executor.BaseExecutor,
  ) -> execution_result_pb2.ExecutorOutput:
      """Invokes executors given an executor instance and input from the Launcher.
  
      Args:
        execution_info: A wrapper of the details of this execution.
        executor: An executor instance.
  
      Returns:
        The output from executor.
      """
      for _, artifact_list in execution_info.input_dict.items():
          for artifact in artifact_list:
              if isinstance(artifact, ValueArtifact):
                  # Read ValueArtifact into memory.
                  artifact.read()
  
      output_dict = copy.deepcopy(execution_info.output_dict)
      result = executor.Do(
          execution_info.input_dict,
          output_dict,
          execution_info.exec_properties,
      )
      if not result:
          # If result is not returned from the Do function, then try to
          # read from the executor_output_uri.
          if fileio.exists(execution_info.execution_output_uri):
              result = execution_result_pb2.ExecutorOutput.FromString(
                  fileio.open(
                      execution_info.execution_output_uri, "rb"
                  ).read()
              )
          else:
              # Old style TFX executor doesn't return executor_output, but modify
              # output_dict and exec_properties in place. For backward compatibility,
              # we use their executor_output and exec_properties to construct
              # ExecutorOutput.
              result = execution_result_pb2.ExecutorOutput()
              outputs_utils.populate_output_artifact(
                  result, output_dict
              )
              outputs_utils.populate_exec_properties(
                  result, execution_info.exec_properties
              )
      return result
  
file_path: tfx/tfx/orchestration/portable/python_executor_operator.py
imports: import copy
  import sys
  from typing import Optional, cast
  from tfx.dsl.components.base import base_executor
  from tfx.dsl.io import fileio
  from tfx.orchestration.portable import base_executor_operator
  from tfx.orchestration.portable import data_types
  from tfx.orchestration.portable import outputs_utils
  from tfx.proto.orchestration import executable_spec_pb2
  from tfx.proto.orchestration import execution_result_pb2
  from tfx.types.value_artifact import ValueArtifact
  from tfx.utils import import_utils
  from google.protobuf import message
label_desc: Use image node_id 1 for calling a global function with example usage: run_with_executor(execution_info, executor) and returns: result