XiaoHongbo-Hope commented on code in PR #8439:
URL: https://github.com/apache/paimon/pull/8439#discussion_r3518117539
##########
paimon-python/pypaimon/multimodal/query.py:
##########
@@ -80,6 +82,128 @@ 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)
+ # Drop all BLOB columns so unrequested ones don't leak in as
descriptors.
+ blob_set = set(self._all_blob_columns())
+ scalar = arrow.select([
+ name for name in arrow.column_names if name not in blob_set
+ ])
+ 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):
+ blob_set = set(self._all_blob_columns())
+ 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([
+ name for name in batch.schema.names if name not in blob_set
+ ])
+ 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):
+ # Flatten all columns into one coalesced read so they fetch together,
not serially.
+ from pypaimon.table.row.blob import BlobDescriptor
+ flat = []
+ for col in blob_cols:
+ for value in data[col]:
+ if value is None:
+ flat.append(None)
+ else:
+ d = BlobDescriptor.deserialize(bytes(value))
+ flat.append((d.uri, d.offset, d.length))
+ fetched = file_io.read_ranges_coalesced(flat, max_workers)
+ 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]:
+ # Read the selected non-BLOB columns + only the requested BLOB
+ # descriptors; never pull unrequested BLOB columns.
+ blob_set = set(self._all_blob_columns())
+ if self._projection is None:
+ base = [f.name for f in self._table.fields if f.name not in
blob_set]
+ else:
+ base = [name for name in self._projection if name not in blob_set]
+ return base + [c for c in blob_cols if c not in base]
Review Comment:
> `read_blobs()` and `stream_blobs()` can silently drop the `where()` filter
when the user projection excludes the predicate column.
`_blob_read_projection()` only keeps the selected non-BLOB columns plus the
requested BLOB descriptors; for data-evolution reads, the row-level predicate
is only applied when all referenced predicate fields are present in the read
schema. For example, `scan().where("idx = 1").select(["image"]).read_blobs()`
reads only `image`, so `idx` is absent and all rows are returned. The same
applies to `stream_blobs()`.
>
> Please include the predicate-referenced columns in the internal
descriptor-read projection, then remove those helper columns from the returned
scalar table so the public `select()` result remains unchanged.
Fixed and added assertion
--
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]
