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 dc17028b71 [python] Coalesce consecutive local cache misses (#10801)
dc17028b71 is described below
commit dc17028b71d3114fd009305b0d7784d5422df43a
Author: XiaoHongbo <[email protected]>
AuthorDate: Mon Oct 26 08:52:05 2026 +0800
[python] Coalesce consecutive local cache misses (#10801)
---
.../pypaimon/filesystem/caching_file_io.py | 69 +++++++++--------
.../pypaimon/tests/caching_file_io_test.py | 90 ++++++++++++++++++++++
2 files changed, 128 insertions(+), 31 deletions(-)
diff --git a/paimon-python/pypaimon/filesystem/caching_file_io.py
b/paimon-python/pypaimon/filesystem/caching_file_io.py
index 58a1a9bd12..38a5b2f2b2 100644
--- a/paimon-python/pypaimon/filesystem/caching_file_io.py
+++ b/paimon-python/pypaimon/filesystem/caching_file_io.py
@@ -303,23 +303,11 @@ class CachingInputStream:
if size <= 0 or self._pos >= self._get_file_size():
return b''
- end = min(self._pos + size, self._get_file_size())
- block_size = self._cache.block_size
-
- first_block = self._pos // block_size
- last_block = (end - 1) // block_size
-
- result = bytearray()
- for bi in range(first_block, last_block + 1):
- block_data = self._read_block(bi)
-
- block_start = bi * block_size
- start_in_block = max(self._pos - block_start, 0)
- end_in_block = min(end - block_start, len(block_data))
- result.extend(block_data[start_in_block:end_in_block])
-
- self._pos = end
- return bytes(result)
+ position = self._pos
+ end = min(position + size, self._get_file_size())
+ result = self.read_at(end - position, position)
+ self._pos = position + len(result)
+ return result
def read_at(self, nbytes: int, offset: int) -> bytes:
"""Position-based read. Does not change the cursor. Thread-safe."""
@@ -333,9 +321,7 @@ class CachingInputStream:
last_block = (end - 1) // block_size
result = bytearray()
- for bi in range(first_block, last_block + 1):
- block_data = self._read_block(bi)
-
+ for bi, block_data in self._read_blocks(first_block, last_block):
block_start = bi * block_size
start_in_block = max(offset - block_start, 0)
end_in_block = min(end - block_start, len(block_data))
@@ -343,18 +329,39 @@ class CachingInputStream:
return bytes(result)
- def _read_block(self, block_index: int) -> bytes:
- cached = self._cache.get_block(self._file_path, block_index)
- if cached is not None:
- return cached
-
+ def _read_blocks(self, first: int, last: int):
block_size = self._cache.block_size
- offset = block_index * block_size
- read_size = min(block_size, self._get_file_size() - offset)
-
- data = self._read_remote(offset, read_size)
- self._cache.put_block(self._file_path, block_index, data)
- return data
+ # Bound temporary buffers; a larger individual block remains unchanged.
+ batch_size = max(1, min(32, (8 * 1024 * 1024) // block_size))
+ for batch_start in range(first, last + 1, batch_size):
+ batch_end = min(batch_start + batch_size, last + 1)
+ blocks = [self._cache.get_block(self._file_path, bi)
+ for bi in range(batch_start, batch_end)]
+ i = 0
+ while i < len(blocks):
+ if blocks[i] is not None:
+ yield batch_start + i, blocks[i]
+ i += 1
+ continue
+ begin = i
+ while i < len(blocks) and blocks[i] is None:
+ i += 1
+ offset = (batch_start + begin) * block_size
+ end = min((batch_start + i) * block_size,
self._get_file_size())
+ data = self._read_remote(offset, end - offset)
+ if len(data) > end - offset:
+ raise OSError("Cache range read exceeded request")
+ complete = len(data) == end - offset
+ for j in range(begin, i):
+ lo = (j - begin) * block_size
+ block = data[lo:lo + block_size]
+ if not block:
+ break
+ if complete:
+ self._cache.put_block(self._file_path, batch_start +
j, block)
+ yield batch_start + j, block
+ if not complete:
+ return
def _read_remote(self, offset: int, size: int) -> bytes:
stream = self._get_remote_stream()
diff --git a/paimon-python/pypaimon/tests/caching_file_io_test.py
b/paimon-python/pypaimon/tests/caching_file_io_test.py
index 88edfbcf4b..25078989c2 100644
--- a/paimon-python/pypaimon/tests/caching_file_io_test.py
+++ b/paimon-python/pypaimon/tests/caching_file_io_test.py
@@ -287,6 +287,96 @@ class CachingInputStreamTest(unittest.TestCase):
# Verify block 1 is only 2 bytes
self.assertEqual(b"ij", cache.get_block("test", 1))
+ def test_coalesce_cold_blocks_and_preserve_cursor(self):
+ data = b"abcdefghijklmn"
+ cache = LocalMemoryCacheManager(1024, block_size=4)
+ stream = CachingInputStream(self._make_file_io(data), "test", cache)
+ stream.seek(2)
+ with patch.object(stream, '_read_remote', wraps=stream._read_remote)
as remote:
+ self.assertEqual(data[1:13], stream.read_at(12, 1))
+ self.assertEqual(2, stream.tell())
+ remote.assert_called_once_with(0, 14)
+ self.assertEqual(data[2:], stream.read())
+ self.assertEqual(14, stream.tell())
+ self.assertEqual(1, remote.call_count)
+ self.assertEqual(b"mn", cache.get_block("test", 3))
+
+ def test_coalescing_does_not_download_cached_holes(self):
+ data = b"abcdefghijklmn"
+ cache = LocalMemoryCacheManager(1024, block_size=4)
+ cache.put_block("test", 1, data[4:8])
+ stream = CachingInputStream(self._make_file_io(data), "test", cache)
+ with patch.object(stream, '_read_remote', wraps=stream._read_remote)
as remote:
+ self.assertEqual(data[1:], stream.read_at(13, 1))
+ self.assertEqual([(0, 4), (8, 6)], [c.args for c in
remote.call_args_list])
+
+ def test_coalescing_bounds_blocks_and_bytes(self):
+ for block_size, blocks in [(4, 40), (1024 * 1024, 10), (9 * 1024 *
1024, 2)]:
+ with self.subTest(block_size=block_size):
+ data = b"x" * (block_size * blocks)
+ cache = LocalMemoryCacheManager(len(data),
block_size=block_size)
+ stream = CachingInputStream(self._make_file_io(data), "test",
cache)
+ with patch.object(stream, '_read_remote',
wraps=stream._read_remote) as remote:
+ self.assertEqual(data, stream.read())
+ self.assertEqual(2, remote.call_count)
+ self.assertEqual(len(data), sum(c.args[1] for c in
remote.call_args_list))
+ for call in remote.call_args_list:
+ self.assertLessEqual(call.args[1], max(block_size, 8 *
1024 * 1024))
+
+ def test_coalesced_short_response_returns_uncached_prefix(self):
+ data = b"abcdefghijkl"
+ cache = LocalMemoryCacheManager(1024, block_size=4)
+ stream = CachingInputStream(self._make_file_io(data), "test", cache)
+ with patch.object(stream, '_read_remote', return_value=data[:-1]):
+ self.assertEqual(data[:-1], stream.read())
+ self.assertEqual(len(data) - 1, stream.tell())
+ for i in range(3):
+ self.assertIsNone(cache.get_block("test", i))
+ self.assertEqual(data[-1:], stream.read())
+ stream.seek(0)
+ self.assertEqual(data, stream.read())
+
+ def test_coalesced_short_response_stops_before_cached_hole(self):
+ data = b"abcdefghijkl"
+ cache = LocalMemoryCacheManager(1024, block_size=4)
+ cache.put_block("test", 2, data[8:12])
+ stream = CachingInputStream(self._make_file_io(data), "test", cache)
+ with patch.object(stream, '_read_remote', return_value=b"abc"):
+ self.assertEqual(b"abc", stream.read_at(12, 0))
+ self.assertEqual(0, stream.tell())
+ self.assertIsNone(cache.get_block("test", 0))
+
+ def test_coalesced_positioned_source(self):
+ data = b"abcdefghijkl"
+
+ class PositionedStream:
+ def __init__(self):
+ self.calls = []
+
+ def read_at(self, size, offset):
+ self.calls.append((size, offset))
+ return data[offset:offset + size]
+
+ source = PositionedStream()
+ file_io = self._make_file_io(data)
+ file_io.new_input_stream.side_effect = lambda p: source
+ stream = CachingInputStream(file_io, "test",
LocalMemoryCacheManager(1024, block_size=4))
+ self.assertEqual(data[1:], stream.read_at(11, 1))
+ self.assertEqual([(12, 0)], source.calls)
+ self.assertEqual(0, stream.tell())
+
+ def test_coalesced_seek_source_handles_short_chunks(self):
+ data = b"abcdefghijkl"
+
+ class ShortChunkStream(io.BytesIO):
+ def read(self, size=-1):
+ return super().read(min(size, 3))
+
+ file_io = self._make_file_io(data)
+ file_io.new_input_stream.side_effect = lambda p: ShortChunkStream(data)
+ stream = CachingInputStream(file_io, "test",
LocalMemoryCacheManager(1024, block_size=4))
+ self.assertEqual(data, stream.read())
+
class CachingFileIOTest(unittest.TestCase):
