fix(filter_v2): measure view-typed payloads on pyarrow < 19

The payload size limit read RecordBatch.nbytes, which pyarrow < 19 does not
implement for view types (string_view, binary_view, list_view): a filter
carrying one failed with a raw ArrowTypeError instead of being parsed, or
rejected as FilterV2Error. Fall back to get_total_buffer_size(), which
handles every layout and can only overcount, so the limit still holds; if
neither works, raise FilterV2Error. Verified on pyarrow 18.1.0, where
nbytes raises on both view types and the fallback measures them.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

Author: rustyconover

tests/test_filter_v2.py

@@ -22,6 +22,7 @@
     FilterVersionError,
     RuntimeFilterAlgorithmCapability,
     deserialize_filters,
+    filter_v2,
 )
 
 _METADATA: dict[bytes, bytes] = {
@@ -767,3 +768,36 @@ def _connect(**_kw: object) -> _Connection:
         with pytest.raises(filter_v2.FilterV2Error, match=f"found {engine} {re.escape(version)}"):
             filter_v2._get_evaluation_connection(context)
         assert opened == []
+
+
+class _NoNbytes:
+    """Stands in for a view-typed batch under pyarrow < 19, where ``nbytes`` raises."""
+
+    def __init__(self, total: int | None) -> None:
+        self._total = total
+
+    @property
+    def nbytes(self) -> int:
+        raise pa.ArrowTypeError("not implemented for string_view")
+
+    def get_total_buffer_size(self) -> int:
+        if self._total is None:
+            raise pa.ArrowNotImplementedError("not implemented")
+        return self._total
+
+
+def test_payload_size_falls_back_when_nbytes_is_unavailable() -> None:
+    """A view-typed payload is measured by total buffer size instead of raising ArrowTypeError."""
+    assert filter_v2._payload_nbytes(_NoNbytes(123)) == 123  # type: ignore[arg-type]
+
+
+def test_payload_size_unmeasurable_is_a_filter_error() -> None:
+    """When no measurement works, the caller sees FilterV2Error, not a raw Arrow error."""
+    with pytest.raises(filter_v2.FilterV2Error, match="cannot measure filter payload size"):
+        filter_v2._payload_nbytes(_NoNbytes(None))  # type: ignore[arg-type]
+
+
+def test_payload_size_of_a_view_typed_batch() -> None:
+    """A real string_view payload is measured without error."""
+    batch = pa.record_batch([pa.array(["a", "bc"], type=pa.string_view())], names=["needles"])
+    assert filter_v2._payload_nbytes(batch) > 0

vgi/filter_v2.py

@@ -117,6 +117,34 @@ class FilterV2Error(ValueError):
     """A Filter Encoding v2 document is malformed or cannot be evaluated."""
 
 
+def _payload_nbytes(batch: pa.RecordBatch) -> int:
+    """Size of a filter payload batch, for the payload limit.
+
+    ``RecordBatch.nbytes`` is not implemented for view types (``string_view``,
+    ``binary_view``, ``list_view``) before pyarrow 19 and raises a raw
+    ``ArrowTypeError``. Fall back to the total buffer size there, which walks
+    every buffer generically and can only overcount (shared buffers), never
+    undercount, so the limit still holds.
+
+    Args:
+        batch: The filter payload batch.
+
+    Returns:
+        Its size in bytes, for comparison with ``MAX_PAYLOAD_BYTES``.
+
+    Raises:
+        FilterV2Error: If neither measurement is available for the batch.
+    """
+    try:
+        return int(batch.nbytes)
+    except (pa.ArrowTypeError, pa.ArrowNotImplementedError):
+        pass
+    try:
+        return int(batch.get_total_buffer_size())
+    except (pa.ArrowTypeError, pa.ArrowNotImplementedError) as exc:
+        raise FilterV2Error(f"cannot measure filter payload size: {exc}") from exc
+
+
 class PredicateMode(StrEnum):
     """Whether a predicate must be applied exactly or is only a pruning hint."""
 
@@ -592,7 +620,7 @@ def _validate_batch(self) -> EvaluationContext:
                 raise FilterV2Error(f"noncanonical payload field name {name!r}")
             if name.startswith("type_") and self.batch.column(i)[0].is_valid:
                 raise FilterV2Error(f"{name} must contain a NULL value")
-        if self.batch.nbytes > MAX_PAYLOAD_BYTES:
+        if _payload_nbytes(self.batch) > MAX_PAYLOAD_BYTES:
             raise FilterV2Error("filter payload exceeds 16 MiB")
         return parse_evaluation_context(self.batch.schema)