feat: Add ValueBitmask to the filter DSL (#16981)

Author: steveniemitz

packages/google-cloud-bigtable/google/cloud/bigtable/row_filters.py

@@ -429,6 +429,39 @@ def to_pb(self):
         return data_v2_pb2.RowFilter(column_range_filter=column_range)
 
 
+class ValueBitmaskFilter(RowFilter):
+    """Row filter for a value bitmask.
+
+    Matches only cells with values that satisfy the condition
+    ``(value & mask) == mask``. The mask length must exactly match the value
+    length, otherwise the cell is not considered a match.
+
+    :type mask: bytes or str
+    :param mask: A bitmask to match against cells with values. String values
+                 will be encoded as ASCII.
+    """
+
+    def __init__(self, mask):
+        self.mask = _to_bytes(mask)
+
+    def __eq__(self, other):
+        if not isinstance(other, self.__class__):
+            return NotImplemented
+        return other.mask == self.mask
+
+    def __ne__(self, other):
+        return not self == other
+
+    def to_pb(self):
+        """Converts the row filter to a protobuf.
+
+        :rtype: :class:`.data_v2_pb2.RowFilter`
+        :returns: The converted current object.
+        """
+        value_bitmask = data_v2_pb2.ValueBitmask(mask=self.mask)
+        return data_v2_pb2.RowFilter(value_bitmask_filter=value_bitmask)
+
+
 class ValueRegexFilter(_RegexFilter):
     """Row filter for a value regular expression.
 

packages/google-cloud-bigtable/tests/unit/v2_client/test_row_filters.py

@@ -535,6 +535,27 @@ def test_value_regex_filter_to_pb_w_str():
     assert pb_val == expected_pb
 
 
+def test_value_bitmask_filter_to_pb_w_bytes():
+    from google.cloud.bigtable.row_filters import ValueBitmaskFilter
+
+    mask = b"value-mask"
+    row_filter = ValueBitmaskFilter(mask)
+    pb_val = row_filter.to_pb()
+    expected_pb = _RowFilterPB(value_bitmask_filter=_ValueBitmaskPB(mask=mask))
+    assert pb_val == expected_pb
+
+
+def test_value_bitmask_filter_to_pb_w_str():
+    from google.cloud.bigtable.row_filters import ValueBitmaskFilter
+
+    mask = "value-mask"
+    mask_bytes = mask.encode("ascii")
+    row_filter = ValueBitmaskFilter(mask)
+    pb_val = row_filter.to_pb()
+    expected_pb = _RowFilterPB(value_bitmask_filter=_ValueBitmaskPB(mask=mask_bytes))
+    assert pb_val == expected_pb
+
+
 def test_exact_value_filter_to_pb_w_bytes():
     from google.cloud.bigtable.row_filters import ExactValueFilter
 
@@ -1173,3 +1194,9 @@ def _ValueRangePB(*args, **kw):
     from google.cloud.bigtable_v2.types import data as data_v2_pb2
 
     return data_v2_pb2.ValueRange(*args, **kw)
+
+
+def _ValueBitmaskPB(*args, **kw):
+    from google.cloud.bigtable_v2.types import data as data_v2_pb2
+
+    return data_v2_pb2.ValueBitmask(*args, **kw)