class_docstr: Bounds the contribution of privacy_id per and cross partitions.
  
  It ensures that each privacy_id contributes to not more than
  max_partitions_contributed partitions (cross-partition contribution
  bounding) and not more than max_contributions_per_partition to each
  contributed partition (per-partition contribution bounding), by performing
  sampling if needed.
class_docstr_tok: ["Bounds","the","contribution","of","privacy_id","per","and","cross","partitions",".","It","ensures","that","each","privacy_id","contributes","to","not","more","than","max_partitions_contributed","partitions","(","cross-partition","contribution","bounding",")","and","not","more","than","max_contributions_per_partition","to","each","contributed","partition","(","per-partition","contribution","bounding",")",",","by","performing","sampling","if","needed","."]
docstr: See docstrings for this class and the base class.
docstr_tok: ["See","docstrings","for","this","class","and","the","base","class","."]
code: def bound_contributions(
      self, col, params, backend, report_generator, aggregate_fn
  ):
      """See docstrings for this class and the base class."""
      max_partitions_contributed = params.max_partitions_contributed
      max_contributions_per_partition = (
          params.max_contributions_per_partition
      )
      col = backend.map_tuple(
          col,
          lambda pid, pk, v: ((pid, pk), v),
          "Rekey to ( (privacy_id, partition_key), value))",
      )
      col = backend.sample_fixed_per_key(
          col,
          params.max_contributions_per_partition,
          "Sample per (privacy_id, partition_key)",
      )
      report_generator.add_stage(
          f"Per-partition contribution bounding: for each privacy_id and each"
          f"partition, randomly select max(actual_contributions_per_partition"
          f", {max_contributions_per_partition}) contributions."
      )
      # ((privacy_id, partition_key), [value])
      col = backend.map_values(
          col,
          aggregate_fn,
          "Apply aggregate_fn after per partition bounding",
      )
      # ((privacy_id, partition_key), accumulator)
      # Cross partition bounding
      col = backend.map_tuple(
          col,
          lambda pid_pk, v: (pid_pk[0], (pid_pk[1], v)),
          "Rekey to (privacy_id, (partition_key, accumulator))",
      )
      col = backend.sample_fixed_per_key(
          col,
          params.max_partitions_contributed,
          "Sample per privacy_id",
      )
  
      report_generator.add_stage(
          f"Cross-partition contribution bounding: for each privacy_id "
          f"randomly select max(actual_partition_contributed, "
          f"{max_partitions_contributed}) partitions"
      )
  
      # (privacy_id, [(partition_key, accumulator)])
      def rekey_by_privacy_id_and_unnest(pid_pk_v):
          pid, pk_values = pid_pk_v
          return (((pid, pk), v) for (pk, v) in pk_values)
  
      return backend.flat_map(
          col,
          rekey_by_privacy_id_and_unnest,
          "Rekey by privacy_id and unnest",
      )
  
file_path: pipelinedp/pipeline_dp/contribution_bounders.py
imports: import abc
  import collections
  from typing import Callable, Iterable
  import pipeline_dp
  from pipeline_dp import pipeline_backend
  from pipeline_dp import sampling_utils
label_desc: Use image node_id 1 for calling the SamplingCrossAndPerPartitionContributionBounder obj's underlying member method code with example usage: obj.bound_contributions(col, params, backend, report_generator, aggregate_fn) and returns: backend, unknown