XiaoHongbo-Hope commented on code in PR #8439:
URL: https://github.com/apache/paimon/pull/8439#discussion_r3519447624
##########
paimon-python/pypaimon/multimodal/query.py:
##########
@@ -98,6 +100,143 @@ def to_pandas(self):
def to_list(self) -> List[dict]:
return self.to_arrow().to_pylist()
+ def read_blobs(
+ self, columns=None, *, max_workers: 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.
+ """
+ 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,
max_workers)
+ scalar = arrow.select(self._scalar_columns(arrow.column_names))
+ return scalar, bodies
+
+ def stream_blobs(self, columns=None, *, max_workers: 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, max_workers)
+
+ def _iter_blobs(self, read_builder, file_io, blob_cols, max_workers):
+ 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,
max_workers)
+ 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, max_workers):
+ # Decode each cell (descriptor / view / inline) via Blob.from_bytes,
+ # then read all columns in one coalesced pass.
+ from pypaimon.table.row.blob import Blob
+ flat = []
+ for col in blob_cols:
+ for value in data[col]:
+ flat.append(None if value is None
+ else Blob.from_bytes(value, file_io))
Review Comment:
> If the descriptor read still returns `BlobViewStruct` bytes (for example
when blob-view resolution is disabled or no catalog loader is available),
`Blob.from_bytes` creates an unresolved `BlobView`. `read_blobs_concurrent`
only treats `BlobRef` specially, so it will put that `BlobView` on the
in-memory path and call `to_data()`, which raises `BlobView is not resolved`.
Since this helper advertises decoding descriptor/view/inline cells, please
either resolve view structs before this point or reject/document that
unresolved blob-view fields are unsupported by `read_blobs`.
Thanks, fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
