class_docstr: Performs DP aggregations.
class_docstr_tok: ["Performs","DP","aggregations","."]
code: def _aggregate(
      self,
      col,
      params: pipeline_dp.AggregateParams,
      data_extractors: pipeline_dp.DataExtractors,
      public_partitions,
  ):
      if params.custom_combiners:
          # TODO(dvadym): after finishing implementation of custom combiners
          # to figure out whether it makes sense to encapsulate creation of
          # combiners in one function instead of considering 2 cases -
          # standard combiners and custom combiners.
          combiner = (
              combiners.create_compound_combiner_with_custom_combiners(
                  params,
                  self._budget_accountant,
                  params.custom_combiners,
              )
          )
      else:
          combiner = self._create_compound_combiner(params)
  
      col = self._extract_columns(col, data_extractors)
      # col : (privacy_id, partition_key, value)
      if (
          public_partitions is not None
          and not params.public_partitions_already_filtered
      ):
          col = self._drop_partitions(
              col,
              public_partitions,
              partition_extractor=lambda row: row[1],
          )
          self._add_report_stage(
              f"Public partition selection: dropped non public partitions"
          )
      if not params.contribution_bounds_already_enforced:
          contribution_bounder = self._create_contribution_bounder(
              params, combiner.expects_per_partition_sampling()
          )
          col = contribution_bounder.bound_contributions(
              col,
              params,
              self._backend,
              self._current_report_generator,
              combiner.create_accumulator,
          )
          # col : ((privacy_id, partition_key), accumulator)
  
          col = self._backend.map_tuple(
              col, lambda pid_pk, v: (pid_pk[1], v), "Drop privacy id"
          )
          # col : (partition_key, accumulator)
      else:
          col = self._backend.map(
              col, lambda row: row[1:], "Remove privacy_id"
          )
          # col : (partition_key, value)
  
          col = self._backend.map_values(
              col,
              lambda value: combiner.create_accumulator([value]),
              "Wrap values into accumulators",
          )
          # col : (partition_key, accumulator)
  
      if public_partitions:
          col = self._add_empty_public_partitions(
              col, public_partitions, combiner.create_accumulator
          )
      # col : (partition_key, accumulator)
  
      col = self._backend.combine_accumulators_per_key(
          col, combiner, "Reduce accumulators per partition key"
      )
      # col : (partition_key, accumulator)
  
      if (
          public_partitions is None
          and not params.post_aggregation_thresholding
      ):
          # Perform private partition selection.
          max_rows_per_privacy_id = 1
  
          if params.contribution_bounds_already_enforced:
              # This regime assumes the input data doesn't have privacy IDs,
              # and therefore we didn't group by them and cannot guarantee one
              # row corresponds to exactly one privacy ID.
              max_rows_per_privacy_id = (
                  params.max_contributions
                  or params.max_contributions_per_partition
              )
  
          col = self._select_private_partitions_internal(
              col,
              params.max_partitions_contributed,
              max_rows_per_privacy_id,
              params.partition_selection_strategy,
              params.pre_threshold,
          )
      # col : (partition_key, accumulator)
  
      # Compute DP metrics.
      self._add_report_stages(combiner.explain_computation())
      col = self._backend.map_values(
          col, combiner.compute_metrics, "Compute DP metrics"
      )
  
      if params.post_aggregation_thresholding:
          col = self._drop_partitions_under_threshold(col)
  
      return col
  
imports: import functools
  from typing import Any, Callable, Optional, Sequence, Tuple, Union
  import pipeline_dp
  from pipeline_dp import budget_accounting
  from pipeline_dp import combiners
  from pipeline_dp import contribution_bounders
  from pipeline_dp import dp_computations
  from pipeline_dp import partition_selection
  from pipeline_dp import pipeline_functions
  from pipeline_dp import report_generator
  from pipeline_dp import sampling_utils
  from pipeline_dp.dataset_histograms import computing_histograms
  from pipeline_dp.private_contribution_bounds import PrivateL0Calculator
label_desc: Use image node_id 8 for calling the DPEngine obj's underlying member method code with example usage: obj._aggregate(col, params, data_extractors, public_partitions) and returns: col