Add MaxContributions (L1 Bounding) to Privacy on Beam - Count

go/checks/checks.go

@@ -254,6 +254,56 @@ func CheckMaxContributionsPerPartition(maxContributionsPerPartition int64) error
 	return nil
 }
 
+// CheckContributionBoundingOptions returns an error unless exactly one of MaxContributions and MaxPartitionsContributed is set.
+func CheckContributionBoundingOptions(maxContributions, maxPartitionsContributed int64) error {
+	if maxContributions > 0 && maxPartitionsContributed > 0 {
+		return fmt.Errorf("MaxContributions and MaxPartitionsContributed are both set, exactly one should be set")
+	}
+	if maxContributions <= 0 && maxPartitionsContributed <= 0 {
+		return fmt.Errorf("MaxContributions and MaxPartitionsContributed are both unset, exactly one should be set")
+	}
+	return nil
+}
+
+// CheckContributionBoundingWithMaxValue checks that either MaxContributions or {MaxValue, MaxPartitionsContributed} is set, but not both.
+// If MaxContributions is set, {MaxValue, MaxPartitionsContributed} must be 0.
+// If {MaxValue, MaxPartitionsContributed} are set, MaxContributions must be 0.
+func CheckContributionBoundingOptionsWithMaxValue(maxContributions, maxPartitionsContributed, maxValue int64) error {
+	if maxContributions < 0 {
+		return fmt.Errorf("MaxContributions must be non-negative, was %d instead", maxContributions)
+	}
+	if maxValue < 0 {
+		return fmt.Errorf("MaxValue must be non-negative, was %d instead", maxValue)
+	}
+	if maxPartitionsContributed < 0 {
+		return fmt.Errorf("MaxPartitionsContributed must be non-negative, was %d instead", maxPartitionsContributed)
+	}
+
+	maxContributionsSet := maxContributions > 0
+	maxValueSet := maxValue > 0
+	maxPartitionsContributedSet := maxPartitionsContributed > 0
+
+	if maxContributionsSet {
+		// MaxContributions configuration must be used
+		if maxValueSet || maxPartitionsContributedSet {
+			return fmt.Errorf("when MaxContributions is set, MaxValue and MaxPartitionsContributed must be 0")
+		}
+		return nil
+	}
+	// MaxValue/MaxPartitionsContributed configuration must be used
+	if !maxValueSet && !maxPartitionsContributedSet {
+		return fmt.Errorf("when MaxContributions is not set, both MaxValue and MaxPartitionsContributed must be set to a positive value")
+	}
+	if !maxValueSet {
+		return fmt.Errorf("MaxValue must be set to a positive value, was %d instead", maxValue)
+	}
+	if !maxPartitionsContributedSet {
+		return fmt.Errorf("MaxPartitionsContributed must be set to a positive value, was %d instead", maxPartitionsContributed)
+	}
+	return nil
+}
+
+
 // CheckAlpha returns an error if the supplied alpha is not between 0 and 1.
 func CheckAlpha(alpha float64) error {
 	if alpha <= 0 || alpha >= 1 || math.IsNaN(alpha) || math.IsInf(alpha, 0) {

go/dpagg/select_partition.go

@@ -129,9 +129,18 @@ type PreAggSelectPartitionOptions struct {
 	// for more information.
 	// Optional.
 	PreThreshold int64
-	// MaxPartitionsContributed is the number of distinct partitions a single
-	// privacy unit can contribute to. Required.
+	// MaxPartitionsContributed is the number of distinct partitions a single privacy unit can contribute to. 
+	// For PreAggSelectionPartition, setting MaxPartitionsContributed is functionally the same as 
+	// setting MaxContributions, but are kept separate for consistency with other aggregations.
+	//
+	// Mutually exclusive with MaxContributions. One of the two options is required.
 	MaxPartitionsContributed int64
+	// MaxContributions is the number of distinct contributions a single
+	// privacy unit can make. For PreAggSelectionPartition, setting MaxContributions is functionally 
+	// the same as setting MaxPartitionsContributed, but are kept separate for consistency with other aggregations.
+	//
+	// Mutually exclusive with MaxPartitionsContributed. One of the two options is required.
+	MaxContributions int64
 }
 
 // NewPreAggSelectPartition constructs a new PreAggSelectPartition from opt.
@@ -148,11 +157,18 @@ func NewPreAggSelectPartition(opt *PreAggSelectPartitionOptions) (*PreAggSelectP
 		opt.PreThreshold = 1
 	}
 
+	if err := checks.CheckContributionBoundingOptions(opt.MaxContributions, opt.MaxPartitionsContributed); err != nil {
+		return nil, fmt.Errorf("NewPreAggSelectPartition: %v", err)
+	}
+	l0Sensitivity := opt.MaxPartitionsContributed
+	if opt.MaxContributions > 0 {
+		l0Sensitivity = opt.MaxContributions
+	}
 	s := PreAggSelectPartition{
 		epsilon:       opt.Epsilon,
 		delta:         opt.Delta,
 		preThreshold:  opt.PreThreshold,
-		l0Sensitivity: opt.MaxPartitionsContributed,
+		l0Sensitivity: l0Sensitivity,
 	}
 
 	if err := checks.CheckDeltaStrict(s.delta); err != nil {

go/dpagg/select_partition_test.go

@@ -31,13 +31,22 @@ func TestNewPreAggSelectPartition(t *testing.T) {
 		want    *PreAggSelectPartition
 		wantErr bool
 	}{
-		{"MaxPartitionsContributed is not set",
+		{"MaxPartitionsContributed and MaxContributions are not set",
 			&PreAggSelectPartitionOptions{
 				Epsilon: ln3,
 				Delta:   tenten,
 			},
 			nil,
 			true},
+		{"MaxPartitionsContributed and MaxContributions are set at same time",
+			&PreAggSelectPartitionOptions{
+				Epsilon:                  ln3,
+				Delta:                    tenten,
+				MaxPartitionsContributed: 1,
+				MaxContributions:         1,
+			},
+			nil,
+			true},
 		{"Epsilon is not set",
 			&PreAggSelectPartitionOptions{
 				Delta:                    tenten,

go/dpagg/sum.go

@@ -74,27 +74,41 @@ func bsEquallyInitializedint64(s1, s2 *BoundedSumInt64) bool {
 
 // BoundedSumInt64Options contains the options necessary to initialize a BoundedSumInt64.
 type BoundedSumInt64Options struct {
-	Epsilon                  float64 // Privacy parameter ε. Required.
-	Delta                    float64 // Privacy parameter δ. Required with Gaussian noise, must be 0 with Laplace noise.
-	MaxPartitionsContributed int64   // How many distinct partitions may a single privacy unit contribute to? Required.
-	// Lower and Upper bounds for clamping. Required; must be such that Lower <= Upper.
+	Epsilon float64 // Privacy parameter ε. Required.
+	Delta   float64 // Privacy parameter δ. Required with Gaussian noise, must be 0 with Laplace noise.
+	// How many distinct partitions may a single privacy unit contribute to?
+	// Mutually exclusive with MaxContributions. Required to be specified along with Lower and Upper when MaxContributions is not set.
+	MaxPartitionsContributed int64
+	// Lower and Upper bounds for clamping. Must be such that Lower <= Upper.
+	// Mutually exclusive with MaxContributions. Required to be specified along with MaxPartitionsContributed when MaxContributions is not set.
 	Lower, Upper int64
 	Noise        noise.Noise // Type of noise used in BoundedSum. Defaults to Laplace noise.
 	// How many times may a single privacy unit contribute to a single partition?
 	// Defaults to 1. This is only needed for other aggregation functions using BoundedSum;
 	// which is why the option is not exported.
+	//
+	// maxContributionsPerPartition is mutually exclusive with MaxContributions. This option has no effect if MaxContributions is set.
 	maxContributionsPerPartition int64
+	// How many times may a single privacy unit contribute in total to all partitions?
+	// Currently only used for Count aggregation function.
+	//
+	// Mutually exclusive with set of {MaxPartitionsContributed, Lower, Upper}. Required when {MaxPartitionsContributed, Lower, Upper} are not set.
+	MaxContributions int64
 }
 
 // NewBoundedSumInt64 returns a new BoundedSumInt64, whose sum is initialized at 0.
 func NewBoundedSumInt64(opt *BoundedSumInt64Options) (*BoundedSumInt64, error) {
 	if opt == nil {
 		opt = &BoundedSumInt64Options{} // Prevents panicking due to a nil pointer dereference.
 	}
+	err := checks.CheckContributionBoundingOptions(opt.MaxContributions, opt.MaxPartitionsContributed)
+	if err != nil {
+		return nil, fmt.Errorf("NewBoundedSumInt64: %w", err)
+	}
 
-	l0 := opt.MaxPartitionsContributed
-	if l0 == 0 {
-		return nil, fmt.Errorf("NewBoundedSumInt64: MaxPartitionsContributed must be set")
+	l0, err := getL0Int(opt.MaxContributions, opt.MaxPartitionsContributed)
+	if err != nil {
+		return nil, fmt.Errorf("NewBoundedSumInt64: %w", err)
 	}
 
 	maxContributionsPerPartition := opt.maxContributionsPerPartition
@@ -108,10 +122,9 @@ func NewBoundedSumInt64(opt *BoundedSumInt64Options) (*BoundedSumInt64, error) {
 	}
 	// Check bounds & use them to compute L_∞ sensitivity
 	lower, upper := opt.Lower, opt.Upper
-	if lower == 0 && upper == 0 {
-		return nil, fmt.Errorf("NewBoundedSumInt64: Lower and Upper must be set (automatic bounds determination is not implemented yet). Lower and Upper cannot be both 0")
+	if opt.MaxPartitionsContributed > 0 && lower == 0 && upper == 0 {
+		return nil, fmt.Errorf("NewBoundedSumInt64: When using MaxPartitionsContributed, Lower and Upper must be set (automatic bounds determination is not implemented yet). Lower and Upper cannot be both 0")
 	}
-	var err error
 	switch noise.ToKind(opt.Noise) {
 	case noise.Unrecognised:
 		err = checks.CheckBoundsInt64IgnoreOverflows(lower, upper)
@@ -121,7 +134,20 @@ func NewBoundedSumInt64(opt *BoundedSumInt64Options) (*BoundedSumInt64, error) {
 	if err != nil {
 		return nil, fmt.Errorf("NewBoundedSumInt64: %w", err)
 	}
-	lInf, err := getLInfInt(lower, upper, maxContributionsPerPartition)
+
+	var lInf int64
+	if opt.MaxContributions > 0 {
+		// When using MaxContributions, we set l0=1 and lInf=MaxContributions to represent
+		// the L1 sensitivity (MaxContributions) in the form expected by the noise layer (l0 * lInf).
+		// This does not hold for privacy-ID count aggregations, where l0=MaxContributions and lInf=1.
+		lInf = opt.MaxContributions
+		// When using MaxContributions, no per-partition contribution bounding is performed.
+		// upper is set to a default of match.MaxInt64 because it is being used in the Add function below, 
+		// making it a no-op.
+		upper = math.MaxInt64
+	} else {
+		lInf, err = getLInfInt(lower, upper, maxContributionsPerPartition)
+	}
 	if err != nil {
 		if noise.ToKind(opt.Noise) == noise.Unrecognised {
 			// Ignore sensitivity overflows if noise is not recognised.
@@ -160,6 +186,16 @@ func lInfIntOverflows(bound, maxContributionsPerPartition int64) bool {
 	return mult/maxContributionsPerPartition != bound
 }
 
+func getL0Int(maxContributions, maxPartitionsContributed int64) (int64, error) {
+	if err := checks.CheckContributionBoundingOptions(maxContributions, maxPartitionsContributed); err != nil {
+		return 0, err
+	}
+	if maxContributions > 0 {
+		return 1, nil
+	}
+	return maxPartitionsContributed, nil
+}
+
 // getLInfInt checks that the sensitivity parameters will not create overflow errors,
 // and returns the L_inf sensitivity of the BoundedSum object, which is calculated by the
 // formula = max(|lower|, |upper|) * maxContributionsPerPartition.

go/dpagg/sum_test.go

@@ -287,7 +287,7 @@ func TestNewBoundedSumInt64(t *testing.T) {
 		want    *BoundedSumInt64
 		wantErr bool
 	}{
-		{"MaxPartitionsContributed is not set",
+		{"MaxPartitionsContributed is not set when using maxContributionsPerPartition",
 			&BoundedSumInt64Options{
 				Epsilon:                      ln3,
 				Delta:                        tenten,
@@ -298,7 +298,7 @@ func TestNewBoundedSumInt64(t *testing.T) {
 			},
 			nil,
 			true},
-		{"maxContributionsPerPartition is not set",
+		{"maxContributionsPerPartition is not set when using MaxPartitionsContributed",
 			&BoundedSumInt64Options{
 				Epsilon:                  ln3,
 				Delta:                    0,
@@ -320,6 +320,16 @@ func TestNewBoundedSumInt64(t *testing.T) {
 				state:           defaultState,
 			},
 			false},
+		{"MaxContributions is not set when not using maxContributionsPerPartition and MaxPartitionsContributed",
+			&BoundedSumInt64Options{
+				Epsilon: ln3,
+				Delta:   0,
+				Lower:   -1,
+				Upper:   5,
+				Noise:   noNoise{},
+			},
+			nil,
+			true},
 		{"Noise is not set",
 			&BoundedSumInt64Options{
 				Epsilon:                      ln3,
@@ -707,6 +717,24 @@ func TestCheckMergeBoundedSumInt64Compatibility(t *testing.T) {
 				maxContributionsPerPartition: 2,
 			},
 			false},
+		{"same options, all fields filled while using MaxContributions",
+			&BoundedSumInt64Options{
+				Epsilon:          ln3,
+				Delta:            tenten,
+				Lower:            -1,
+				Upper:            5,
+				Noise:            noise.Gaussian(),
+				MaxContributions: 2,
+			},
+			&BoundedSumInt64Options{
+				Epsilon:          ln3,
+				Delta:            tenten,
+				Lower:            -1,
+				Upper:            5,
+				Noise:            noise.Gaussian(),
+				MaxContributions: 2,
+			},
+			false},
 		{"same options, only required fields filled",
 			&BoundedSumInt64Options{
 				Epsilon:                  ln3,
@@ -721,6 +749,20 @@ func TestCheckMergeBoundedSumInt64Compatibility(t *testing.T) {
 				MaxPartitionsContributed: 1,
 			},
 			false},
+		{"same options, only required fields filled while using MaxContributions",
+			&BoundedSumInt64Options{
+				Epsilon:          ln3,
+				Lower:            -1,
+				Upper:            5,
+				MaxContributions: 2,
+			},
+			&BoundedSumInt64Options{
+				Epsilon:          ln3,
+				Lower:            -1,
+				Upper:            5,
+				MaxContributions: 2,
+			},
+			false},
 		{"different epsilon",
 			&BoundedSumInt64Options{
 				Epsilon:                  ln3,
@@ -783,6 +825,20 @@ func TestCheckMergeBoundedSumInt64Compatibility(t *testing.T) {
 				MaxPartitionsContributed:     1,
 			},
 			true},
+		{"different MaxContributions",
+			&BoundedSumInt64Options{
+				Epsilon:                  ln3,
+				Lower:                    -1,
+				Upper:                    5,
+				MaxContributions:         2,
+			},
+			&BoundedSumInt64Options{
+				Epsilon:                  ln3,
+				Lower:                    -1,
+				Upper:                    5,
+				MaxContributions:         5,
+			},
+			true},
 		{"different lower bound",
 			&BoundedSumInt64Options{
 				Epsilon:                  ln3,

privacy-on-beam/pbeam/aggregations.go

@@ -98,7 +98,8 @@ func randBool(_, _ beam.V) bool {
 // boundContributions takes a PCollection<K,V> as input, and for each key, selects and returns
 // at most contributionLimit records with this key. The selection is "mostly random":
 // the records returned are selected randomly, but the randomness isn't secure.
-// This is fine to use in the cross-partition bounding stage or in the per-partition bounding stage,
+// This is fine to use in the cross-partition bounding stage, the per-partition bounding stage,
+// or per-privacy identifier contribution bounding stage,
 // since the privacy guarantee doesn't depend on the privacy unit contributions being selected randomly.
 //
 // In order to do the cross-partition contribution bounding we need:
@@ -111,6 +112,13 @@ func randBool(_, _ beam.V) bool {
 //  1. the key to be the pair = {privacy ID, partition ID}.
 //  2. the value to be just the value which is associated with that {privacy ID, partition ID} pair
 //     (there could be multiple entries with the same key).
+//
+// In order to do per-privacy-ID contribution bounding (L1 norm) we need:
+//  1. each record to represent a contribution of 1, such as Count. It cannot be used for aggregations
+//     such as Sum since the function can only bound the number of contributions, not the value of the
+//     contributions.
+//  2. the key should be the privacy ID.
+//  3. the value should be the partition ID.
 func boundContributions(s beam.Scope, kvCol beam.PCollection, contributionLimit int64) beam.PCollection {
 	s = s.Scope("boundContributions")
 	// Transform the PCollection<K,V> into a PCollection<K,[]V>, where
@@ -299,6 +307,7 @@ type boundedSumInt64Fn struct {
 	MaxPartitionsContributed  int64
 	Lower                     int64
 	Upper                     int64
+	MaxContributions          int64
 	NoiseKind                 noise.Kind
 	noise                     noise.Noise // Set during Setup phase according to NoiseKind.
 	PublicPartitions          bool
@@ -319,6 +328,7 @@ func newBoundedSumInt64Fn(spec PrivacySpec, params SumParams, noiseKind noise.Ki
 		MaxPartitionsContributed:  params.MaxPartitionsContributed,
 		Lower:                     int64(params.MinValue),
 		Upper:                     int64(params.MaxValue),
+		MaxContributions:          params.maxContributions,
 		NoiseKind:                 noiseKind,
 		PublicPartitions:          publicPartitions,
 		TestMode:                  spec.testMode,
@@ -345,6 +355,7 @@ func (fn *boundedSumInt64Fn) CreateAccumulator() (boundedSumAccumInt64, error) {
 		MaxPartitionsContributed: fn.MaxPartitionsContributed,
 		Lower:                    fn.Lower,
 		Upper:                    fn.Upper,
+		MaxContributions:         fn.MaxContributions,
 		Noise:                    fn.noise,
 	})
 	if err != nil {
@@ -357,6 +368,7 @@ func (fn *boundedSumInt64Fn) CreateAccumulator() (boundedSumAccumInt64, error) {
 			Delta:                    fn.PartitionSelectionDelta,
 			PreThreshold:             fn.PreThreshold,
 			MaxPartitionsContributed: fn.MaxPartitionsContributed,
+			MaxContributions:         fn.MaxContributions,
 		})
 	}
 	return accum, err