From d100e5b5274a79da2fc6c69ff736a303010a923b Mon Sep 17 00:00:00 2001 From: "wenchao.wu" Date: Fri, 30 Oct 2026 16:12:45 +0800 Subject: [PATCH] [python] Open one remote stream for concurrent cached reads CachingInputStream.read_at is documented as thread-safe and is called concurrently by the parallel readers, but _get_remote_stream tested and assigned _stream without a lock. Concurrent cache misses each opened a remote stream while only the last one was retained, so close() left the rest open. Reads stayed correct because _io_lock still serialized the actual I/O; the cost was leaked object store connections. Guard the lazy open with a dedicated lock and re-check inside it. The lock is separate from _io_lock, which serializes seek and read on an already open stream, so opening does not block in-flight reads. With 16 threads reading disjoint cold offsets, the stream count drops from 16 to 1 and close() leaves none behind. --- .../pypaimon/filesystem/caching_file_io.py | 9 +++- .../pypaimon/tests/caching_file_io_test.py | 48 +++++++++++++++++++ 2 files changed, 56 insertions(+), 1 deletion(-) diff --git a/paimon-python/pypaimon/filesystem/caching_file_io.py b/paimon-python/pypaimon/filesystem/caching_file_io.py index d60d8c2ecdcf..f968f1b3d88c 100644 --- a/paimon-python/pypaimon/filesystem/caching_file_io.py +++ b/paimon-python/pypaimon/filesystem/caching_file_io.py @@ -274,6 +274,7 @@ def __init__(self, file_io, file_path: str, cache, file_size=None): self._cache = cache self._pos = 0 self._io_lock = threading.Lock() + self._open_lock = threading.Lock() self._remote_supports_pread = None def readable(self): @@ -398,8 +399,14 @@ def _read_fully(self, stream, size: int) -> bytes: return bytes(buf) def _get_remote_stream(self): + # read_at is documented as thread-safe, so concurrent misses must not + # each open their own stream: only the first one is retained and the + # rest would leak. Kept separate from _io_lock, which serializes seek + # and read on an already open stream. if self._stream is None: - self._stream = self._file_io.new_input_stream(self._file_path) + with self._open_lock: + if self._stream is None: + self._stream = self._file_io.new_input_stream(self._file_path) return self._stream def close(self): diff --git a/paimon-python/pypaimon/tests/caching_file_io_test.py b/paimon-python/pypaimon/tests/caching_file_io_test.py index ae5ff0704423..6b09c71c2a80 100644 --- a/paimon-python/pypaimon/tests/caching_file_io_test.py +++ b/paimon-python/pypaimon/tests/caching_file_io_test.py @@ -26,6 +26,7 @@ import shutil import tempfile import threading +import time import tracemalloc import unittest from concurrent.futures import ThreadPoolExecutor @@ -279,6 +280,53 @@ def test_cache_hit_avoids_remote_read(self): self.assertEqual(data, result) file_io2.new_input_stream.assert_not_called() + def test_concurrent_read_at_opens_one_remote_stream(self): + # read_at is documented as thread-safe and is called concurrently by + # the parallel readers, so a cold stream must be opened exactly once. + data = bytes(i % 251 for i in range(200000)) + opened = [] + lock = threading.Lock() + + class _SeekOnlyStream(io.BytesIO): + def fileno(self): + raise OSError("no fileno") + + file_io = MagicMock() + file_io.get_file_size.side_effect = lambda p: len(data) + + def open_stream(path): + # Hold the opener briefly so concurrent misses overlap. + time.sleep(0.02) + stream = _SeekOnlyStream(data) + with lock: + opened.append(stream) + return stream + + file_io.new_input_stream.side_effect = open_stream + + cache = LocalMemoryCacheManager(2 ** 63 - 1, block_size=4096) + stream = CachingInputStream(file_io, "test.index", cache) + errors = [] + + def reader(offset): + try: + self.assertEqual(data[offset:offset + 100], stream.read_at(100, offset)) + except Exception as error: # noqa: B902 - surfaced through errors + errors.append(error) + + threads = [threading.Thread(target=reader, args=(i * 7000,)) for i in range(16)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + + self.assertEqual([], errors) + self.assertEqual(1, len(opened)) + + # Closing the stream must leave no remote stream behind. + stream.close() + self.assertTrue(all(opened_stream.closed for opened_stream in opened)) + def test_read_empty(self): cache = LocalDiskCacheManager(self.cache_dir, 2 ** 63 - 1, block_size=8) stream = CachingInputStream(self._make_file_io(b""), "test", cache)