[python] Open one remote stream for concurrent cached reads

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):

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)