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 2cd3305b41 [python] Read BlobStore listings in bounded batches (#10044)
2cd3305b41 is described below
commit 2cd3305b4176a56bd16ce08cd41c23024c3feafc
Author: chaoyang <[email protected]>
AuthorDate: Mon Sep 21 15:04:21 2026 +0800
[python] Read BlobStore listings in bounded batches (#10044)
---
docs/docs/pypaimon/blob-store.md | 5 +-
paimon-python/pypaimon/multimodal/blob_store.py | 33 +++++++----
.../pypaimon/tests/multimodal_table_test.py | 67 ++++++++++++++++++++++
3 files changed, 93 insertions(+), 12 deletions(-)
diff --git a/docs/docs/pypaimon/blob-store.md b/docs/docs/pypaimon/blob-store.md
index 2875bcc8dd..faeb717d2f 100644
--- a/docs/docs/pypaimon/blob-store.md
+++ b/docs/docs/pypaimon/blob-store.md
@@ -133,7 +133,10 @@ object stream. `get_object`, `head_object`, and
`list_objects` expose non-key,
non-BLOB table columns through `columns`. These columns are all returned by
default. Pass `columns` to return only selected columns, or `[]` to skip them.
`list_objects` requires a non-negative `limit`; `limit=0` returns an empty
-list.
+list. It reads object metadata in batches and stops after collecting `limit`
+matching objects. Prefix filtering can still require scanning nonmatching rows;
+object payloads are not fetched. Without a limit, the returned list holds all
+matching objects.
```python
obj = store.get_object("images/cat.jpg", range="bytes=0-1023")
diff --git a/paimon-python/pypaimon/multimodal/blob_store.py
b/paimon-python/pypaimon/multimodal/blob_store.py
index 1c4e25d851..bed1ff22a0 100644
--- a/paimon-python/pypaimon/multimodal/blob_store.py
+++ b/paimon-python/pypaimon/multimodal/blob_store.py
@@ -201,18 +201,29 @@ class BlobStore:
raise ValueError("limit must be greater than or equal to 0.")
if limit == 0:
return []
- rows = self._read_rows(
- include_blob=True,
- columns=columns,
- )
+ read_table =
self._raw_table.copy({CoreOptions.BLOB_AS_DESCRIPTOR.key(): "true"})
+ read_builder = read_table.new_read_builder().with_projection(
+ self._projection(True, columns))
+ # A prefix filters the exposed key's string representation, including
+ # non-string keys. Only push the limit when every row is a match.
+ if limit is not None and prefix is None:
+ read_builder = read_builder.with_limit(limit)
+ reader = read_builder.new_read()._to_managed_arrow_batch_reader(
+ read_builder.new_scan().plan().splits())
objects = []
- for row in rows:
- key = row[self.key_column]
- if prefix is not None and not str(key).startswith(prefix):
- continue
- objects.append(self._row_to_info(row))
- if limit is not None and len(objects) >= limit:
- break
+ try:
+ for batch in reader:
+ data = batch.to_pydict()
+ for values in zip(*data.values()):
+ row = dict(zip(data, values))
+ key = row[self.key_column]
+ if prefix is not None and not str(key).startswith(prefix):
+ continue
+ objects.append(self._row_to_info(row))
+ if limit is not None and len(objects) >= limit:
+ return objects
+ finally:
+ reader.close()
return objects
def delete_object(self, key) -> None:
diff --git a/paimon-python/pypaimon/tests/multimodal_table_test.py
b/paimon-python/pypaimon/tests/multimodal_table_test.py
index e4264ae914..51edc266dc 100644
--- a/paimon-python/pypaimon/tests/multimodal_table_test.py
+++ b/paimon-python/pypaimon/tests/multimodal_table_test.py
@@ -609,6 +609,73 @@ class MultimodalTableTest(unittest.TestCase):
store.delete_object("images/cat.jpg")
self.assertEqual([], store.list_objects(prefix="images/"))
+ def test_blob_store_list_reads_batches_and_stops_at_limit(self):
+ from pypaimon.read.table_read import TableRead
+
+ table = self.conn.create_table(
+ "list_batches", schema=_schema({
+ "key": pa.string(), "image": pa.large_binary(), "owner":
pa.string(),
+ }), options=dict(_PARQUET_OPTIONS, **{"read.batch-size": "2"}))
+ table.add([
+ {"key": key, "image": b"body", "owner": "alice"}
+ for key in ("other/a", "other/b", "images/a", "images/b",
"images/c", "images/d")
+ ])
+ store = table.blobs(column="image")
+ original = TableRead._arrow_batch_generator
+ batches, closed = [], []
+
+ def tracked_read(read, *args, **kwargs):
+ reader = original(read, *args, **kwargs)
+ try:
+ for batch in reader:
+ batches.append(batch.num_rows)
+ yield batch
+ finally:
+ reader.close()
+ closed.append(True)
+
+ with patch.object(TableRead, "to_arrow",
side_effect=AssertionError("full read")), \
+ patch.object(TableRead, "_arrow_batch_generator",
tracked_read):
+ self.assertEqual([], store.list_objects(limit=0))
+ self.assertEqual([], batches)
+ listed = store.list_objects(prefix="images/", limit=1,
columns=["owner"])
+ self.assertEqual(["images/a"], [obj.key for obj in listed])
+ self.assertEqual({"owner": "alice"}, listed[0].columns)
+ self.assertEqual(4, listed[0].size)
+ self.assertEqual([2, 2], batches)
+ self.assertEqual([True], closed)
+
+ batches.clear()
+ listed = store.list_objects(limit=1, columns=[])
+ self.assertEqual(1, len(listed))
+ self.assertEqual({}, listed[0].columns)
+ self.assertEqual([1], batches)
+ self.assertEqual(2, len(closed))
+
+ with patch.object(store, "_row_to_info",
side_effect=ValueError("invalid descriptor")):
+ try:
+ store.list_objects()
+ except ValueError as error:
+ self.assertEqual("invalid descriptor", str(error))
+ # Check while the traceback still holds the reader's frame.
+ self.assertEqual(3, len(closed))
+ else:
+ self.fail("Expected invalid descriptor")
+ self.assertEqual([], store.list_objects(prefix="absent/"))
+ self.assertEqual(4, len(closed))
+
+ def test_blob_store_list_preserves_non_string_prefix_semantics(self):
+ table = self.conn.create_table(
+ "numeric_keys", schema=_schema({"key": pa.int32(), "image":
pa.large_binary()}),
+ options=_PARQUET_OPTIONS)
+ table.add([
+ {"key": value, "image": b"body"} for value in (None, 10, 20, 21)
+ ])
+ store = table.blobs(column="image")
+ self.assertEqual([None], [obj.key for obj in
store.list_objects(prefix="N", limit=1)])
+ self.assertEqual([20], [obj.key for obj in
store.list_objects(prefix="2", limit=1)])
+ self.assertEqual(4, len(store.list_objects(prefix="")))
+
def test_blob_store_put_object_accepts_blob_without_materializing(self):
from pypaimon.table.row.blob import Blob, BlobDescriptor
