code: def aggregate_sketch_true(
      ops: pipeline_dp.pipeline_backend.PipelineBackend,
      col,
      metric: pipeline_dp.Metrics,
  ):
      """Computes raw aggregation on sketches generated by peeker without adding noises.
  
      Aggregation means aggregate values group by partition_key. Both values and
      partition_key are extracted by data extractors.
  
      The input `col` has one entry for each unique (partition_key, privacy_id).
      They are in the format of (partition_key, partition_value, partition_count).
  
      partition_key: the hashed version of the current partition key
      partition_value: the per privacy id per partition_key aggregated value
      partition_count: number of partitions this privacy id contributes to
  
      Note, this is not a DP aggregation.
      Note, this function only supports one aggregation that needs to be either
      count or sum.
  
      Args:
          col: Sketches in the format of (partition_key, partition_value,
              partition_count).
          params: Specifies which metrics to compute and computation parameters.
  
      Returns:
          Raw aggregation metrics.
      """
      if metric == pipeline_dp.Metrics.SUM:
          aggregator_fn = sum
      elif metric == pipeline_dp.Metrics.COUNT:
          aggregator_fn = len
      else:
          raise ValueError(
              "Aggregate sketch only supports sum or count"
          )
      # col: (partition_key, per_user_aggregated_value, partition_count)
      col = ops.map_tuple(
          col, lambda pk, pval, _: (pk, pval), "Drop partition count"
      )
      # col: (partition_key, per_user_aggregated_value)
      col = ops.group_by_key(col, "Group by partition key")
      # col: (partition_key, [per_user_aggregated_value])
      col = ops.map_values(
          col,
          lambda val: aggregator_fn(val),
          "Aggregate by partition key",
      )
      return col
  
file_path: pipelinedp/utility_analysis/peeker_engine.py
imports: import functools
  from typing import Any, Sequence, Tuple, Union
  from apache_beam import pvalue
  import numpy
  import pydp
  import pipeline_dp
label_desc: Use image node_id 1 for calling a global function with example usage: aggregate_sketch_true(ops, col, metric) and returns: col