code: def _resolve_input_graph_ref(
      mlmd_handle: metadata.Metadata,
      node_inputs: pipeline_pb2.NodeInputs,
      input_key: str,
      resolved: Dict[str, List[_Entry]],
  ) -> None:
      """Resolves an `InputGraphRef` and put resolved result into `resolved`.
  
      This also resolves other `node_inputs.inputs` that share the same graph source
      (i.e. `InputGraphRef` with the same `graph_id`).
  
      Args:
        mlmd_handle: A `Metadata` instance.
        node_inputs: A `NodeInputs` proto.
        input_key: A target input key whose corresponding `InputSpec` has an
            `InputGraphRef`.
        resolved: A dict that contains the already resolved inputs, and to which the
            resolved result would be written from this function.
      """
      graph_id = node_inputs.inputs[input_key].input_graph_ref.graph_id
      input_graph = node_inputs.input_graphs[graph_id]
      graph_output_type = input_graph.nodes[
          input_graph.result_node
      ].output_data_type
      # We will resolve all node_inputs.inputs at once that has the `InputGraphRef`
      # for the same graph (== `graph_id`).
      same_graph_inputs = {
          key: input_spec.input_graph_ref
          for key, input_spec in node_inputs.inputs.items()
          if input_spec.input_graph_ref.graph_id == graph_id
      }
  
      graph_fn, graph_input_keys = input_graph_resolver.build_graph_fn(
          mlmd_handle, node_inputs.input_graphs[graph_id]
      )
      for partition, input_dict in _join_artifacts(
          resolved, graph_input_keys
      ):
          result = graph_fn(input_dict)
          if graph_output_type == _DataType.ARTIFACT_LIST:
              # result == [Artifact()]
              resolved[input_key].append(
                  (partition, _filter_live(result))
              )
          elif graph_output_type == _DataType.ARTIFACT_MULTIMAP:
              # result == {'x': [Artifact()], 'y': [Artifact()]}
              for (
                  each_input_key,
                  input_graph_ref,
              ) in same_graph_inputs.items():
                  resolved[each_input_key].append(
                      (
                          partition,
                          _filter_live(result[input_graph_ref.key]),
                      )
                  )
          elif graph_output_type == _DataType.ARTIFACT_MULTIMAP_LIST:
              # result == [{'x': [Artifact()]}, {'x': [Artifact()]}]
              for index, each_result in enumerate(result):
                  new_partition = partition | {graph_id: index}
                  for (
                      each_input_key,
                      input_graph_ref,
                  ) in same_graph_inputs.items():
                      resolved[each_input_key].append(
                          (
                              new_partition,
                              _filter_live(
                                  each_result[input_graph_ref.key]
                              ),
                          )
                      )
  
imports: import collections
  from typing import Dict, Iterable, List, Mapping, Sequence, Tuple, TypeVar
  from tfx import types
  from tfx.dsl.compiler import placeholder_utils
  from tfx.orchestration import metadata
  from tfx.orchestration import mlmd_connection_manager
  from tfx.orchestration.portable import data_types
  from tfx.orchestration.portable.input_resolution import exceptions
  from tfx.orchestration.portable.input_resolution import input_graph_resolver
  from tfx.orchestration.portable.input_resolution import partition_utils
  from tfx.orchestration.portable.input_resolution import channel_resolver
  from tfx.proto.orchestration import pipeline_pb2
  from tfx.types import artifact_utils
  from tfx.utils import topsort
  from tfx.utils import typing_utils
  from tfx.utils import governance_utils
label_desc: Use image node_id 7 for calling a global function with example usage: _resolve_input_graph_ref(mlmd_handle, node_inputs, input_key, resolved) without return types