feat: literal value filter

google/cloud/bigtable/row_filters.py

@@ -481,20 +481,56 @@ def to_dict(self) -> dict[str, bytes]:
         return {"value_regex_filter": self.regex}
 
 
-class ExactValueFilter(ValueRegexFilter):
+class LiteralValueFilter(ValueRegexFilter):
     """Row filter for an exact value.
 
 
     :type value: bytes or str or int
     :param value:
-        a literal string encodable as ASCII, or the
-        equivalent bytes, or an integer (which will be packed into 8-bytes).
+        a literal string, integer, or the equivalent bytes.
+        Integer values will be packed into signed 8-bytes.
     """
 
     def __init__(self, value: bytes | str | int):
         if isinstance(value, int):
             value = _PACK_I64(value)
-        super(ExactValueFilter, self).__init__(value)
+        elif isinstance(value, str):
+            value = value.encode("utf-8")
+        value = self._write_literal_regex(value)
+        super(LiteralValueFilter, self).__init__(value)
+
+    @staticmethod
+    def _write_literal_regex(input_bytes: bytes) -> bytes:
+        """
+        Escape re2 special characters from literal bytes.
+
+        Extracted from: re2 QuoteMeta:
+        https://github.com/google/re2/blob/70f66454c255080a54a8da806c52d1f618707f8a/re2/re2.cc#L456
+        """
+        result = bytearray()
+        for byte in input_bytes:
+            # If this is the part of a UTF8 or Latin1 character, we need \
+            # to copy this byte without escaping.  Experimentally this is \
+            # what works correctly with the regexp library. \
+            utf8_latin1_check = (byte & 128) == 0
+            if (
+                (byte < ord("a") or byte > ord("z"))
+                and (byte < ord("A") or byte > ord("Z"))
+                and (byte < ord("0") or byte > ord("9"))
+                and byte != ord("_")
+                and utf8_latin1_check
+            ):
+                if byte == 0:
+                    # Special handling for null chars.
+                    # Note that this special handling is not strictly required for RE2,
+                    # but this quoting is required for other regexp libraries such as
+                    # PCRE.
+                    # Can't use "\\0" since the next character might be a digit.
+                    result.extend([ord("\\"), ord("x"), ord("0"), ord("0")])
+                    continue
+                result.append(ord(b"\\"))
+            result.append(byte)
+        return bytes(result)
 
     def __repr__(self) -> str:
         return f"{self.__class__.__name__}(value={self.regex!r})"

tests/system/test_system.py

@@ -160,6 +160,10 @@ def __init__(self, table):
     async def add_row(
         self, row_key, family=TEST_FAMILY, qualifier=b"q", value=b"test-value"
     ):
+        if isinstance(value, str):
+            value = value.encode("utf-8")
+        elif isinstance(value, int):
+            value = value.to_bytes(8, byteorder="big", signed=True)
         request = {
             "table_name": self.table.table_name,
             "row_key": row_key,
@@ -391,3 +395,49 @@ async def test_read_rows_stream_inactive_timer(table, temp_rows):
         await generator.__anext__()
         assert "inactivity" in str(e)
         assert "idle_timeout=0.1" in str(e)
+
+
+@retry.Retry(predicate=retry.if_exception_type(ClientError), initial=1, maximum=5)
+@pytest.mark.parametrize(
+    "cell_value,filter_input,expect_match",
+    [
+        (b"abc", b"abc", True),
+        (b"abc", "abc", True),
+        (b".", ".", True),
+        (".*", ".*", True),
+        (".*", b".*", True),
+        ("a", ".*", False),
+        (b".*", b".*", True),
+        (r"\a", r"\a", True),
+        (b"\xe2\x98\x83", "☃", True),
+        ("☃", "☃", True),
+        (r"\C☃", r"\C☃", True),
+        (1, 1, True),
+        (2, 1, False),
+        (68, 68, True),
+        ("D", 68, False),
+        (68, "D", False),
+        (-1, -1, True),
+        (2852126720, 2852126720, True),
+        (-1431655766, -1431655766, True),
+        (-1431655766, -1, False),
+    ],
+)
+@pytest.mark.asyncio
+async def test_literal_value_filter(
+    table, temp_rows, cell_value, filter_input, expect_match
+):
+    """
+    Literal value filter does complex escaping on re2 strings.
+    Make sure inputs are properly interpreted by the server
+    """
+    from google.cloud.bigtable.row_filters import LiteralValueFilter
+    from google.cloud.bigtable import ReadRowsQuery
+
+    f = LiteralValueFilter(filter_input)
+    await temp_rows.add_row(b"row_key_1", value=cell_value)
+    query = ReadRowsQuery(row_filter=f)
+    row_list = await table.read_rows(query)
+    assert len(row_list) == bool(
+        expect_match
+    ), f"row {type(cell_value)}({cell_value}) not found with {type(filter_input)}({filter_input}) filter"

tests/unit/test_row_filters.py

@@ -822,84 +822,90 @@ def test_value_regex_filter___repr__():
     assert eval(repr(row_filter)) == row_filter
 
 
-def test_exact_value_filter_to_pb_w_bytes():
-    from google.cloud.bigtable.row_filters import ExactValueFilter
+def test_literal_value_filter_to_pb_w_bytes():
+    from google.cloud.bigtable.row_filters import LiteralValueFilter
 
-    value = regex = b"value-regex"
-    row_filter = ExactValueFilter(value)
+    value = regex = b"value_regex"
+    row_filter = LiteralValueFilter(value)
     pb_val = row_filter._to_pb()
     expected_pb = _RowFilterPB(value_regex_filter=regex)
     assert pb_val == expected_pb
 
 
-def test_exact_value_filter_to_dict_w_bytes():
-    from google.cloud.bigtable.row_filters import ExactValueFilter
+def test_literal_value_filter_to_dict_w_bytes():
+    from google.cloud.bigtable.row_filters import LiteralValueFilter
     from google.cloud.bigtable_v2.types import data as data_v2_pb2
 
-    value = regex = b"value-regex"
-    row_filter = ExactValueFilter(value)
+    value = regex = b"value_regex"
+    row_filter = LiteralValueFilter(value)
     expected_dict = {"value_regex_filter": regex}
     assert row_filter.to_dict() == expected_dict
     expected_pb_value = row_filter._to_pb()
     assert data_v2_pb2.RowFilter(**expected_dict) == expected_pb_value
 
 
-def test_exact_value_filter_to_pb_w_str():
-    from google.cloud.bigtable.row_filters import ExactValueFilter
+def test_literal_value_filter_to_pb_w_str():
+    from google.cloud.bigtable.row_filters import LiteralValueFilter
 
-    value = "value-regex"
+    value = "value_regex"
     regex = value.encode("ascii")
-    row_filter = ExactValueFilter(value)
+    row_filter = LiteralValueFilter(value)
     pb_val = row_filter._to_pb()
     expected_pb = _RowFilterPB(value_regex_filter=regex)
     assert pb_val == expected_pb
 
 
-def test_exact_value_filter_to_dict_w_str():
-    from google.cloud.bigtable.row_filters import ExactValueFilter
+def test_literal_value_filter_to_dict_w_str():
+    from google.cloud.bigtable.row_filters import LiteralValueFilter
     from google.cloud.bigtable_v2.types import data as data_v2_pb2
 
-    value = "value-regex"
+    value = "value_regex"
     regex = value.encode("ascii")
-    row_filter = ExactValueFilter(value)
+    row_filter = LiteralValueFilter(value)
     expected_dict = {"value_regex_filter": regex}
     assert row_filter.to_dict() == expected_dict
     expected_pb_value = row_filter._to_pb()
     assert data_v2_pb2.RowFilter(**expected_dict) == expected_pb_value
 
 
-def test_exact_value_filter_to_pb_w_int():
-    import struct
-    from google.cloud.bigtable.row_filters import ExactValueFilter
+@pytest.mark.parametrize(
+    "value,expected_byte_string",
+    [
+        # null bytes are encoded as "\x00" in ascii characters
+        # others are just prefixed with "\"
+        (0, b"\\x00\\x00\\x00\\x00\\x00\\x00\\x00\\x00"),
+        (1, b"\\x00\\x00\\x00\\x00\\x00\\x00\\x00\\\x01"),
+        (
+            68,
+            b"\\x00\\x00\\x00\\x00\\x00\\x00\\x00D",
+        ),  # bytes that encode to alphanum are not escaped
+        (570, b"\\x00\\x00\\x00\\x00\\x00\\x00\\\x02\\\x3a"),
+        (2852126720, b"\\x00\\x00\\x00\\x00\xaa\\x00\\x00\\x00"),
+        (-1, b"\xff\xff\xff\xff\xff\xff\xff\xff"),
+        (-1096642724096, b"\xff\xff\xff\\x00\xaa\xff\xff\\x00"),
+    ],
+)
+def test_literal_value_filter_w_int(value, expected_byte_string):
+    from google.cloud.bigtable.row_filters import LiteralValueFilter
+    from google.cloud.bigtable_v2.types import data as data_v2_pb2
 
-    value = 1
-    regex = struct.Struct(">q").pack(value)
-    row_filter = ExactValueFilter(value)
+    row_filter = LiteralValueFilter(value)
+    # test pb
     pb_val = row_filter._to_pb()
-    expected_pb = _RowFilterPB(value_regex_filter=regex)
+    expected_pb = _RowFilterPB(value_regex_filter=expected_byte_string)
     assert pb_val == expected_pb
-
-
-def test_exact_value_filter_to_dict_w_int():
-    import struct
-    from google.cloud.bigtable.row_filters import ExactValueFilter
-    from google.cloud.bigtable_v2.types import data as data_v2_pb2
-
-    value = 1
-    regex = struct.Struct(">q").pack(value)
-    row_filter = ExactValueFilter(value)
-    expected_dict = {"value_regex_filter": regex}
+    # test dict
+    expected_dict = {"value_regex_filter": expected_byte_string}
     assert row_filter.to_dict() == expected_dict
-    expected_pb_value = row_filter._to_pb()
-    assert data_v2_pb2.RowFilter(**expected_dict) == expected_pb_value
+    assert data_v2_pb2.RowFilter(**expected_dict) == pb_val
 
 
-def test_exact_value_filter___repr__():
-    from google.cloud.bigtable.row_filters import ExactValueFilter
+def test_literal_value_filter___repr__():
+    from google.cloud.bigtable.row_filters import LiteralValueFilter
 
-    value = "value-regex"
-    row_filter = ExactValueFilter(value)
-    expected = "ExactValueFilter(value=b'value-regex')"
+    value = "value_regex"
+    row_filter = LiteralValueFilter(value)
+    expected = "LiteralValueFilter(value=b'value_regex')"
     assert repr(row_filter) == expected
     assert repr(row_filter) == str(row_filter)
     assert eval(repr(row_filter)) == row_filter
@@ -1907,6 +1913,30 @@ def test_conditional_row_filter___str__():
     assert str(row_filter4) == expected
 
 
+@pytest.mark.parametrize(
+    "input_arg, expected_bytes",
+    [
+        (b"abc", b"abc"),
+        ("abc", b"abc"),
+        (1, b"\\x00\\x00\\x00\\x00\\x00\\x00\\x00\\\x01"),  # null bytes are ascii
+        (b"*", b"\\*"),
+        (".", b"\\."),
+        (b"\\", b"\\\\"),
+        (b"h.*i", b"h\\.\\*i"),
+        (b'""', b'\\"\\"'),
+        (b"[xyz]", b"\\[xyz\\]"),
+        (b"\xe2\x98\xba\xef\xb8\x8f", b"\xe2\x98\xba\xef\xb8\x8f"),
+        ("☃", b"\xe2\x98\x83"),
+        (r"\C☃", b"\\\\C\xe2\x98\x83"),
+    ],
+)
+def test_literal_value__write_literal_regex(input_arg, expected_bytes):
+    from google.cloud.bigtable.row_filters import LiteralValueFilter
+
+    filter_ = LiteralValueFilter(input_arg)
+    assert filter_.regex == expected_bytes
+
+
 def _ColumnRangePB(*args, **kw):
     from google.cloud.bigtable_v2.types import data as data_v2_pb2
 
@@ -1955,15 +1985,15 @@ def _get_regex_filters():
         FamilyNameRegexFilter,
         ColumnQualifierRegexFilter,
         ValueRegexFilter,
-        ExactValueFilter,
+        LiteralValueFilter,
     )
 
     return [
         RowKeyRegexFilter,
         FamilyNameRegexFilter,
         ColumnQualifierRegexFilter,
         ValueRegexFilter,
-        ExactValueFilter,
+        LiteralValueFilter,
     ]