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 0f2c69716a [python] Close BLOB stream iterators on errors (#10059)
0f2c69716a is described below
commit 0f2c69716ae9df4c74972073d831713594ad44fc
Author: chaoyang <[email protected]>
AuthorDate: Tue Sep 22 10:44:23 2026 +0800
[python] Close BLOB stream iterators on errors (#10059)
---
paimon-python/pypaimon/multimodal/query.py | 9 ++---
.../pypaimon/tests/multimodal_table_test.py | 41 ++++++++++++++++++++++
2 files changed, 43 insertions(+), 7 deletions(-)
diff --git a/paimon-python/pypaimon/multimodal/query.py
b/paimon-python/pypaimon/multimodal/query.py
index 17ea818eef..1d65eb4ffb 100644
--- a/paimon-python/pypaimon/multimodal/query.py
+++ b/paimon-python/pypaimon/multimodal/query.py
@@ -318,20 +318,15 @@ class ScanQuery:
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())
map_blob_cols, array_blob_cols = self._nested_blob_columns()
- try:
+ with read_builder.new_read()._to_managed_arrow_batch_reader(
+ read_builder.new_scan().plan().splits()) as reader:
for batch in reader:
bodies = self._fetch_bodies(
file_io, batch.select(blob_cols).to_pydict(), blob_cols,
parallelism, map_blob_cols, array_blob_cols)
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_query_read_builder(self):
from pypaimon.common.options.core_options import CoreOptions
diff --git a/paimon-python/pypaimon/tests/multimodal_table_test.py
b/paimon-python/pypaimon/tests/multimodal_table_test.py
index b5e04a7238..71160f8a8d 100644
--- a/paimon-python/pypaimon/tests/multimodal_table_test.py
+++ b/paimon-python/pypaimon/tests/multimodal_table_test.py
@@ -1662,6 +1662,47 @@ class MultimodalTableTest(unittest.TestCase):
self.assertEqual(
[], list(obs.scan().where("clip = 'none'").stream_blobs("image")))
+ def test_scan_stream_blobs_closes_underlying_iterator(self):
+ from pypaimon.read.table_read import TableRead
+
+ table = self.conn.create_table(
+ "stream_cleanup", schema=_schema({
+ "id": pa.int32(), "image": pa.large_binary(),
+ }), options=dict(_PARQUET_OPTIONS, **{"read.batch-size": "1"}))
+ table.add([{"id": i, "image": b"body"} for i in range(3)])
+ original = TableRead._arrow_batch_generator
+ closed = []
+
+ def tracked_read(read, *args, **kwargs):
+ reader = original(read, *args, **kwargs)
+ try:
+ yield from reader
+ finally:
+ reader.close()
+ closed.append(True)
+
+ with patch.object(TableRead, "_try_native_batches",
return_value=None), \
+ patch.object(TableRead, "_arrow_batch_generator",
tracked_read):
+ stream = table.scan().stream_blobs("image")
+ next(stream)
+ stream.close()
+ self.assertEqual([True], closed)
+
+ for method in ("_fetch_bodies", "_scalar_columns"):
+ with self.subTest(method=method):
+ closed.clear()
+ query = table.scan()
+ with patch.object(query, method,
side_effect=ValueError("batch failed")):
+ stream = query.stream_blobs("image")
+ try:
+ next(stream)
+ except ValueError as error:
+ self.assertEqual("batch failed", str(error))
+ # Assert while the traceback still retains the
reader's frame.
+ self.assertEqual([True], closed)
+ else:
+ self.fail("Expected batch failure")
+
@unittest.skipIf(ray is None, "ray is not installed")
def test_scan_to_ray_map_with_blobs(self):
started_ray = False
