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 d9c861f1a1 [python] Support MAP<K, BLOB> in multimodal BLOB reads
(#9428)
d9c861f1a1 is described below
commit d9c861f1a1e0478740a23d63362af8031fe13430
Author: XiaoHongbo <[email protected]>
AuthorDate: Fri Aug 28 15:18:53 2026 +0800
[python] Support MAP<K, BLOB> in multimodal BLOB reads (#9428)
---
docs/docs/pypaimon/multimodal-api.mdx | 12 ++-
paimon-python/pypaimon/multimodal/blob_read.py | 98 ++++++++++++++++------
paimon-python/pypaimon/multimodal/query.py | 43 ++++++----
.../pypaimon/tests/multimodal_table_test.py | 76 +++++++++++++++++
4 files changed, 183 insertions(+), 46 deletions(-)
diff --git a/docs/docs/pypaimon/multimodal-api.mdx
b/docs/docs/pypaimon/multimodal-api.mdx
index 995966a77c..ad4cc7604a 100644
--- a/docs/docs/pypaimon/multimodal-api.mdx
+++ b/docs/docs/pypaimon/multimodal-api.mdx
@@ -463,8 +463,9 @@ result = (
`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.
+tables. It returns `(scalar_table, {column: rows})`, row-aligned. Scalar BLOB
+rows are `bytes | None`; `MAP<K, BLOB>` rows are `None` or ordered lists of
+`(key, bytes | None)` pairs. The scalar table drops the readable BLOB columns.
```python
scalar, blobs = (
@@ -486,10 +487,13 @@ 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):
+Read several BLOB columns at once, including `MAP<K, BLOB>`:
```python
-scalar, blobs = docs.scan().where("category = 'lake'").read_blobs(["image",
"audio"])
+scalar, blobs = docs.scan().where("category = 'lake'").read_blobs(
+ ["image", "audio", "renditions"]
+)
+renditions = [None if row is None else dict(row) for row in
blobs["renditions"]]
```
For a memory-bounded read (e.g. streaming into a trainer), `stream_blobs`
yields
diff --git a/paimon-python/pypaimon/multimodal/blob_read.py
b/paimon-python/pypaimon/multimodal/blob_read.py
index 2db4621b19..ec6930453c 100644
--- a/paimon-python/pypaimon/multimodal/blob_read.py
+++ b/paimon-python/pypaimon/multimodal/blob_read.py
@@ -18,44 +18,88 @@
"""Shared helpers for materialising multimodal BLOB descriptor columns."""
-def fetch_blob_bodies(file_io, data, blob_cols, parallelism):
- """Fetch BLOB payload bytes for descriptor/inline/null cells.
+def fetch_blob_bodies(
+ file_io, data, blob_cols, parallelism, map_blob_cols=()):
+ """Fetch scalar and MAP BLOB payload bytes.
``data`` is a ``dict`` mapping each BLOB column name to row-aligned cells.
- Each cell may be serialized ``BlobDescriptor`` bytes, inline payload bytes,
- or ``None``. Returned values preserve row order and are grouped per column.
+ A cell may be serialized ``BlobDescriptor`` bytes, inline payload bytes,
+ ``None``, or a MAP represented by key-value pairs. Returned values preserve
+ row and MAP entry order and are grouped per column.
"""
from pypaimon.table.row.blob import BlobDescriptor, BlobViewStruct
ranges = []
inline = {}
- index = 0
+ targets = []
+ bodies = {col: [] for col in blob_cols}
+ scalar_offsets = {}
+ map_blob_cols = set(map_blob_cols)
+
+ def queue_blob_fetch(value):
+ index = len(ranges)
+ 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):
+ descriptor = BlobDescriptor.deserialize(raw)
+ ranges.append(
+ (descriptor.uri, descriptor.offset, descriptor.length)
+ )
+ else:
+ ranges.append(None)
+ inline[index] = raw
+ return index
+
for col in blob_cols:
+ if col not in map_blob_cols:
+ start = len(ranges)
+ for value in data[col]:
+ queue_blob_fetch(value)
+ scalar_offsets[col] = (start, len(ranges))
+ continue
+
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):
- descriptor = BlobDescriptor.deserialize(raw)
- ranges.append((descriptor.uri, descriptor.offset,
descriptor.length))
- else:
- ranges.append(None)
- inline[index] = raw
- index += 1
-
- fetched = file_io.read_ranges_coalesced(ranges, parallelism)
+ bodies[col].append(None)
+ continue
+
+ entries = _map_entries(value)
+ row_index = len(bodies[col])
+ row = []
+ bodies[col].append(row)
+ for key, item in entries:
+ entry_index = len(row)
+ row.append((key, None))
+ range_index = queue_blob_fetch(item)
+ targets.append((col, row_index, entry_index, range_index))
+
+ fetched = (
+ file_io.read_ranges_coalesced(ranges, parallelism)
+ if ranges
+ else []
+ )
for index, raw in inline.items():
fetched[index] = raw
- bodies = {}
- offset = 0
- for col in blob_cols:
- count = len(data[col])
- bodies[col] = fetched[offset:offset + count]
- offset += count
+ for col, (start, end) in scalar_offsets.items():
+ bodies[col] = fetched[start:end]
+ for col, row_index, entry_index, index in targets:
+ key = bodies[col][row_index][entry_index][0]
+ bodies[col][row_index][entry_index] = (key, fetched[index])
return bodies
+
+
+def _map_entries(value):
+ if isinstance(value, dict):
+ return list(value.items())
+ if isinstance(value, (list, tuple)) and all(
+ isinstance(entry, (list, tuple)) and len(entry) == 2
+ for entry in value):
+ return value
+ return None
diff --git a/paimon-python/pypaimon/multimodal/query.py
b/paimon-python/pypaimon/multimodal/query.py
index d492fc3a58..39d24fd88d 100644
--- a/paimon-python/pypaimon/multimodal/query.py
+++ b/paimon-python/pypaimon/multimodal/query.py
@@ -21,7 +21,7 @@ import pyarrow as pa
from pypaimon.common.where_parser import parse_where_clause
from pypaimon.multimodal.blob_read import fetch_blob_bodies
-from pypaimon.schema.data_types import is_blob_type
+from pypaimon.schema.data_types import is_blob_type, is_map_blob_type
from pypaimon.table.special_fields import SpecialFields
@@ -140,14 +140,16 @@ class ScanQuery:
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,
+ ) -> Tuple[pa.Table, Dict[str, List[Any]]]:
+ """Materialise BLOB or MAP 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.
+ ``select(...)``). Scalar BLOB values are ``bytes|None``; MAP BLOB rows
+ are ``None`` or key-value pairs with ``bytes|None`` values. Returns a
+ row-aligned ``(scalar_arrow_table, blobs_by_column)`` tuple. Use
+ :meth:`stream_blobs` for a memory-bounded read.
Unresolved blob-view columns are not supported and raise
``ValueError``.
"""
@@ -155,15 +157,16 @@ class ScanQuery:
read_builder, file_io = self._blob_descriptor_read_builder(blob_cols)
arrow = read_builder.new_read().to_arrow(
read_builder.new_scan().plan().splits())
+ map_blob_cols = set(blob_cols) - set(self._all_blob_columns())
bodies = self._fetch_bodies(
- file_io, arrow.select(blob_cols).to_pydict(), blob_cols,
parallelism)
+ file_io, arrow.select(blob_cols).to_pydict(), blob_cols,
+ parallelism, map_blob_cols)
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.
+ """Memory-bounded streaming variant of :meth:`read_blobs`, with the
same
+ return shape per Arrow batch and one-batch peak memory.
"""
# Validate eagerly so a bad column raises here, not on the first
next().
blob_cols = self._resolve_blob_columns(columns)
@@ -173,10 +176,12 @@ class ScanQuery:
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())
+ map_blob_cols = set(blob_cols) - set(self._all_blob_columns())
try:
for batch in reader:
bodies = self._fetch_bodies(
- file_io, batch.select(blob_cols).to_pydict(), blob_cols,
parallelism)
+ file_io, batch.select(blob_cols).to_pydict(), blob_cols,
+ parallelism, map_blob_cols)
scalar = batch.select(self._scalar_columns(batch.schema.names))
yield scalar, bodies
finally:
@@ -232,7 +237,8 @@ class ScanQuery:
return read_builder, read_table.file_io
@staticmethod
- def _fetch_bodies(file_io, data, blob_cols, parallelism):
+ def _fetch_bodies(
+ file_io, data, blob_cols, parallelism, map_blob_cols=()):
# 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
@@ -240,7 +246,8 @@ class ScanQuery:
# ``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.
- return fetch_blob_bodies(file_io, data, blob_cols, parallelism)
+ return fetch_blob_bodies(
+ file_io, data, blob_cols, parallelism, map_blob_cols)
def _all_blob_columns(self) -> List[str]:
return [
@@ -248,8 +255,14 @@ class ScanQuery:
if is_blob_type(field.type)
]
+ def _readable_blob_columns(self) -> List[str]:
+ return [
+ field.name for field in self._table.fields
+ if is_blob_type(field.type) or is_map_blob_type(field.type)
+ ]
+
def _resolve_blob_columns(self, columns) -> List[str]:
- all_blob = self._all_blob_columns()
+ all_blob = self._readable_blob_columns()
if columns is None:
selected = all_blob
if self._projection is not None:
@@ -270,7 +283,7 @@ class ScanQuery:
# 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())
+ blob_set = set(self._readable_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]
@@ -285,7 +298,7 @@ class ScanQuery:
# 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())
+ blob_set = set(self._readable_blob_columns())
effective = self._effective_projection()
if effective is None:
return [name for name in available if name not in blob_set]
diff --git a/paimon-python/pypaimon/tests/multimodal_table_test.py
b/paimon-python/pypaimon/tests/multimodal_table_test.py
index 2cfa1ad0ba..cc847186e4 100644
--- a/paimon-python/pypaimon/tests/multimodal_table_test.py
+++ b/paimon-python/pypaimon/tests/multimodal_table_test.py
@@ -842,6 +842,82 @@ class MultimodalTableTest(unittest.TestCase):
_, blobs3 = obs.scan().where("clip = 'c1'").read_blobs(["img", "img"])
self.assertEqual({"img"}, set(blobs3))
+ def test_scan_read_and_stream_map_blobs(self):
+ map_blob_type = pa.map_(pa.string(), pa.large_binary())
+ schema = _schema({
+ "id": pa.int32(),
+ "preview": pa.large_binary(),
+ "assets": map_blob_type,
+ })
+ obs = self.conn.create_table(
+ "map_blobs",
+ schema=schema,
+ options=_PARQUET_OPTIONS,
+ )
+ obs.add(pa.Table.from_pydict({
+ "id": [1, 2, 3, 4],
+ "preview": [b"preview-1", None, b"preview-3", b"preview-4"],
+ "assets": pa.array(
+ [
+ [
+ ("thumb", b"thumb-1"),
+ ("empty", b""),
+ ("missing", None),
+ ],
+ None,
+ [],
+ [("original", b"original-4")],
+ ],
+ type=map_blob_type,
+ ),
+ }, schema=schema))
+
+ scalar, blobs = obs.scan().read_blobs(
+ ["preview", "assets"], parallelism=2
+ )
+ ids = scalar.column("id").to_pylist()
+ self.assertNotIn("assets", scalar.column_names)
+ self.assertEqual(
+ {
+ 1: b"preview-1",
+ 2: None,
+ 3: b"preview-3",
+ 4: b"preview-4",
+ },
+ dict(zip(ids, blobs["preview"])),
+ )
+ self.assertEqual(
+ {
+ 1: {"thumb": b"thumb-1", "empty": b"", "missing": None},
+ 2: None,
+ 3: {},
+ 4: {"original": b"original-4"},
+ },
+ {
+ row_id: None if values is None else dict(values)
+ for row_id, values in zip(ids, blobs["assets"])
+ },
+ )
+
+ streamed = {}
+ for scalar_batch, blob_batch in obs.scan().stream_blobs("assets"):
+ for row_id, values in zip(
+ scalar_batch.column("id").to_pylist(),
+ blob_batch["assets"]):
+ streamed[row_id] = None if values is None else dict(values)
+ self.assertEqual(
+ {
+ 1: {"thumb": b"thumb-1", "empty": b"", "missing": None},
+ 2: None,
+ 3: {},
+ 4: {"original": b"original-4"},
+ },
+ streamed,
+ )
+
+ _, selected = obs.scan().select(["id", "assets"]).read_blobs()
+ self.assertEqual({"assets"}, set(selected))
+
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(
