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 18fcff65e3 [python] Close exhausted blob fallback readers to improve
performance (#8622)
18fcff65e3 is described below
commit 18fcff65e350466127cf8245c8ff063813831746
Author: umi <[email protected]>
AuthorDate: Wed Jul 15 08:03:53 2026 +0800
[python] Close exhausted blob fallback readers to improve performance
(#8622)
- close cached blob fallback readers once their selected ranges are
permanently behind the current batch
- preserve final reader cleanup through BlobFallbackBatchReader.close
- add coverage that verifies exhausted readers are closed as batches
advance
---
.../pypaimon/read/reader/concat_batch_reader.py | 15 ++--
paimon-python/pypaimon/tests/blob_test.py | 94 ++++++++++++++++++++++
2 files changed, 103 insertions(+), 6 deletions(-)
diff --git a/paimon-python/pypaimon/read/reader/concat_batch_reader.py
b/paimon-python/pypaimon/read/reader/concat_batch_reader.py
index e45418a018..f018ccf756 100644
--- a/paimon-python/pypaimon/read/reader/concat_batch_reader.py
+++ b/paimon-python/pypaimon/read/reader/concat_batch_reader.py
@@ -324,6 +324,8 @@ class BlobFallbackBatchReader(RecordBatchReader):
group[row_id] = (blob.to_descriptor().serialize(),
False)
else:
group[row_id] = (blob.to_data(), False)
+ if state.selected_range_index >= len(state.selected_ranges):
+ self._close_state_reader(state)
if not groups:
return None
@@ -415,9 +417,8 @@ class BlobFallbackBatchReader(RecordBatchReader):
row_id - self._deletion_vector_range.from_
)
- @staticmethod
def _state_overlaps_batch(
- state: _BlobFileState, batch_first: int, batch_last: int
+ self, state: _BlobFileState, batch_first: int, batch_last: int
) -> bool:
selected_ranges = state.selected_ranges
while (
@@ -428,10 +429,12 @@ class BlobFallbackBatchReader(RecordBatchReader):
selected_ranges[state.selected_range_index].count()
)
state.selected_range_index += 1
- return (
- state.selected_range_index < len(selected_ranges)
- and selected_ranges[state.selected_range_index].from_ <= batch_last
- )
+ if state.selected_range_index >= len(selected_ranges):
+ # Batch row ids only move forward. Once the last selected range is
+ # behind this batch, the reader can never be used again.
+ self._close_state_reader(state)
+ return False
+ return selected_ranges[state.selected_range_index].from_ <= batch_last
def _read_blob_values(
self, state: _BlobFileState, batch_row_ids: List[int]
diff --git a/paimon-python/pypaimon/tests/blob_test.py
b/paimon-python/pypaimon/tests/blob_test.py
index 86c7b5daa0..68b72da696 100644
--- a/paimon-python/pypaimon/tests/blob_test.py
+++ b/paimon-python/pypaimon/tests/blob_test.py
@@ -437,6 +437,100 @@ class BlobTest(unittest.TestCase):
self.assertTrue(created_by_file["old.blob"][0].closed)
self.assertTrue(created_by_file["new.blob"][0].closed)
+ def test_blob_fallback_batch_reader_closes_exhausted_readers(self):
+ class FakeBlobReader:
+ def __init__(self, file_path, offset):
+ self._file_io = None
+ self.file_path = file_path
+ self.blob_lengths = [20]
+ self.blob_offsets = [offset]
+ self._input_stream = None
+ self.closed = False
+
+ def close(self):
+ self.closed = True
+
+ def data_file(name, first_row_id, max_sequence_number):
+ return DataFileMeta(
+ file_name=name,
+ file_size=0,
+ row_count=1,
+ min_key=None,
+ max_key=None,
+ key_stats=None,
+ value_stats=None,
+ min_sequence_number=max_sequence_number,
+ max_sequence_number=max_sequence_number,
+ schema_id=0,
+ level=0,
+ extra_files=[],
+ first_row_id=first_row_id,
+ file_path=name,
+ )
+
+ def supplier(file_path, offset):
+ def create_reader():
+ reader = FakeBlobReader(file_path, offset)
+ created_by_file.setdefault(file_path, []).append(reader)
+ return reader
+
+ return create_reader
+
+ def descriptor_offsets(batch):
+ return [
+ BlobDescriptor.deserialize(value.as_py()).offset
+ for value in batch.column(0)
+ ]
+
+ for batch_size in [1, 1024]:
+ with self.subTest(batch_size=batch_size):
+ created_by_file = {}
+ reader = BlobFallbackBatchReader(
+ [
+ (data_file("first.blob", 0, 1), supplier("first.blob",
0)),
+ (
+ data_file("second.blob", 10, 2),
+ supplier("second.blob", 100),
+ ),
+ (
+ data_file("third.blob", 20, 3),
+ supplier("third.blob", 200),
+ ),
+ ],
+ "picture",
+ pa.large_binary(),
+ blob_as_descriptor=True,
+ batch_size=batch_size,
+ )
+
+ if batch_size == 1:
+ first = reader.read_arrow_batch()
+ self.assertEqual([4], descriptor_offsets(first))
+ self.assertFalse(created_by_file["first.blob"][0].closed)
+
+ second = reader.read_arrow_batch()
+ self.assertEqual([104], descriptor_offsets(second))
+ self.assertTrue(created_by_file["first.blob"][0].closed)
+ self.assertFalse(created_by_file["second.blob"][0].closed)
+
+ third = reader.read_arrow_batch()
+ self.assertEqual([204], descriptor_offsets(third))
+ self.assertTrue(created_by_file["second.blob"][0].closed)
+ self.assertFalse(created_by_file["third.blob"][0].closed)
+ else:
+ batch = reader.read_arrow_batch()
+ self.assertEqual([4, 104, 204], descriptor_offsets(batch))
+ self.assertTrue(created_by_file["first.blob"][0].closed)
+ self.assertTrue(created_by_file["second.blob"][0].closed)
+ self.assertFalse(created_by_file["third.blob"][0].closed)
+
+ self.assertIsNone(reader.read_arrow_batch())
+ self.assertEqual(1, len(created_by_file["first.blob"]))
+ self.assertEqual(1, len(created_by_file["second.blob"]))
+ self.assertEqual(1, len(created_by_file["third.blob"]))
+ reader.close()
+ self.assertTrue(created_by_file["third.blob"][0].closed)
+
def test_blob_data_interface_compliance(self):
"""Test that BlobData properly implements Blob interface."""
test_data = b"interface test data"
