code: def get_unprocessed_inputs(
      metadata_handle: metadata.Metadata,
      resolved_info: ResolvedInfo,
      node: node_proto_view.NodeProtoView,
  ) -> List[InputAndParam]:
      """Get a list of unprocessed input from resolved_info.
  
      Args:
        metadata_handle: A handle to access local MLMD db.
        resolved_info: Resolved input of a node. It may contain processed and
          unprocessed input.
        node: The pipeline node of the input.
  
      Returns:
        A list of InputAndParam that have not been processed.
      """
      if not resolved_info.input_and_params:
          return []
  
      # Finds out the keys that should be ignored.
      input_triggers = (
          node.execution_options.async_trigger.input_triggers
      )
      ignore_keys = {
          k
          for k, t in input_triggers.items()
          if k.startswith("_") or t.no_trigger
      }
  
      max_timestamp_in_each_input: List[int] = []
      for input_and_param in resolved_info.input_and_params:
          max_timestamp_in_one_input = 0
          for key, artifacts in input_and_param.input_artifacts.items():
              if key in ignore_keys or not artifacts:
                  continue
              max_timestamp_in_one_input = max(
                  max_timestamp_in_one_input,
                  max(
                      a.mlmd_artifact.create_time_since_epoch
                      for a in artifacts
                  ),
              )
          max_timestamp_in_each_input.append(max_timestamp_in_one_input)
  
      # A resolved input whose artifacts with max timestamp T is not an input
      # to a execution having creation timestamp < T. So, we only need to
      # get executions with timestamp larger than the minimum timestamp of all
      # the inputs in resolved_info.
      executions = get_executions(
          metadata_handle,
          node,
          additional_filters=[
              (
                  "create_time_since_epoch >="
                  f" {min(max_timestamp_in_each_input, default=0)}"
              ),
              q.Or(
                  [
                      "last_known_state = COMPLETE",
                      "last_known_state = CACHED",
                      "last_known_state = FAILED",
                  ]
              ),
          ],
      )
  
      # Get the successful and failed executions, and group them by input.
      successful_executions_by_input = collections.defaultdict(list)
      failed_executions_by_input = collections.defaultdict(list)
      events = metadata_handle.store.get_events_by_execution_ids(
          [e.id for e in executions]
      )
      for execution in executions:
          input_events = [
              e
              for e in events
              if e.type == metadata_store_pb2.Event.INPUT
              and event_lib.is_valid_input_event(e)
              and e.execution_id == execution.id
          ]
          input_ids_by_key = event_lib.reconstruct_artifact_id_multimap(
              input_events
          )
          # Filters out the keys starting with '_' and the keys should be ingored.
          input_ids_by_key = {
              k: tuple(sorted(v))
              for k, v in input_ids_by_key.items()
              if k not in ignore_keys
          }
          encoded_input = json.dumps(input_ids_by_key, sort_keys=True)
          if execution_lib.is_execution_successful(execution):
              successful_executions_by_input[encoded_input].append(
                  execution
              )
          elif execution_lib.is_execution_failed(execution):
              failed_executions_by_input[encoded_input].append(
                  execution
              )
  
      # Some input artifacts are from external pipelines, so we need to find out the
      # external_id to id mapping in the local db.
      local_id_by_external_id: Dict[str, int] = {}
      for input_and_param in resolved_info.input_and_params:
          for artifact in itertools.chain(
              *input_and_param.input_artifacts.values()
          ):
              if artifact.mlmd_artifact.external_id:
                  local_id_by_external_id[
                      artifact.mlmd_artifact.external_id
                  ] = -1
      if local_id_by_external_id:
          try:
              for (
                  artifact
              ) in metadata_handle.store.get_artifacts_by_external_ids(
                  external_ids=local_id_by_external_id
              ):
                  local_id_by_external_id[
                      artifact.external_id
                  ] = artifact.id
          except errors.NotFoundError:
              # If all the external ids do not exist in local db, we get NotFoundError.
              # It is safe to pass, and we will handle them in the following code.
              pass
          except Exception as e:  # pylint:disable=broad-except
              logging.exception(
                  "Error when getting artifacts by external ids: %s", e
              )
              return []
  
      # Finds out the unprocessed inputs.
      # By default, the retry limit in async pipeline is infinite.
      retry_limit = sys.maxsize
      if node.execution_options.HasField("max_execution_retries"):
          retry_limit = node.execution_options.max_execution_retries
      unprocessed_inputs = []
      for input_and_param in resolved_info.input_and_params:
          resolved_input_ids_by_key = collections.defaultdict(list)
          for key, artifacts in input_and_param.input_artifacts.items():
              for a in artifacts:
                  if a.id:
                      resolved_input_ids_by_key[key].append(a.id)
                  elif a.mlmd_artifact.external_id:
                      resolved_input_ids_by_key[key].append(
                          local_id_by_external_id[
                              a.mlmd_artifact.external_id
                          ]
                      )
              resolved_input_ids_by_key[key] = tuple(
                  resolved_input_ids_by_key[key]
              )
  
          # Filters out the keys starting with '_' and the keys should be ingored.
          resolved_input_ids_by_key = {
              k: tuple(sorted(v))
              for k, v in resolved_input_ids_by_key.items()
              if k not in ignore_keys
          }
  
          encoded_input = json.dumps(
              resolved_input_ids_by_key, sort_keys=True
          )
          if (
              len(failed_executions_by_input[encoded_input])
              >= retry_limit + 1
          ):
              # This input has failed and has also reached its retry limit.
              logging.info(
                  "Node %s has reach retry limit of %d.",
                  node.node_info.id,
                  retry_limit,
              )
          elif encoded_input not in successful_executions_by_input:
              # This input should be processed.
              # If the previous stateful_working_dir_index should be reused, save the
              # index into input_and_param.exec_properties
              if (
                  not node.execution_options.reset_stateful_working_dir
                  and failed_executions_by_input[encoded_input]
              ):
                  failed_executions = failed_executions_by_input[
                      encoded_input
                  ]
                  failed_executions = (
                      execution_lib.sort_executions_newest_to_oldest(
                          failed_executions
                      )
                  )
                  last_failed_execution = failed_executions[0]
                  if input_and_param.exec_properties is None:
                      input_and_param.exec_properties = {}
                  input_and_param.exec_properties[
                      constants.STATEFUL_WORKING_DIR_INDEX
                  ] = outputs_utils.get_stateful_working_dir_index(
                      last_failed_execution
                  )
              unprocessed_inputs.append(input_and_param)
  
      return unprocessed_inputs
  
imports: import collections
  import itertools
  import json
  import sys
  import textwrap
  from typing import Callable, Dict, Iterable, List, MutableMapping, Optional, Sequence, Type
  import uuid
  from absl import logging
  import attr
  from tfx import types
  from tfx.dsl.compiler import constants
  from tfx.dsl.compiler import placeholder_utils
  from tfx.orchestration import data_types_utils
  from tfx.orchestration import metadata
  from tfx.orchestration import node_proto_view
  from tfx.orchestration.experimental.core import constants
  from tfx.orchestration.experimental.core import mlmd_state
  from tfx.orchestration.experimental.core import task
  from tfx.orchestration import mlmd_connection_manager
  from tfx.orchestration.portable import data_types
  from tfx.orchestration.portable import inputs_utils
  from tfx.orchestration.portable import outputs_utils
  from tfx.orchestration.portable.input_resolution import exceptions
  from tfx.orchestration.portable.mlmd import common_utils
  from tfx.orchestration.portable.mlmd import context_lib
  from tfx.orchestration.portable.mlmd import event_lib
  from tfx.orchestration.portable.mlmd import execution_lib
  from tfx.orchestration.portable.mlmd import filter_query_builder
  from tfx.proto.orchestration import pipeline_pb2
  from tfx.utils import proto_utils
  from tfx.utils import status
  from tfx.utils import typing_utils
  from tfx.orchestration.experimental.core import deployment_config_utils
  import ml_metadata
  from ml_metadata import errors
  from ml_metadata.proto import metadata_store_pb2
label_desc: Use image node_id 14 for calling a global function with example usage: get_unprocessed_inputs(metadata_handle, resolved_info, node) and returns: unprocessed_inputs, list, list