class_docstr: The handler for the system Resolver node.
class_docstr_tok: ["The","handler","for","the","system","Resolver","node","."]
docstr: Runs Resolver specific logic.
  
  Args:
    mlmd_connection: ML metadata connection.
    pipeline_node: The specification of the node that this launcher lauches.
    pipeline_info: The information of the pipeline that this node runs in.
    pipeline_runtime_spec: The runtime information of the pipeline that this
      node runs in.
  
  Returns:
    The execution of the run.
docstr_tok: ["Runs","Resolver","specific","logic",".","Args",":","mlmd_connection",":","ML","metadata","connection",".","pipeline_node",":","The","specification","of","the","node","that","this","launcher","lauches",".","pipeline_info",":","The","information","of","the","pipeline","that","this","node","runs","in",".","pipeline_runtime_spec",":","The","runtime","information","of","the","pipeline","that","this","node","runs","in",".","Returns",":","The","execution","of","the","run","."]
code: def run(
      self,
      mlmd_connection: metadata.Metadata,
      pipeline_node: pipeline_pb2.PipelineNode,
      pipeline_info: pipeline_pb2.PipelineInfo,
      pipeline_runtime_spec: pipeline_pb2.PipelineRuntimeSpec,
  ) -> data_types.ExecutionInfo:
      """Runs Resolver specific logic.
  
      Args:
        mlmd_connection: ML metadata connection.
        pipeline_node: The specification of the node that this launcher lauches.
        pipeline_info: The information of the pipeline that this node runs in.
        pipeline_runtime_spec: The runtime information of the pipeline that this
          node runs in.
  
      Returns:
        The execution of the run.
      """
      logging.info("Running as an resolver node.")
      with mlmd_connection as m:
          # 1.Prepares all contexts.
          contexts = context_lib.prepare_contexts(
              metadata_handle=m, node_contexts=pipeline_node.contexts
          )
  
          # 2. Resolves inputs and execution properties.
          exec_properties = data_types_utils.build_parsed_value_dict(
              inputs_utils.resolve_parameters_with_schema(
                  node_parameters=pipeline_node.parameters
              )
          )
          try:
              resolved_inputs = inputs_utils.resolve_input_artifacts(
                  pipeline_node=pipeline_node, metadata_handle=m
              )
              logging.info(
                  "[%s] Resolved inputs: %s",
                  pipeline_node.node_info.id,
                  resolved_inputs,
              )
          except exceptions.InputResolutionError as e:
              logging.exception(
                  "[%s] Input resolution error: %s",
                  pipeline_node.node_info.id,
                  e,
              )
              execution = execution_publish_utils.register_execution(
                  metadata_handle=m,
                  execution_type=pipeline_node.node_info.type,
                  contexts=contexts,
                  exec_properties=exec_properties,
              )
              execution_publish_utils.publish_failed_execution(
                  metadata_handle=m,
                  contexts=contexts,
                  execution_id=execution.id,
                  executor_output=self._build_error_output(
                      code=e.grpc_code_value
                  ),
              )
              return data_types.ExecutionInfo(
                  execution_id=execution.id,
                  exec_properties=exec_properties,
                  pipeline_node=pipeline_node,
                  pipeline_info=pipeline_info,
              )
  
          # 2a. If Skip (i.e. inside conditional), no execution should be made.
          # TODO(b/197907821): Publish special execution for Skip?
          if isinstance(resolved_inputs, inputs_utils.Skip):
              return data_types.ExecutionInfo()
  
          # 3. Registers execution in metadata.
          execution = execution_publish_utils.register_execution(
              metadata_handle=m,
              execution_type=pipeline_node.node_info.type,
              contexts=contexts,
              exec_properties=exec_properties,
          )
  
          # TODO(b/197741942): Support len > 1.
          if len(resolved_inputs) > 1:
              execution_publish_utils.publish_failed_execution(
                  metadata_handle=m,
                  contexts=contexts,
                  execution_id=execution.id,
                  executor_output=self._build_error_output(
                      _ERROR_CODE_UNIMPLEMENTED,
                      "Handling more than one input dicts not implemented yet.",
                  ),
              )
              return data_types.ExecutionInfo(
                  execution_id=execution.id,
                  exec_properties=exec_properties,
                  pipeline_node=pipeline_node,
                  pipeline_info=pipeline_info,
              )
  
          input_artifacts = resolved_inputs[0]
  
          # 4. Publish the execution as a cached execution with
          # resolved input artifact as the output artifacts.
          execution_publish_utils.publish_internal_execution(
              metadata_handle=m,
              contexts=contexts,
              execution_id=execution.id,
              output_artifacts=input_artifacts,
          )
  
          return data_types.ExecutionInfo(
              execution_id=execution.id,
              input_dict=input_artifacts,
              output_dict=input_artifacts,
              exec_properties=exec_properties,
              pipeline_node=pipeline_node,
              pipeline_info=pipeline_info,
          )
  
file_path: tfx/tfx/orchestration/portable/resolver_node_handler.py
imports: import sys
  from typing import Any, Dict, Optional
  from absl import logging
  import grpc
  from tfx.orchestration import data_types_utils
  from tfx.orchestration import metadata
  from tfx.orchestration.portable import data_types
  from tfx.orchestration.portable import execution_publish_utils
  from tfx.orchestration.portable import inputs_utils
  from tfx.orchestration.portable import system_node_handler
  from tfx.orchestration.portable.input_resolution import exceptions
  from tfx.orchestration.portable.mlmd import context_lib
  from tfx.proto.orchestration import execution_result_pb2
  from tfx.proto.orchestration import pipeline_pb2
label_desc: Use image node_id 2 for calling the ResolverNodeHandler obj's underlying member method code with example usage: obj.run(mlmd_connection, pipeline_node, pipeline_info, pipeline_runtime_spec) and returns: data_types, data_types, data_types, data_types