cherry pick always charging privacy bucket mapper

src/main/kotlin/org/wfanet/measurement/eventdataprovider/eventfiltration/EventFilters.kt

@@ -44,7 +44,8 @@ object EventFilters {
    *   operation. If provided, [celExpr] is normalized to operative negation normal form by bubbling
    *   down all the negation operations to the leafs by applying De Morgan's laws recursively and by
    *   setting all the leaf comparison nodes (e.g. x == 47 ) that contain any field other than the
-   *   operative fields to true.
+   *   operative fields to true. If not provided or empty, the normalization operation will not be
+   *   performed.
    * @throws [EventFilterValidationException] if [celExpr] is not valid.
    */
   fun compileProgram(

src/main/kotlin/org/wfanet/measurement/eventdataprovider/privacybudgetmanagement/PrivacyBucketFilter.kt

@@ -17,7 +17,6 @@ import java.time.Clock
 import java.time.ZoneOffset
 import org.wfanet.measurement.common.OpenEndTimeRange
 import org.wfanet.measurement.common.rangeTo
-import org.wfanet.measurement.eventdataprovider.eventfiltration.EventFilters
 
 class PrivacyBucketFilter(
   private val privacyBucketMapper: PrivacyBucketMapper,
@@ -84,12 +83,7 @@ class PrivacyBucketFilter(
                   vidsIntervalStartPoint,
                   PrivacyLandscape.PRIVACY_BUCKET_VID_SAMPLE_WIDTH
                 )
-              if (
-                EventFilters.matches(
-                  privacyBucketMapper.toEventMessage(privacyBucketGroup),
-                  program
-                )
-              ) {
+              if (privacyBucketMapper.matches(privacyBucketGroup, program)) {
                 yield(privacyBucketGroup)
               }
             }

src/main/kotlin/org/wfanet/measurement/eventdataprovider/privacybudgetmanagement/PrivacyBucketMapper.kt

@@ -15,13 +15,31 @@ package org.wfanet.measurement.eventdataprovider.privacybudgetmanagement
 
 import com.google.protobuf.Message
 import org.projectnessie.cel.Program
+import org.wfanet.measurement.eventdataprovider.eventfiltration.EventFilters
 
 /** Maps Privacy bucket related objects to event filter related objects and vice versa. */
 interface PrivacyBucketMapper {
 
+  /**
+   * Fields in the cel expression that will not be altered by normalization. If left empty, it is
+   * assumed that all buckets will be charged.
+   */
+  val operativeFields: Set<String>
+
   /** Maps [filterExpression] to a [Program] by using privacy related fields and [Message] */
   fun toPrivacyFilterProgram(filterExpression: String): Program
 
   /** Maps [privacyBucketGroup] to an event [Message] */
   fun toEventMessage(privacyBucketGroup: PrivacyBucketGroup): Message
+
+  /**
+   * Returns whether a [privacyBucketGroup] matches for a given cel [program]. Always returns `true`
+   * if [operativeFields] are empty.
+   */
+  fun matches(privacyBucketGroup: PrivacyBucketGroup, program: Program): Boolean {
+    if (operativeFields.isEmpty()) {
+      return true
+    }
+    return EventFilters.matches(toEventMessage(privacyBucketGroup), program)
+  }
 }

src/main/kotlin/org/wfanet/measurement/eventdataprovider/privacybudgetmanagement/testing/AlwaysChargingPrivacyBucketMapper.kt

@@ -0,0 +1,44 @@
+/*
+ * Copyright 2023 The Cross-Media Measurement Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
+ * in compliance with the License. You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ */
+package org.wfanet.measurement.eventdataprovider.privacybudgetmanagement.testing
+
+import com.google.protobuf.Message
+import org.projectnessie.cel.Program
+import org.wfanet.measurement.api.v2alpha.event_templates.testing.TestEvent
+import org.wfanet.measurement.eventdataprovider.eventfiltration.EventFilters.compileProgram
+import org.wfanet.measurement.eventdataprovider.eventfiltration.validation.EventFilterValidationException
+import org.wfanet.measurement.eventdataprovider.privacybudgetmanagement.PrivacyBucketGroup
+import org.wfanet.measurement.eventdataprovider.privacybudgetmanagement.PrivacyBucketMapper
+import org.wfanet.measurement.eventdataprovider.privacybudgetmanagement.PrivacyBudgetManagerException
+import org.wfanet.measurement.eventdataprovider.privacybudgetmanagement.PrivacyBudgetManagerExceptionType
+
+/** [PrivacyBucketMapper] for [TestEvent] instances that charges all buckets. */
+class AlwaysChargingPrivacyBucketMapper : PrivacyBucketMapper {
+
+  override val operativeFields = emptySet<String>()
+
+  override fun toPrivacyFilterProgram(filterExpression: String): Program =
+    try {
+      compileProgram(TestEvent.getDescriptor(), "true == true", operativeFields)
+    } catch (e: EventFilterValidationException) {
+      throw PrivacyBudgetManagerException(
+        PrivacyBudgetManagerExceptionType.INVALID_PRIVACY_BUCKET_FILTER,
+        e
+      )
+    }
+
+  override fun toEventMessage(privacyBucketGroup: PrivacyBucketGroup): Message {
+    return TestEvent.getDefaultInstance()
+  }
+}

src/main/kotlin/org/wfanet/measurement/eventdataprovider/privacybudgetmanagement/testing/TestPrivacyBucketMapper.kt

@@ -30,13 +30,12 @@ import org.wfanet.measurement.eventdataprovider.privacybudgetmanagement.PrivacyB
 
 /** [PrivacyBucketMapper] for [TestEvent] instances. */
 class TestPrivacyBucketMapper : PrivacyBucketMapper {
+
+  override val operativeFields = setOf("person.age_group", "person.gender")
+
   override fun toPrivacyFilterProgram(filterExpression: String): Program =
     try {
-      compileProgram(
-        TestEvent.getDescriptor(),
-        filterExpression,
-        setOf("person.age_group", "person.gender")
-      )
+      compileProgram(TestEvent.getDescriptor(), filterExpression, operativeFields)
     } catch (e: EventFilterValidationException) {
       throw PrivacyBudgetManagerException(
         PrivacyBudgetManagerExceptionType.INVALID_PRIVACY_BUCKET_FILTER,

src/main/kotlin/org/wfanet/measurement/loadtest/config/PrivacyBudgets.kt

@@ -29,14 +29,12 @@ import org.wfanet.measurement.loadtest.config.LoadTestEventKt.privacy
 
 class TestPrivacyBucketMapper : PrivacyBucketMapper {
 
+  override val operativeFields = setOf("privacy.filterable")
+
   /** This mapper does not charge any bucket [filterExpression] is ignored. */
   override fun toPrivacyFilterProgram(filterExpression: String): Program =
     try {
-      compileProgram(
-        LoadTestEvent.getDescriptor(),
-        "privacy.filterable == true",
-        setOf("privacy.filterable")
-      )
+      compileProgram(LoadTestEvent.getDescriptor(), "privacy.filterable == true", operativeFields)
     } catch (e: EventFilterValidationException) {
       throw PrivacyBudgetManagerException(
         PrivacyBudgetManagerExceptionType.INVALID_PRIVACY_BUCKET_FILTER,