code: def ComputeQueryBasedMetrics(  # pylint: disable=invalid-name
      extracts: beam.pvalue.PCollection,
      prediction_key: str,
      query_id: str,
      combine_fns: List[beam.CombineFn],
  ) -> beam.pvalue.PCollection:
      """Computes metrics and plots using the EvalSavedModel.
  
      Args:
        extracts: PCollection of Extracts. The extracts MUST contain a
          FeaturesPredictionsLabels extract keyed by
          tfma.FEATURE_PREDICTIONS_LABELS_KEY and a list of SliceKeyType extracts
          keyed by tfma.SLICE_KEY_TYPES_KEY. Typically these will be added by
          calling the default_extractors function.
        prediction_key: Key in predictions dictionary to use as the prediction (for
          sorting examples within the query). Use the empty string if the Estimator
          returns a predictions Tensor (not a dictionary).
        query_id: Key of query ID column in the features dictionary.
        combine_fns: List of query based metrics combine functions.
  
      Returns:
        PCollection of (slice key, query-based metrics).
      """
  
      missing_query_id_counter = beam.metrics.Metrics.counter(
          constants.METRICS_NAMESPACE, "missing_query_id"
      )
  
      def key_by_query_id(
          extract: types.Extracts, query_id: str
      ) -> Iterator[Tuple[str, types.Extracts]]:
          """Extract the query ID from the extract and key by that."""
          features = extract[
              constants.FEATURES_PREDICTIONS_LABELS_KEY
          ].features
          if query_id not in features:
              missing_query_id_counter.inc()
              return
          feature_value = features[query_id][encoding.NODE_SUFFIX]
          if isinstance(feature_value, tf.compat.v1.SparseTensorValue):
              feature_value = feature_value.values
          if feature_value.size != 1:
              raise ValueError(
                  'Query ID feature "%s" should have exactly 1 value, but '
                  "found %d instead. Values were: %s"
                  % (query_id, feature_value.size(), feature_value)
              )
          yield ("{}".format(feature_value.item()), extract)
  
      def merge_dictionaries(
          dictionaries: Tuple[Dict[str, Any], ...]
      ) -> Dict[str, Any]:
          """Merge dictionaries in a tuple into a single dictionary."""
          result = dict()
          for d in dictionaries:
              intersection = set(d) & set(result)
              if intersection:
                  raise ValueError(
                      "Overlapping keys found when merging dictionaries. "
                      "Intersection was: %s. Keys up to this point: %s "
                      "keys from next dictionary: %s"
                      % (intersection, result.keys(), d.keys())
                  )
              result.update(d)
          return result
  
      # pylint: disable=no-value-for-parameter
      return (
          extracts
          | "KeyByQueryId" >> beam.FlatMap(key_by_query_id, query_id)
          | "CreateQueryExamples"
          >> beam.CombinePerKey(
              CreateQueryExamples(prediction_key=prediction_key)
          )
          | "DropQueryId"
          >> beam.Map(lambda kv: kv[1]._replace(query_id=kv[0]))
          | "CombineGlobally"
          >> beam.CombineGlobally(
              beam.combiners.SingleInputTupleCombineFn(*combine_fns)
          )
          | "MergeDictionaries" >> beam.Map(merge_dictionaries)
          | "AddOverallSliceKey" >> beam.Map(lambda v: ((), v))
      )
  
imports: from typing import Any, Dict, Iterable, Iterator, List, Optional, Tuple
  import apache_beam
  import numpy
  import tensorflow
  from tensorflow_model_analysis import constants
  from tensorflow_model_analysis.api import types
  from tensorflow_model_analysis.eval_saved_model import constants
  from tensorflow_model_analysis.eval_saved_model import encoding
  from tensorflow_model_analysis.eval_saved_model import util
  from tensorflow_model_analysis.evaluators import evaluator
  from tensorflow_model_analysis.evaluators.query_metrics import query_types
  from tensorflow_model_analysis.extractors import extractor
  from tensorflow_model_analysis.extractors import slice_key_extractor
label_desc: Use image node_id 2 for calling a global function with example usage: ComputeQueryBasedMetrics(extracts, prediction_key, query_id, combine_fns) and returns: unknown, result