This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 1c5743f632 [python] Add scan().read_blobs / stream_blobs for
concurrent blob reads (#8439)
1c5743f632 is described below
commit 1c5743f632aa3821ccde515a6845ec027c1d18c3
Author: XiaoHongbo <[email protected]>
AuthorDate: Fri Jul 3 21:47:59 2026 +0800
[python] Add scan().read_blobs / stream_blobs for concurrent blob reads
(#8439)
---
docs/docs/pypaimon/multimodal-api.mdx | 51 ++++
paimon-python/pypaimon/multimodal/query.py | 172 ++++++++++++-
.../pypaimon/tests/multimodal_table_test.py | 282 +++++++++++++++++++++
3 files changed, 504 insertions(+), 1 deletion(-)
diff --git a/docs/docs/pypaimon/multimodal-api.mdx
b/docs/docs/pypaimon/multimodal-api.mdx
index ebd1427367..b2b63ab19f 100644
--- a/docs/docs/pypaimon/multimodal-api.mdx
+++ b/docs/docs/pypaimon/multimodal-api.mdx
@@ -332,6 +332,57 @@ result = (
)
```
+### Reading BLOB columns
+
+`scan().read_blobs(column)` bulk-fetches a BLOB column's bytes for the filtered
+rows using concurrent, same-file coalesced ranged reads. This is much faster
than
+a per-row loop and avoids the slow row-by-row blob resolution on data-evolution
+tables. It returns `(scalar_table, {column: [bytes | None]})`, row-aligned; the
+scalar table drops all BLOB columns.
+
+```python
+scalar, blobs = (
+ docs.scan()
+ .where("category = 'lake'")
+ .read_blobs("image", parallelism=32)
+)
+images = blobs["image"] # list[bytes | None], aligned with scalar rows
+```
+
+`where()` takes SQL-like strings only, so filter by a list of keys with `IN
(...)`:
+
+```python
+ids = ["a", "b", "c"]
+in_clause = ", ".join(f"'{i}'" for i in ids)
+scalar, blobs = docs.scan().where(f"id IN ({in_clause})").read_blobs("image")
+```
+
+`where()` is a SQL string, so build the `IN (...)` clause only from trusted,
+already-escaped ids -- do not interpolate untrusted external input.
+
+Read several BLOB columns at once (fetched together, not column-by-column):
+
+```python
+scalar, blobs = docs.scan().where("category = 'lake'").read_blobs(["image",
"audio"])
+```
+
+For a memory-bounded read (e.g. streaming into a trainer), `stream_blobs`
yields
+one batch at a time, so peak memory is a single batch rather than the whole
result:
+
+```python
+for scalar, blobs in docs.scan().where("category =
'lake'").stream_blobs("image"):
+ for img in blobs["image"]:
+ ... # feed the trainer, then drop the batch
+```
+
+Notes:
+
+- `parallelism` ~16–32 is the sweet spot over the public network; go higher on
an
+ internal/VPC endpoint.
+- Fastest when the matching rows are contiguous (bulk/range reads coalesce
into a
+ few large reads); scattered point reads coalesce less.
+- Blob reads are available only on `scan()`, not on the `search()` queries.
+
## Row IDs
Multimodal tables enable `row-tracking.enabled` by default, so each row has a
diff --git a/paimon-python/pypaimon/multimodal/query.py
b/paimon-python/pypaimon/multimodal/query.py
index 86012b9608..0ea45b7df5 100644
--- a/paimon-python/pypaimon/multimodal/query.py
+++ b/paimon-python/pypaimon/multimodal/query.py
@@ -15,7 +15,9 @@
# specific language governing permissions and limitations
# under the License.
-from typing import Callable, List, Optional
+from typing import Callable, Dict, List, Optional, Tuple
+
+import pyarrow as pa
from pypaimon.common.where_parser import parse_where_clause
from pypaimon.table.special_fields import SpecialFields
@@ -98,6 +100,167 @@ class ScanQuery:
def to_list(self) -> List[dict]:
return self.to_arrow().to_pylist()
+ def read_blobs(
+ self, columns=None, *, parallelism: int = 64
+ ) -> Tuple[pa.Table, Dict[str, List[Optional[bytes]]]]:
+ """Materialise BLOB column(s) for the filtered rows with concurrent,
+ coalesced ranged reads. Reads via blob-as-descriptor to skip the slow
+ row-by-row blob resolution on multi-group data-evolution splits.
+
+ ``columns`` picks the BLOB column(s) (default: all, intersected with
+ ``select(...)``). Returns ``(scalar_arrow_table, {column:
[bytes|None]})``,
+ row-aligned. Use :meth:`stream_blobs` for a memory-bounded read.
+
+ Unresolved blob-view columns are not supported and raise
``ValueError``.
+ """
+ blob_cols = self._resolve_blob_columns(columns)
+ read_builder, file_io = self._blob_descriptor_read_builder(blob_cols)
+ arrow = read_builder.new_read().to_arrow(
+ read_builder.new_scan().plan().splits())
+ bodies = self._fetch_bodies(
+ file_io, arrow.select(blob_cols).to_pydict(), blob_cols,
parallelism)
+ scalar = arrow.select(self._scalar_columns(arrow.column_names))
+ return scalar, bodies
+
+ def stream_blobs(self, columns=None, *, parallelism: int = 64):
+ """Memory-bounded streaming variant of :meth:`read_blobs`: yield
+ ``(scalar_batch, {column: [bytes|None]})`` per Arrow batch, so peak
memory
+ is one batch rather than the whole result.
+ """
+ # Validate eagerly so a bad column raises here, not on the first
next().
+ blob_cols = self._resolve_blob_columns(columns)
+ read_builder, file_io = self._blob_descriptor_read_builder(blob_cols)
+ return self._iter_blobs(read_builder, file_io, blob_cols, parallelism)
+
+ def _iter_blobs(self, read_builder, file_io, blob_cols, parallelism):
+ reader = read_builder.new_read().to_arrow_batch_reader(
+ read_builder.new_scan().plan().splits())
+ try:
+ for batch in reader:
+ bodies = self._fetch_bodies(
+ file_io, batch.select(blob_cols).to_pydict(), blob_cols,
parallelism)
+ scalar = batch.select(self._scalar_columns(batch.schema.names))
+ yield scalar, bodies
+ finally:
+ # Close the reader even if the caller breaks out early.
+ if hasattr(reader, "close"):
+ reader.close()
+
+ def _blob_descriptor_read_builder(self, blob_cols: List[str]):
+ """Blob-as-descriptor read builder with this query's
filter/projection/limit;
+ returns ``(read_builder, file_io)``."""
+ from pypaimon.common.options.core_options import CoreOptions
+ read_table = self._table.copy({
+ CoreOptions.BLOB_AS_DESCRIPTOR.key(): "true"
+ })
+ read_builder = read_table.new_read_builder()
+ if self._predicate is not None:
+ read_builder = read_builder.with_filter(self._predicate)
+ projection = self._blob_read_projection(blob_cols)
+ if projection is not None:
+ read_builder = read_builder.with_projection(projection)
+ if self._limit is not None:
+ read_builder = read_builder.with_limit(self._limit)
+ return read_builder, read_table.file_io
+
+ @staticmethod
+ def _fetch_bodies(file_io, data, blob_cols, parallelism):
+ # Decode each descriptor to a (uri, offset, length) range and read
them all in
+ # one coalesced pass on ``file_io`` -- the read table's FileIO, which
already
+ # carries the merged DLF/OSS token. Going through Blob.from_bytes here
would
+ # build a BlobRef whose uri_reader re-resolves the URI via
+ # ``FileIO.get(uri, catalog_options)`` off the raw options (no merged
token),
+ # failing with "endpoint should be non-empty" / "Init credential
failed" unless
+ # the caller also passes fs.oss.* -- which users should not have to.
+ from pypaimon.table.row.blob import BlobDescriptor, BlobViewStruct
+ ranges = []
+ inline = {} # cell index -> blob stored inline (returned as-is, no
ranged read)
+ i = 0
+ for col in blob_cols:
+ for value in data[col]:
+ if value is None:
+ ranges.append(None)
+ else:
+ raw = bytes(value)
+ if BlobViewStruct.is_blob_view_struct(raw):
+ raise ValueError(
+ "read_blobs does not support unresolved blob-view
columns; "
+ "read such a column on its own, or enable
blob-view resolution.")
+ if BlobDescriptor.is_blob_descriptor(raw):
+ d = BlobDescriptor.deserialize(raw)
+ ranges.append((d.uri, d.offset, d.length))
+ else: # blob stored inline: the bytes are the payload
+ ranges.append(None)
+ inline[i] = raw
+ i += 1
+ fetched = file_io.read_ranges_coalesced(ranges, parallelism)
+ for idx, raw in inline.items():
+ fetched[idx] = raw
+ bodies = {}
+ offset = 0
+ for col in blob_cols:
+ n = len(data[col])
+ bodies[col] = fetched[offset:offset + n]
+ offset += n
+ return bodies
+
+ def _all_blob_columns(self) -> List[str]:
+ return [
+ field.name for field in self._table.fields
+ if getattr(field.type, "type", None) == "BLOB"
+ ]
+
+ def _resolve_blob_columns(self, columns) -> List[str]:
+ all_blob = self._all_blob_columns()
+ if columns is None:
+ selected = all_blob
+ if self._projection is not None:
+ selected = [c for c in all_blob if c in self._projection]
+ if not selected:
+ raise ValueError("No BLOB column to read; pass columns=.")
+ return selected
+ if isinstance(columns, str):
+ columns = [columns]
+ columns = list(dict.fromkeys(columns)) # de-dup, keep order
+ for name in columns:
+ if name not in all_blob:
+ raise ValueError(f"Column {name!r} is not a BLOB column.")
+ return columns
+
+ def _blob_read_projection(self, blob_cols: List[str]) -> List[str]:
+ # Base on _effective_projection() (honours with_row_id), drop BLOB
+ # descriptors, then append the requested BLOB columns and every
predicate
+ # column -- including predicate columns that are themselves BLOB (read
as
+ # descriptor) -- so SplitRead keeps the row-level filter for where().
+ blob_set = set(self._all_blob_columns())
+ effective = self._effective_projection()
+ if effective is None:
+ base = [f.name for f in self._table.fields if f.name not in
blob_set]
+ else:
+ base = [name for name in effective if name not in blob_set]
+ for name in self._predicate_fields():
+ if name not in base and name not in blob_cols:
+ base.append(name)
+ return base + list(blob_cols)
+
+ def _scalar_columns(self, available) -> List[str]:
+ # Non-BLOB columns to expose, based on _effective_projection() so
+ # with_row_id() is honoured; drop BLOBs and predicate-only helpers, and
+ # skip unknown projected names to match to_arrow()'s silent drop.
+ blob_set = set(self._all_blob_columns())
+ effective = self._effective_projection()
+ if effective is None:
+ return [name for name in available if name not in blob_set]
+ available = set(available)
+ return [name for name in effective
+ if name in available and name not in blob_set]
+
+ def _predicate_fields(self):
+ if self._predicate is None:
+ return ()
+ from pypaimon.read.push_down_utils import _get_all_fields
+ return _get_all_fields(self._predicate)
+
def _coerce_predicate(self, predicate, method):
if predicate is None:
return None
@@ -130,6 +293,13 @@ class _PreFilterQuery(ScanQuery):
self._pre_filter = self._and_predicate(self._pre_filter, predicate)
return self
+ # Blob reads bypass the search result_factory, so reject them (else a
plain scan).
+ def read_blobs(self, *args, **kwargs):
+ raise TypeError("read_blobs is only supported on scan(), not search
queries.")
+
+ def stream_blobs(self, *args, **kwargs):
+ raise TypeError("stream_blobs is only supported on scan(), not search
queries.")
+
class VectorQuery(_PreFilterQuery):
"""Chainable query wrapper for vector global-index search."""
diff --git a/paimon-python/pypaimon/tests/multimodal_table_test.py
b/paimon-python/pypaimon/tests/multimodal_table_test.py
index f880c888e0..d6ed03c969 100644
--- a/paimon-python/pypaimon/tests/multimodal_table_test.py
+++ b/paimon-python/pypaimon/tests/multimodal_table_test.py
@@ -749,6 +749,288 @@ class MultimodalTableTest(unittest.TestCase):
rows,
)
+ def test_scan_read_blobs(self):
+ obs = self.conn.create_table(
+ "obs",
+ schema=_schema({
+ "clip": pa.string(),
+ "idx": pa.int32(),
+ "image": pa.large_binary(),
+ }),
+ options=_PARQUET_OPTIONS,
+ partitioned=["clip"],
+ )
+ payloads = [("blob-%d-" % i).encode() * (i + 1) for i in range(6)]
+ obs.add([
+ {"clip": "c1", "idx": 0, "image": payloads[0]},
+ {"clip": "c1", "idx": 1, "image": payloads[1]},
+ {"clip": "c1", "idx": 2, "image": payloads[2]},
+ {"clip": "c2", "idx": 0, "image": payloads[3]},
+ {"clip": "c2", "idx": 1, "image": payloads[4]},
+ {"clip": "c3", "idx": 0, "image": None},
+ ])
+
+ # scalar table is row-aligned with the blob list and drops the blob
column
+ scalar, blobs = obs.scan().where("clip = 'c1'").read_blobs("image")
+ images = blobs["image"]
+ self.assertEqual(3, len(images))
+ self.assertNotIn("image", scalar.column_names)
+ got = dict(zip(scalar.column("idx").to_pylist(), images))
+ self.assertEqual({0: payloads[0], 1: payloads[1], 2: payloads[2]}, got)
+
+ # blob column auto-detected when columns=None
+ _, blobs2 = obs.scan().where("clip = 'c2'").read_blobs()
+ self.assertEqual({payloads[3], payloads[4]}, set(blobs2["image"]))
+
+ # null blob -> None
+ _, blobs3 = obs.scan().where("clip = 'c3'").read_blobs("image")
+ self.assertEqual([None], blobs3["image"])
+
+ # non-BLOB column is rejected
+ with self.assertRaisesRegex(ValueError, "not a BLOB column"):
+ obs.scan().read_blobs("idx")
+
+ # empty result: 0 matching rows -> empty blob list, schema preserved
+ scalar_e, blobs_e = obs.scan().where("clip =
'none'").read_blobs("image")
+ self.assertEqual(0, scalar_e.num_rows)
+ self.assertEqual([], blobs_e["image"])
+
+ def test_scan_read_blobs_multi_column(self):
+ obs = self.conn.create_table(
+ "multi",
+ schema=_schema({
+ "clip": pa.string(),
+ "idx": pa.int32(),
+ "img": pa.large_binary(),
+ "aud": pa.large_binary(),
+ }),
+ options=_PARQUET_OPTIONS,
+ partitioned=["clip"],
+ )
+ obs.add([
+ {"clip": "c1", "idx": i,
+ "img": ("img-%d" % i).encode(), "aud": ("aud-%d" % i).encode()}
+ for i in range(4)
+ ])
+
+ # both BLOB columns fetched in one call, each row-aligned with the
scalars
+ scalar, blobs = obs.scan().where("clip = 'c1'").read_blobs(["img",
"aud"])
+ self.assertEqual({"img", "aud"}, set(blobs))
+ self.assertNotIn("img", scalar.column_names)
+ self.assertNotIn("aud", scalar.column_names)
+ idx = scalar.column("idx").to_pylist()
+ self.assertEqual({i: ("img-%d" % i).encode() for i in range(4)},
+ dict(zip(idx, blobs["img"])))
+ self.assertEqual({i: ("aud-%d" % i).encode() for i in range(4)},
+ dict(zip(idx, blobs["aud"])))
+
+ # reading only one BLOB column must not leak the other into scalar as
+ # descriptor bytes
+ scalar1, blobs1 = obs.scan().where("clip = 'c1'").read_blobs("img")
+ self.assertEqual({"img"}, set(blobs1))
+ self.assertEqual(["clip", "idx"], scalar1.column_names)
+
+ # columns=None intersects all BLOBs with select() -> only projected
BLOB
+ _, blobs2 = obs.scan().where("clip = 'c1'").select(["clip", "idx",
"img"]).read_blobs()
+ self.assertEqual({"img"}, set(blobs2))
+
+ # duplicate columns are de-duplicated
+ _, blobs3 = obs.scan().where("clip = 'c1'").read_blobs(["img", "img"])
+ self.assertEqual({"img"}, set(blobs3))
+
+ def test_scan_read_blobs_filter_column_not_selected(self):
+ # The row filter must apply even when its column is not in select().
+ obs = self.conn.create_table(
+ "obs",
+ schema=_schema({
+ "clip": pa.string(),
+ "idx": pa.int32(),
+ "image": pa.large_binary(),
+ }),
+ options=_PARQUET_OPTIONS,
+ partitioned=["clip"],
+ )
+ obs.add([
+ {"clip": "c1", "idx": i, "image": ("v-%d" % i).encode()}
+ for i in range(4)
+ ])
+
+ scalar, blobs = (
+ obs.scan().select(["image"]).where("idx = 1").read_blobs("image"))
+ self.assertEqual([b"v-1"], blobs["image"])
+ self.assertNotIn("idx", scalar.schema.names)
+
+ scalar2, blobs2 = (
+ obs.scan().select(["idx", "image"]).where("idx =
2").read_blobs("image"))
+ self.assertEqual([b"v-2"], blobs2["image"])
+ self.assertEqual([2], scalar2.column("idx").to_pylist())
+
+ streamed = []
+ for _, body in (
+ obs.scan().select(["image"]).where("idx =
1").stream_blobs("image")):
+ streamed.extend(body["image"])
+ self.assertEqual([b"v-1"], streamed)
+
+ scalar3, _ = obs.scan().select(["missing",
"image"]).read_blobs("image")
+ self.assertNotIn("missing", scalar3.schema.names)
+
+ def test_scan_read_blobs_with_row_id(self):
+ # with_row_id() must expose _ROW_ID in the scalar table, like
to_arrow().
+ obs = self.conn.create_table(
+ "obs",
+ schema=_schema({
+ "clip": pa.string(),
+ "idx": pa.int32(),
+ "image": pa.large_binary(),
+ }),
+ options=_PARQUET_OPTIONS,
+ partitioned=["clip"],
+ )
+ obs.add([
+ {"clip": "c1", "idx": i, "image": ("v-%d" % i).encode()}
+ for i in range(3)
+ ])
+ row_id = "_ROW_ID"
+
+ expected = {r["idx"]: r[row_id]
+ for r in obs.scan().with_row_id().to_arrow().to_pylist()}
+
+ scalar, blobs = (
+ obs.scan().with_row_id().select(["idx",
"image"]).read_blobs("image"))
+ self.assertIn(row_id, scalar.schema.names)
+ self.assertEqual(expected, dict(zip(scalar.column("idx").to_pylist(),
+
scalar.column(row_id).to_pylist())))
+
+ got = {}
+ for sb, _ in obs.scan().with_row_id().select(["idx",
"image"]).stream_blobs("image"):
+ self.assertIn(row_id, sb.schema.names)
+ got.update(zip(sb.column("idx").to_pylist(),
sb.column(row_id).to_pylist()))
+ self.assertEqual(expected, got)
+
+ def test_scan_read_blobs_where_on_blob_column(self):
+ # A where() on one BLOB column must still filter when a different BLOB
is
+ # read (the predicate BLOB column is read as descriptor for filtering).
+ obs = self.conn.create_table(
+ "obs",
+ schema=_schema({
+ "clip": pa.string(),
+ "idx": pa.int32(),
+ "imga": pa.large_binary(),
+ "imgb": pa.large_binary(),
+ }),
+ options=_PARQUET_OPTIONS,
+ partitioned=["clip"],
+ )
+ obs.add([
+ {"clip": "c1", "idx": 0, "imga": b"a0", "imgb": b"b0"},
+ {"clip": "c1", "idx": 1, "imga": None, "imgb": b"b1"},
+ {"clip": "c1", "idx": 2, "imga": b"a2", "imgb": b"b2"},
+ ])
+
+ scalar, blobs = obs.scan().where("imga IS NOT NULL").read_blobs("imgb")
+ self.assertEqual([b"b0", b"b2"], blobs["imgb"])
+ self.assertNotIn("imga", scalar.schema.names)
+
+ def test_search_query_rejects_blob_reads(self):
+ t = self.conn.create_table(
+ "srch",
+ schema=_schema({
+ "id": pa.int32(),
+ "emb": _vector(3),
+ "img": pa.large_binary(),
+ }),
+ options=_PARQUET_OPTIONS,
+ )
+ with self.assertRaisesRegex(TypeError, "only supported on scan"):
+ t.search([1.0, 0.0, 0.0], column="emb").read_blobs("img")
+ with self.assertRaisesRegex(TypeError, "only supported on scan"):
+ t.search([1.0, 0.0, 0.0], column="emb").stream_blobs("img")
+
+ def test_scan_stream_blobs(self):
+ obs = self.conn.create_table(
+ "obs",
+ schema=_schema({
+ "clip": pa.string(),
+ "idx": pa.int32(),
+ "image": pa.large_binary(),
+ }),
+ options=_PARQUET_OPTIONS,
+ partitioned=["clip"],
+ )
+ payloads = {i: ("s-%d-" % i).encode() * (i + 1) for i in range(5)}
+ obs.add([
+ {"clip": "c1", "idx": i, "image": payloads[i]} for i in range(5)
+ ])
+
+ # collecting every streamed batch must reproduce the full, row-aligned
result
+ got = {}
+ batches = 0
+ for scalar, blobs in obs.scan().where("clip =
'c1'").stream_blobs("image"):
+ batches += 1
+ self.assertNotIn("image", scalar.schema.names)
+ for idx, img in zip(scalar.column("idx").to_pylist(),
blobs["image"]):
+ got[idx] = img
+ self.assertGreaterEqual(batches, 1)
+ self.assertEqual({i: payloads[i] for i in range(5)}, got)
+
+ # bad column is rejected eagerly (not deferred to first iteration)
+ with self.assertRaisesRegex(ValueError, "not a BLOB column"):
+ obs.scan().stream_blobs("idx")
+ # breaking out early closes the reader without error
+ it = obs.scan().where("clip = 'c1'").stream_blobs("image")
+ next(it)
+ it.close()
+ # empty result streams nothing
+ self.assertEqual(
+ [], list(obs.scan().where("clip = 'none'").stream_blobs("image")))
+
+ def test_fetch_bodies_decodes_descriptor_inline_and_null(self):
+ # Cells may be descriptor bytes (incl. -1 read-to-EOF), inline bytes,
or null.
+ from pypaimon.multimodal.query import ScanQuery
+ from pypaimon.common.file_io import FileIO
+ from pypaimon.table.row.blob import BlobDescriptor
+ data = bytes(range(64))
+ path = os.path.join(self.temp_dir, "blob.bin")
+ with open(path, "wb") as f:
+ f.write(data)
+ file_io = FileIO.get("file://" + self.temp_dir, {})
+ cells = [
+ BlobDescriptor(path, 0, -1).serialize(),
+ BlobDescriptor(path, 10, 8).serialize(),
+ b"raw-inline-bytes",
+ None,
+ ]
+ bodies = ScanQuery._fetch_bodies(
+ file_io, {"image": cells}, ["image"], 4)
+ self.assertEqual(
+ [data, data[10:18], b"raw-inline-bytes", None], bodies["image"])
+
+ def test_fetch_bodies_reads_via_file_io_not_uri_reader(self):
+ # Blobs must be read through file_io.read_ranges_coalesced (which
carries the
+ # resolved DLF/OSS token), never via uri_reader_factory -- that would
rebuild a
+ # FileIO from the raw catalog options and fail in DLF data-token mode.
+ from pypaimon.multimodal.query import ScanQuery
+ from pypaimon.table.row.blob import BlobDescriptor
+
+ class _FakeIO:
+ @property
+ def uri_reader_factory(self):
+ raise AssertionError("_fetch_bodies must not use
uri_reader_factory")
+
+ def read_ranges_coalesced(self, ranges, parallelism):
+ return [None if r is None else b"BODY:" + r[0].encode() for r
in ranges]
+
+ cells = [BlobDescriptor("oss://bucket/x", 4, 10).serialize(),
b"inline-blob", None]
+ bodies = ScanQuery._fetch_bodies(_FakeIO(), {"img": cells}, ["img"], 8)
+ self.assertEqual([b"BODY:oss://bucket/x", b"inline-blob", None],
bodies["img"])
+
+ def test_fetch_bodies_rejects_unresolved_blob_view(self):
+ from pypaimon.multimodal.query import ScanQuery
+ from pypaimon.table.row.blob import BlobViewStruct
+ view_bytes = BlobViewStruct("db.tbl", 1, 2).serialize()
+ with self.assertRaisesRegex(ValueError, "blob-view"):
+ ScanQuery._fetch_bodies(None, {"image": [view_bytes]}, ["image"],
4)
+
def test_scan_does_not_expose_pre_filter(self):
users = self.conn.create_table(
"users",
