From cd18a1be6a8fa991577636252639525434176a2b Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Sun, 9 Aug 2026 02:52:52 -0700 Subject: [PATCH 01/16] [python] Reuse streams for coalesced blob reads --- .../catalog/rest/rest_token_file_io.py | 3 + paimon-python/pypaimon/common/file_io.py | 142 ++++++++++++--- .../pypaimon/filesystem/caching_file_io.py | 20 ++- .../filesystem/jindo_file_system_handler.py | 4 + .../pypaimon/filesystem/pyarrow_file_io.py | 5 + .../pypaimon/filesystem/resolving_file_io.py | 3 + paimon-python/pypaimon/tests/blob_test.py | 163 ++++++++++++++++-- .../pypaimon/tests/caching_file_io_test.py | 15 ++ .../pypaimon/tests/jindo_file_system_test.py | 33 ++++ .../pypaimon/tests/resolving_file_io_test.py | 11 ++ .../tests/rest/rest_token_file_io_test.py | 18 ++ 11 files changed, 377 insertions(+), 40 deletions(-) diff --git a/paimon-python/pypaimon/catalog/rest/rest_token_file_io.py b/paimon-python/pypaimon/catalog/rest/rest_token_file_io.py index 42dabb268cb2..c89455f64fec 100644 --- a/paimon-python/pypaimon/catalog/rest/rest_token_file_io.py +++ b/paimon-python/pypaimon/catalog/rest/rest_token_file_io.py @@ -143,6 +143,9 @@ def _merge_token_with_catalog_options(self, token: dict) -> dict: def new_input_stream(self, path: str): return self.file_io().new_input_stream(path) + def new_range_input_stream(self, path: str): + return self.file_io().new_range_input_stream(path) + def new_output_stream(self, path: str): return self.file_io().new_output_stream(path) diff --git a/paimon-python/pypaimon/common/file_io.py b/paimon-python/pypaimon/common/file_io.py index bfde96dc9058..a45513d0fa1f 100644 --- a/paimon-python/pypaimon/common/file_io.py +++ b/paimon-python/pypaimon/common/file_io.py @@ -17,6 +17,7 @@ import logging import os +import threading import uuid from abc import ABC, abstractmethod from pathlib import Path @@ -30,7 +31,7 @@ def supports_pread(stream) -> bool: """Check if the stream supports position-based reads (thread-safe I/O).""" - if hasattr(stream, 'read_at'): + if hasattr(stream, 'read_at') or hasattr(stream, 'pread'): return True if hasattr(stream, 'fileno'): try: @@ -43,6 +44,8 @@ def supports_pread(stream) -> bool: def pread(stream, length: int, offset: int) -> bytes: """Position-based read without changing the stream cursor. Thread-safe.""" + if hasattr(stream, 'pread'): + return stream.pread(length, offset) if hasattr(stream, 'read_at'): return stream.read_at(length, offset) return os.pread(stream.fileno(), length, offset) @@ -95,6 +98,15 @@ class FileIO(ABC): def new_input_stream(self, path: str): pass + def new_range_input_stream(self, path: str): + """Open a stream for shared positional range reads. + + Implementations may return a lower-level stream whose positional read + avoids cursor emulation. The default preserves third-party FileIO + compatibility. + """ + return self.new_input_stream(path) + @abstractmethod def new_output_stream(self, path: str): pass @@ -186,8 +198,9 @@ def read_ranges_coalesced(self, ranges, parallelism, max_gap=_COALESCE_GAP, max_span=_COALESCE_SPAN): """Read ``ranges`` (each ``None`` or ``(path, offset, length)``), returning bytes in the same order. Same-file nearby ranges are merged into one read - to cut round trips, then sliced; reads run on a thread pool. Negative - length (read to EOF) is read on its own, never merged. + to cut round trips, then sliced. All spans for one path share an input + stream; reads run on a thread pool. Negative length (read to EOF) is read + on its own, never merged. A failed read propagates and aborts the whole batch (unlike a per-row ``file.open()`` loop that fails one row at a time). @@ -232,36 +245,109 @@ def _read_ranges_coalesced(self, ranges, parallelism, max_gap, max_span, coalescible.append((index, path, offset, length)) spans = _coalesce_ranges(coalescible, max_gap, max_span) + tasks_by_path = {} + for span in spans: + tasks_by_path.setdefault(span[0], []).append(("span", span)) + for singleton in singletons: + tasks_by_path.setdefault(singleton[1], []).append( + ("one", singleton)) + tasks = [ + task for path_tasks in tasks_by_path.values() + for task in path_tasks + ] + if not tasks: + return results + + class _SharedRangeStream: + def __init__(self, file_io, path, task_count): + self._file_io = file_io + self._path = path + self._stream = None + self._open_lock = threading.Lock() + self._seek_lock = threading.Lock() + self._positional = None + self._remaining = task_count + + def _get(self): + if self._stream is None: + with self._open_lock: + if self._stream is None: + self._stream = self._file_io.new_range_input_stream( + self._path) + self._positional = supports_pread(self._stream) + return self._stream + + def read(self, offset, length): + stream = self._get() + if length >= 0 and self._positional: + return pread(stream, length, offset) + with self._seek_lock: + stream.seek(offset) + return stream.read() if length < 0 else stream.read(length) + + def release(self): + with self._open_lock: + self._remaining -= 1 + if self._remaining == 0: + self._close() + + def close(self): + with self._open_lock: + self._close() + + def _close(self): + if self._stream is not None: + self._stream.close() + self._stream = None + + streams = { + path: _SharedRangeStream(self, path, len(path_tasks)) + for path, path_tasks in tasks_by_path.items() + } + + def _read(path, offset, length): + try: + return streams[path].read(offset, length) + except Exception: + # Preserve the previous independent-open behavior as a retry + # when a shared stream becomes unusable. + return self.read_file_range(path, offset, length) def _run(task): kind, payload = task - if kind == "span": - path, span_off, span_len, members = payload - buf = self.read_file_range(path, span_off, span_len) - if return_views: - buf = memoryview(buf) - useful = sum(length for _, _, length in members) - share_buffer = ( - max_retained_amplification <= 0 - or span_len <= useful * max_retained_amplification - ) - for idx, off, length in members: - s = off - span_off - value = buf[s:s + length] - if return_views and not share_buffer: - value = memoryview(bytes(value)) - results[idx] = value - else: - idx, path, off, length = payload - result = self.read_file_range(path, off, length) - results[idx] = memoryview(result) if return_views else result + path = payload[0] if kind == "span" else payload[1] + try: + if kind == "span": + path, span_off, span_len, members = payload + buf = _read(path, span_off, span_len) + if return_views: + buf = memoryview(buf) + useful = sum(length for _, _, length in members) + share_buffer = ( + max_retained_amplification <= 0 + or span_len <= useful * max_retained_amplification + ) + for idx, off, length in members: + s = off - span_off + value = buf[s:s + length] + if return_views and not share_buffer: + value = memoryview(bytes(value)) + results[idx] = value + else: + idx, path, off, length = payload + result = _read(path, off, length) + results[idx] = ( + memoryview(result) if return_views else result) + finally: + streams[path].release() - tasks = [("span", s) for s in spans] + [("one", g) for g in singletons] - if not tasks: - return results workers = max(1, min(parallelism, len(tasks))) - with ThreadPoolExecutor(workers) as pool: - list(pool.map(_run, tasks)) + try: + with ThreadPoolExecutor(workers) as pool: + list(pool.map(_run, tasks)) + finally: + for stream in streams.values(): + stream.close() return results def read_blobs_concurrent(self, blobs, parallelism): diff --git a/paimon-python/pypaimon/filesystem/caching_file_io.py b/paimon-python/pypaimon/filesystem/caching_file_io.py index 8aa64e227f19..5be8e6421e5e 100644 --- a/paimon-python/pypaimon/filesystem/caching_file_io.py +++ b/paimon-python/pypaimon/filesystem/caching_file_io.py @@ -195,14 +195,16 @@ def put_file_size(self, file_path: str, size: int) -> None: class CachingInputStream: """Wraps a remote stream with block-level caching.""" - def __init__(self, file_io, file_path: str, cache): + def __init__(self, file_io, file_path: str, cache, range_reads=False): self._file_io = file_io self._stream = None self._file_path = file_path self._file_size = -1 self._cache = cache + self._range_reads = range_reads self._pos = 0 self._io_lock = threading.Lock() + self._stream_lock = threading.Lock() self._remote_supports_pread = None def _get_file_size(self) -> int: @@ -309,7 +311,14 @@ def _read_fully(self, stream, size: int) -> bytes: def _get_remote_stream(self): if self._stream is None: - self._stream = self._file_io.new_input_stream(self._file_path) + with self._stream_lock: + if self._stream is None: + if self._range_reads: + self._stream = self._file_io.new_range_input_stream( + self._file_path) + else: + self._stream = self._file_io.new_input_stream( + self._file_path) return self._stream def close(self): @@ -397,6 +406,13 @@ def new_input_stream(self, path: str): return self._delegate.new_input_stream(path) return CachingInputStream(self._delegate, path, self._cache) + def new_range_input_stream(self, path: str): + file_type = FileType.classify(path) + if self._cache is None or file_type not in self._whitelist or FileType.is_mutable(path): + return self._delegate.new_range_input_stream(path) + return CachingInputStream( + self._delegate, path, self._cache, range_reads=True) + def new_output_stream(self, path: str): return self._delegate.new_output_stream(path) diff --git a/paimon-python/pypaimon/filesystem/jindo_file_system_handler.py b/paimon-python/pypaimon/filesystem/jindo_file_system_handler.py index efbfe7f1899d..e375ec279c2f 100644 --- a/paimon-python/pypaimon/filesystem/jindo_file_system_handler.py +++ b/paimon-python/pypaimon/filesystem/jindo_file_system_handler.py @@ -328,6 +328,10 @@ def open_input_file(self, path: str): jindo_stream = self._jindo_fs.open(normalized, "rb") return PythonFile(JindoInputFile(jindo_stream), mode="r") + def open_range_input_stream(self, path: str): + """Open the native stream so callers can use Jindo ``pread`` directly.""" + return self._jindo_fs.open(self._normalize_path(path), "rb") + def open_output_stream(self, path: str, metadata): normalized = self._normalize_path(path) jindo_stream = self._jindo_fs.open(normalized, "wb") diff --git a/paimon-python/pypaimon/filesystem/pyarrow_file_io.py b/paimon-python/pypaimon/filesystem/pyarrow_file_io.py index 4423c9559e30..f171f8cf7e2e 100644 --- a/paimon-python/pypaimon/filesystem/pyarrow_file_io.py +++ b/paimon-python/pypaimon/filesystem/pyarrow_file_io.py @@ -358,6 +358,11 @@ def new_input_stream(self, path: str): path_str = self.to_filesystem_path(path) return self.filesystem.open_input_file(path_str) + def new_range_input_stream(self, path: str): + if self._use_jindo: + return self.filesystem.handler.open_range_input_stream(path) + return self.new_input_stream(path) + def new_output_stream(self, path: str): path_str = self.to_filesystem_path(path) diff --git a/paimon-python/pypaimon/filesystem/resolving_file_io.py b/paimon-python/pypaimon/filesystem/resolving_file_io.py index 576d160c8155..a1a1cecb1e4e 100644 --- a/paimon-python/pypaimon/filesystem/resolving_file_io.py +++ b/paimon-python/pypaimon/filesystem/resolving_file_io.py @@ -66,6 +66,9 @@ def is_object_store(self) -> bool: def new_input_stream(self, path: str): return self._get_fileio(path).new_input_stream(path) + def new_range_input_stream(self, path: str): + return self._get_fileio(path).new_range_input_stream(path) + def new_output_stream(self, path: str): return self._get_fileio(path).new_output_stream(path) diff --git a/paimon-python/pypaimon/tests/blob_test.py b/paimon-python/pypaimon/tests/blob_test.py index 19aacce9e044..50168b82d35c 100644 --- a/paimon-python/pypaimon/tests/blob_test.py +++ b/paimon-python/pypaimon/tests/blob_test.py @@ -22,6 +22,7 @@ import shutil import struct import tempfile +import time import unittest import zlib from decimal import Decimal @@ -2412,18 +2413,27 @@ def test_array_blob_parallelism_uses_concurrent_resolver(self): calls = [] range_reads = [] original_read = file_io.read_blobs_concurrent - original_range_read = file_io.read_file_range + original_open = file_io.new_range_input_stream def read_blobs_concurrent(blobs, parallelism): calls.append((list(blobs), parallelism)) return original_read(blobs, parallelism) - def read_file_range(path, offset, length): - range_reads.append((path, offset, length)) - return original_range_read(path, offset, length) + def new_range_input_stream(path): + stream = original_open(path) + + class TrackingStream: + def pread(self, length, offset): + range_reads.append((path, offset, length)) + return os.pread(stream.fileno(), length, offset) + + def close(self): + stream.close() + + return TrackingStream() file_io.read_blobs_concurrent = read_blobs_concurrent - file_io.read_file_range = read_file_range + file_io.new_range_input_stream = new_range_input_stream reader = FormatBlobReader( file_io=file_io, file_path=blob_file_path, @@ -3681,13 +3691,22 @@ def test_sparse_views_preserve_coalesced_read(self): output.write(data) file_io = FileIO.get(f"file://{tmp_dir}", {}) reads = [] - original_read = file_io.read_file_range + original_open = file_io.new_range_input_stream - def read_file_range(file_path, offset, length): - reads.append((file_path, offset, length)) - return original_read(file_path, offset, length) + def new_range_input_stream(file_path): + stream = original_open(file_path) - file_io.read_file_range = read_file_range + class TrackingStream: + def pread(self, length, offset): + reads.append((file_path, offset, length)) + return os.pread(stream.fileno(), length, offset) + + def close(self): + stream.close() + + return TrackingStream() + + file_io.new_range_input_stream = new_range_input_stream got = file_io.read_ranges_coalesced_views( [(path, 0, 10), (path, 1000, 10)], parallelism=4, @@ -3711,6 +3730,130 @@ def read_file_range(file_path, offset, length): self.assertEqual(reads, [(path, 0, 1010)]) self.assertIs(shared[0].obj, shared[1].obj) + def test_fetch_bodies_reuses_one_stream_for_same_uri(self): + from pypaimon.common.file_io import FileIO + from pypaimon.multimodal.query import ScanQuery + from pypaimon.table.row.blob import BlobDescriptor + + span_count = 64 + span_gap = 2 << 20 + payloads = [("blob-%02d" % i).encode() for i in range(span_count)] + with tempfile.TemporaryDirectory() as tmp_dir: + path = os.path.join(tmp_dir, "blobs.bin") + with open(path, "wb") as output: + for i, payload in enumerate(payloads): + output.seek(i * span_gap) + output.write(payload) + + file_io = FileIO.get(f"file://{tmp_dir}", {}) + original_open = file_io.new_range_input_stream + opens = [] + reads = [] + closes = [] + + def new_range_input_stream(file_path): + opens.append(file_path) + stream = original_open(file_path) + + class TrackingStream: + def pread(self, length, offset): + reads.append((offset, length)) + return os.pread(stream.fileno(), length, offset) + + def close(self): + closes.append(file_path) + stream.close() + + return TrackingStream() + + file_io.new_range_input_stream = new_range_input_stream + cells = [ + BlobDescriptor( + path, i * span_gap, len(payload)).serialize() + for i, payload in enumerate(payloads) + ] + + bodies = ScanQuery._fetch_bodies( + file_io, {"image": cells}, ["image"], parallelism=16) + + self.assertEqual(payloads, bodies["image"]) + self.assertEqual([path], opens) + self.assertEqual(span_count, len(reads)) + self.assertEqual([path], closes) + + def test_shared_stream_failure_reopens_range(self): + from pypaimon.common.file_io import FileIO + + data = bytes(range(64)) + with tempfile.TemporaryDirectory() as tmp_dir: + path = os.path.join(tmp_dir, "blob.bin") + with open(path, "wb") as output: + output.write(data) + file_io = FileIO.get(f"file://{tmp_dir}", {}) + fallbacks = [] + original_read = file_io.read_file_range + + class FailingStream: + def pread(self, length, offset): + raise IOError("shared stream failed") + + def close(self): + pass + + def read_file_range(file_path, offset, length): + fallbacks.append((file_path, offset, length)) + return original_read(file_path, offset, length) + + file_io.new_range_input_stream = lambda _: FailingStream() + file_io.read_file_range = read_file_range + ranges = [(path, 0, 4), (path, 16, 4)] + + self.assertEqual( + [data[0:4], data[16:20]], + file_io.read_ranges_coalesced( + ranges, parallelism=2, max_gap=0), + ) + self.assertEqual(2, len(fallbacks)) + + def test_shared_non_positional_stream_serializes_reads(self): + from pypaimon.common.file_io import FileIO + + data = bytes(range(128)) + file_io = FileIO.get("file:///tmp", {}) + + class SerialStream: + def __init__(self): + self.position = 0 + self.reading = False + + def seek(self, position): + self.position = position + + def read(self, length): + if self.reading: + raise AssertionError("non-positional reads overlapped") + self.reading = True + try: + time.sleep(0.001) + result = data[self.position:self.position + length] + self.position += len(result) + return result + finally: + self.reading = False + + def close(self): + pass + + stream = SerialStream() + file_io.new_range_input_stream = lambda _: stream + ranges = [("blob", i * 4, 2) for i in range(16)] + + self.assertEqual( + [data[i * 4:i * 4 + 2] for i in range(16)], + file_io.read_ranges_coalesced( + ranges, parallelism=8, max_gap=0), + ) + class ReadFileRangeTest(unittest.TestCase): """read_file_range must accept length == -1 (read to EOF) -- the valid diff --git a/paimon-python/pypaimon/tests/caching_file_io_test.py b/paimon-python/pypaimon/tests/caching_file_io_test.py index 12288ef1203b..ca5be3ba5b30 100644 --- a/paimon-python/pypaimon/tests/caching_file_io_test.py +++ b/paimon-python/pypaimon/tests/caching_file_io_test.py @@ -289,6 +289,21 @@ def test_manifest_file_is_cached(self): with caching_io.new_input_stream("manifest-abc") as s: self.assertEqual(data, s.read()) + def test_cached_range_stream_uses_delegate_range_stream(self): + data = b"manifest data" + delegate = self._make_delegate({"manifest-abc": data}) + delegate.new_range_input_stream.side_effect = ( + lambda path: io.BytesIO(data)) + cache = LocalDiskCacheManager( + self.cache_dir, 2 ** 63 - 1, block_size=64) + caching_io = CachingFileIO(delegate, cache) + + with caching_io.new_range_input_stream("manifest-abc") as stream: + self.assertEqual(data, stream.read_at(len(data), 0)) + + delegate.new_range_input_stream.assert_called_once_with("manifest-abc") + delegate.new_input_stream.assert_not_called() + def test_global_index_file_is_cached(self): data = b"index data" delegate = self._make_delegate({"global-index-uuid.index": data}) diff --git a/paimon-python/pypaimon/tests/jindo_file_system_test.py b/paimon-python/pypaimon/tests/jindo_file_system_test.py index 9bd45aeddaad..00ae1c6cc7fd 100644 --- a/paimon-python/pypaimon/tests/jindo_file_system_test.py +++ b/paimon-python/pypaimon/tests/jindo_file_system_test.py @@ -18,6 +18,7 @@ import os import unittest import uuid +from unittest.mock import MagicMock import pyarrow.fs as pafs @@ -25,6 +26,38 @@ from pypaimon.common.options import Options from pypaimon.common.options.config import OssOptions from pypaimon.filesystem.jindo_file_system_handler import JindoFileSystemHandler, JINDO_AVAILABLE +from pypaimon.filesystem.pyarrow_file_io import PyArrowFileIO + + +class JindoRangeInputStreamTest(unittest.TestCase): + + def test_returns_native_stream_for_pread(self): + handler = MagicMock() + handler._normalize_path.return_value = "oss://bucket/blob" + native_stream = object() + handler._jindo_fs.open.return_value = native_stream + + result = JindoFileSystemHandler.open_range_input_stream( + handler, "blob") + + self.assertIs(native_stream, result) + handler._jindo_fs.open.assert_called_once_with( + "oss://bucket/blob", "rb") + + def test_pyarrow_file_io_exposes_native_jindo_stream(self): + file_io = object.__new__(PyArrowFileIO) + file_io._use_jindo = True + file_io.filesystem = MagicMock() + native_stream = object() + file_io.filesystem.handler.open_range_input_stream.return_value = ( + native_stream) + + self.assertIs( + native_stream, + file_io.new_range_input_stream("oss://bucket/blob"), + ) + file_io.filesystem.handler.open_range_input_stream.assert_called_once_with( + "oss://bucket/blob") class JindoFileSystemTest(unittest.TestCase): diff --git a/paimon-python/pypaimon/tests/resolving_file_io_test.py b/paimon-python/pypaimon/tests/resolving_file_io_test.py index c3dec4ebe5c5..d0694eac7afe 100644 --- a/paimon-python/pypaimon/tests/resolving_file_io_test.py +++ b/paimon-python/pypaimon/tests/resolving_file_io_test.py @@ -19,6 +19,7 @@ import shutil import tempfile import unittest +from unittest.mock import MagicMock from pypaimon.common.file_io import FileIO from pypaimon.common.options import Options @@ -85,6 +86,16 @@ def test_is_object_store_with_local_warehouse(self): resolving = ResolvingFileIO(opts) self.assertFalse(resolving.is_object_store()) + def test_range_stream_is_forwarded(self): + resolving = ResolvingFileIO(Options({})) + delegate = MagicMock() + expected = object() + delegate.new_range_input_stream.return_value = expected + resolving._get_fileio = MagicMock(return_value=delegate) + + self.assertIs(expected, resolving.new_range_input_stream("file:///x")) + delegate.new_range_input_stream.assert_called_once_with("file:///x") + class ResolvingFileIOReadWriteTest(unittest.TestCase): """End-to-end read/write tests using ResolvingFileIO with local filesystem.""" diff --git a/paimon-python/pypaimon/tests/rest/rest_token_file_io_test.py b/paimon-python/pypaimon/tests/rest/rest_token_file_io_test.py index dbb919eed28b..78d929cdfd13 100644 --- a/paimon-python/pypaimon/tests/rest/rest_token_file_io_test.py +++ b/paimon-python/pypaimon/tests/rest/rest_token_file_io_test.py @@ -140,6 +140,24 @@ def test_new_output_stream_behavior_matches_parent(self): read_content = stream.read() self.assertEqual(read_content, test_content) + def test_range_stream_is_forwarded(self): + with patch.object(RESTTokenFileIO, 'try_to_refresh_token'): + file_io = RESTTokenFileIO( + self.identifier, + self.warehouse_path, + self.catalog_options, + ) + delegate = MagicMock() + expected = object() + delegate.new_range_input_stream.return_value = expected + with patch.object(file_io, 'file_io', return_value=delegate): + self.assertIs( + expected, + file_io.new_range_input_stream("file:///blob"), + ) + delegate.new_range_input_stream.assert_called_once_with( + "file:///blob") + def test_pickle_serialization(self): with patch.object(RESTTokenFileIO, 'try_to_refresh_token'): original_file_io = RESTTokenFileIO( From 5df758b587f55a419c118bb000ee952cd5cd4504 Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Sun, 9 Aug 2026 03:26:32 -0700 Subject: [PATCH 02/16] [python] Bound concurrent range stream reuse --- paimon-python/pypaimon/common/file_io.py | 154 ++++++++++++++---- .../pypaimon/filesystem/caching_file_io.py | 2 + .../filesystem/jindo_file_system_handler.py | 6 +- paimon-python/pypaimon/tests/blob_test.py | 62 ++++++- .../pypaimon/tests/jindo_file_system_test.py | 5 +- 5 files changed, 187 insertions(+), 42 deletions(-) diff --git a/paimon-python/pypaimon/common/file_io.py b/paimon-python/pypaimon/common/file_io.py index a45513d0fa1f..d1da4652cb28 100644 --- a/paimon-python/pypaimon/common/file_io.py +++ b/paimon-python/pypaimon/common/file_io.py @@ -30,7 +30,7 @@ def supports_pread(stream) -> bool: - """Check if the stream supports position-based reads (thread-safe I/O).""" + """Check if the stream supports position-based reads.""" if hasattr(stream, 'read_at') or hasattr(stream, 'pread'): return True if hasattr(stream, 'fileno'): @@ -43,7 +43,7 @@ def supports_pread(stream) -> bool: def pread(stream, length: int, offset: int) -> bytes: - """Position-based read without changing the stream cursor. Thread-safe.""" + """Position-based read without changing the stream cursor.""" if hasattr(stream, 'pread'): return stream.pread(length, offset) if hasattr(stream, 'read_at'): @@ -51,11 +51,17 @@ def pread(stream, length: int, offset: int) -> bytes: return os.pread(stream.fileno(), length, offset) +def supports_concurrent_pread(stream) -> bool: + return (supports_pread(stream) + and getattr(stream, 'supports_concurrent_pread', True)) + + # Coalescing bounds: merge same-file ranges whose gap is within GAP, capping a # merged read at SPAN so threads stay busy and memory stays bounded. _COALESCE_GAP = 1 << 20 _COALESCE_SPAN = 8 << 20 _COALESCE_VIEW_MAX_RETAINED_AMPLIFICATION = 2.0 +_MAX_EXCLUSIVE_RANGE_STREAMS = 16 def create_temp_path(path: str) -> str: @@ -258,50 +264,129 @@ def _read_ranges_coalesced(self, ranges, parallelism, max_gap, max_span, if not tasks: return results - class _SharedRangeStream: - def __init__(self, file_io, path, task_count): + workers = max(1, min(parallelism, len(tasks))) + + class _RangeStreamPool: + def __init__(self, file_io, path, task_count, max_streams): self._file_io = file_io self._path = path - self._stream = None - self._open_lock = threading.Lock() + self._condition = threading.Condition() + self._streams = [] + self._available = [] + self._opening = 0 + self._detecting = False + self._concurrent = None self._seek_lock = threading.Lock() - self._positional = None self._remaining = task_count - - def _get(self): - if self._stream is None: - with self._open_lock: - if self._stream is None: - self._stream = self._file_io.new_range_input_stream( - self._path) - self._positional = supports_pread(self._stream) - return self._stream + self._max_streams = min( + max_streams, task_count, _MAX_EXCLUSIVE_RANGE_STREAMS) + + def _open(self): + return self._file_io.new_range_input_stream(self._path) + + def _detect(self): + with self._condition: + while self._concurrent is None and self._detecting: + self._condition.wait() + if self._concurrent is not None: + return + self._detecting = True + try: + stream = self._open() + except Exception: + with self._condition: + self._detecting = False + self._condition.notify_all() + raise + with self._condition: + self._streams.append(stream) + self._concurrent = supports_concurrent_pread(stream) + if not self._concurrent: + self._available.append(stream) + self._detecting = False + self._condition.notify_all() + + def _acquire(self): + self._detect() + with self._condition: + if self._concurrent: + return self._streams[0], False + while True: + if self._available: + return self._available.pop(), True + if (len(self._streams) + self._opening + < self._max_streams): + self._opening += 1 + break + self._condition.wait() + try: + stream = self._open() + except Exception: + with self._condition: + self._opening -= 1 + self._condition.notify_all() + raise + with self._condition: + self._streams.append(stream) + self._opening -= 1 + self._condition.notify_all() + return stream, True + + def _return(self, stream): + with self._condition: + self._available.append(stream) + self._condition.notify() + + def _discard(self, stream): + try: + stream.close() + except Exception: + pass + with self._condition: + self._streams.remove(stream) + self._condition.notify_all() def read(self, offset, length): - stream = self._get() - if length >= 0 and self._positional: - return pread(stream, length, offset) - with self._seek_lock: + stream, exclusive = self._acquire() + failed = False + try: + if length >= 0 and supports_pread(stream): + return pread(stream, length, offset) + if not exclusive: + with self._seek_lock: + stream.seek(offset) + return (stream.read() if length < 0 + else stream.read(length)) stream.seek(offset) return stream.read() if length < 0 else stream.read(length) - - def release(self): - with self._open_lock: + except Exception: + failed = True + raise + finally: + if exclusive: + if failed: + self._discard(stream) + else: + self._return(stream) + + def task_done(self): + with self._condition: self._remaining -= 1 - if self._remaining == 0: - self._close() + close = self._remaining == 0 + if close: + self.close() def close(self): - with self._open_lock: - self._close() - - def _close(self): - if self._stream is not None: - self._stream.close() - self._stream = None + with self._condition: + streams = self._streams + self._streams = [] + self._available = [] + for stream in streams: + stream.close() streams = { - path: _SharedRangeStream(self, path, len(path_tasks)) + path: _RangeStreamPool( + self, path, len(path_tasks), workers) for path, path_tasks in tasks_by_path.items() } @@ -339,9 +424,8 @@ def _run(task): results[idx] = ( memoryview(result) if return_views else result) finally: - streams[path].release() + streams[path].task_done() - workers = max(1, min(parallelism, len(tasks))) try: with ThreadPoolExecutor(workers) as pool: list(pool.map(_run, tasks)) diff --git a/paimon-python/pypaimon/filesystem/caching_file_io.py b/paimon-python/pypaimon/filesystem/caching_file_io.py index 5be8e6421e5e..b6e31ca3b0ea 100644 --- a/paimon-python/pypaimon/filesystem/caching_file_io.py +++ b/paimon-python/pypaimon/filesystem/caching_file_io.py @@ -195,6 +195,8 @@ def put_file_size(self, file_path: str, size: int) -> None: class CachingInputStream: """Wraps a remote stream with block-level caching.""" + supports_concurrent_pread = False + def __init__(self, file_io, file_path: str, cache, range_reads=False): self._file_io = file_io self._stream = None diff --git a/paimon-python/pypaimon/filesystem/jindo_file_system_handler.py b/paimon-python/pypaimon/filesystem/jindo_file_system_handler.py index e375ec279c2f..039a69c7951b 100644 --- a/paimon-python/pypaimon/filesystem/jindo_file_system_handler.py +++ b/paimon-python/pypaimon/filesystem/jindo_file_system_handler.py @@ -114,6 +114,8 @@ def create_jindo_oss_filesystem(root_uri: str, catalog_options: Options): class JindoInputFile: + supports_concurrent_pread = False + def __init__(self, jindo_stream): self._stream = jindo_stream self._closed = False @@ -329,8 +331,8 @@ def open_input_file(self, path: str): return PythonFile(JindoInputFile(jindo_stream), mode="r") def open_range_input_stream(self, path: str): - """Open the native stream so callers can use Jindo ``pread`` directly.""" - return self._jindo_fs.open(self._normalize_path(path), "rb") + stream = self._jindo_fs.open(self._normalize_path(path), "rb") + return JindoInputFile(stream) def open_output_stream(self, path: str, metadata): normalized = self._normalize_path(path) diff --git a/paimon-python/pypaimon/tests/blob_test.py b/paimon-python/pypaimon/tests/blob_test.py index 50168b82d35c..f50f0aafd4f2 100644 --- a/paimon-python/pypaimon/tests/blob_test.py +++ b/paimon-python/pypaimon/tests/blob_test.py @@ -3815,7 +3815,7 @@ def read_file_range(file_path, offset, length): ) self.assertEqual(2, len(fallbacks)) - def test_shared_non_positional_stream_serializes_reads(self): + def test_non_positional_streams_are_exclusive(self): from pypaimon.common.file_io import FileIO data = bytes(range(128)) @@ -3844,8 +3844,14 @@ def read(self, length): def close(self): pass - stream = SerialStream() - file_io.new_range_input_stream = lambda _: stream + streams = [] + + def new_range_input_stream(_): + stream = SerialStream() + streams.append(stream) + return stream + + file_io.new_range_input_stream = new_range_input_stream ranges = [("blob", i * 4, 2) for i in range(16)] self.assertEqual( @@ -3853,6 +3859,56 @@ def close(self): file_io.read_ranges_coalesced( ranges, parallelism=8, max_gap=0), ) + self.assertGreater(len(streams), 1) + self.assertLessEqual(len(streams), 8) + + def test_non_thread_safe_pread_uses_bounded_stream_pool(self): + from pypaimon.common.file_io import FileIO + + data = bytes(range(256)) * 64 + file_io = FileIO.get("file:///tmp", {}) + streams = [] + + class PositionalStream: + supports_concurrent_pread = False + + def __init__(self): + self.reading = False + self.reads = 0 + self.closed = False + + def pread(self, length, offset): + if self.reading: + raise AssertionError("one stream was used concurrently") + self.reading = True + try: + time.sleep(0.01) + self.reads += 1 + return data[offset:offset + length] + finally: + self.reading = False + + def close(self): + self.closed = True + + def new_range_input_stream(_): + stream = PositionalStream() + streams.append(stream) + return stream + + file_io.new_range_input_stream = new_range_input_stream + ranges = [("blob", i * 128, 16) for i in range(64)] + + self.assertEqual( + [data[offset:offset + length] + for _, offset, length in ranges], + file_io.read_ranges_coalesced( + ranges, parallelism=32, max_gap=0), + ) + self.assertGreater(len(streams), 1) + self.assertLessEqual(len(streams), 16) + self.assertEqual(64, sum(stream.reads for stream in streams)) + self.assertTrue(all(stream.closed for stream in streams)) class ReadFileRangeTest(unittest.TestCase): diff --git a/paimon-python/pypaimon/tests/jindo_file_system_test.py b/paimon-python/pypaimon/tests/jindo_file_system_test.py index 00ae1c6cc7fd..ea059a4c7f59 100644 --- a/paimon-python/pypaimon/tests/jindo_file_system_test.py +++ b/paimon-python/pypaimon/tests/jindo_file_system_test.py @@ -31,7 +31,7 @@ class JindoRangeInputStreamTest(unittest.TestCase): - def test_returns_native_stream_for_pread(self): + def test_marks_native_stream_as_non_concurrent(self): handler = MagicMock() handler._normalize_path.return_value = "oss://bucket/blob" native_stream = object() @@ -40,7 +40,8 @@ def test_returns_native_stream_for_pread(self): result = JindoFileSystemHandler.open_range_input_stream( handler, "blob") - self.assertIs(native_stream, result) + self.assertIs(native_stream, result._stream) + self.assertFalse(result.supports_concurrent_pread) handler._jindo_fs.open.assert_called_once_with( "oss://bucket/blob", "rb") From 50588b26b47a3a08bb5e0a4d66ca236ff76f99b6 Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Sun, 9 Aug 2026 03:37:24 -0700 Subject: [PATCH 03/16] [python] Preserve concurrent positional range reads --- .../filesystem/hdfs_native_file_io.py | 3 ++ .../filesystem/jindo_file_system_handler.py | 5 +- .../pypaimon/tests/hdfs_native_test.py | 46 +++++++++++++++++++ .../pypaimon/tests/jindo_file_system_test.py | 5 +- 4 files changed, 52 insertions(+), 7 deletions(-) diff --git a/paimon-python/pypaimon/filesystem/hdfs_native_file_io.py b/paimon-python/pypaimon/filesystem/hdfs_native_file_io.py index 78cad2140e28..b5f479cba2fc 100644 --- a/paimon-python/pypaimon/filesystem/hdfs_native_file_io.py +++ b/paimon-python/pypaimon/filesystem/hdfs_native_file_io.py @@ -117,6 +117,9 @@ def read(self, size: int = -1) -> bytes: def read1(self, size: int = -1) -> bytes: return self.read(size) + def read_at(self, nbytes: int, offset: int) -> bytes: + return self._fr.read_range(offset, nbytes) + def seek(self, pos: int, whence: int = 0) -> int: self._fr.seek(pos, whence) return self._fr.tell() diff --git a/paimon-python/pypaimon/filesystem/jindo_file_system_handler.py b/paimon-python/pypaimon/filesystem/jindo_file_system_handler.py index 039a69c7951b..17e1ade2de69 100644 --- a/paimon-python/pypaimon/filesystem/jindo_file_system_handler.py +++ b/paimon-python/pypaimon/filesystem/jindo_file_system_handler.py @@ -114,8 +114,6 @@ def create_jindo_oss_filesystem(root_uri: str, catalog_options: Options): class JindoInputFile: - supports_concurrent_pread = False - def __init__(self, jindo_stream): self._stream = jindo_stream self._closed = False @@ -331,8 +329,7 @@ def open_input_file(self, path: str): return PythonFile(JindoInputFile(jindo_stream), mode="r") def open_range_input_stream(self, path: str): - stream = self._jindo_fs.open(self._normalize_path(path), "rb") - return JindoInputFile(stream) + return self._jindo_fs.open(self._normalize_path(path), "rb") def open_output_stream(self, path: str, metadata): normalized = self._normalize_path(path) diff --git a/paimon-python/pypaimon/tests/hdfs_native_test.py b/paimon-python/pypaimon/tests/hdfs_native_test.py index 957ee8f123dd..d82739183822 100644 --- a/paimon-python/pypaimon/tests/hdfs_native_test.py +++ b/paimon-python/pypaimon/tests/hdfs_native_test.py @@ -18,6 +18,8 @@ import os import sys import tempfile +import threading +import time import types import unittest from unittest.mock import MagicMock, patch @@ -462,6 +464,50 @@ def test_reader_adapter_read_negative_reads_all(self): self.assertEqual(adapter.read(), b"all-content") fr.read.assert_called_once_with(-1) + def test_reader_adapter_read_at(self): + from pypaimon.filesystem.hdfs_native_file_io import _HdfsReaderAdapter + fr = MagicMock() + fr.read_range.return_value = b"data" + adapter = _HdfsReaderAdapter(fr) + self.assertEqual(adapter.read_at(4, 7), b"data") + fr.read_range.assert_called_once_with(7, 4) + + def test_reader_adapter_read_at_is_concurrent(self): + from pypaimon.filesystem.hdfs_native_file_io import _HdfsReaderAdapter + + class RangeReader: + def __init__(self): + self.lock = threading.Lock() + self.active = 0 + self.max_active = 0 + + def read_range(self, offset, length): + with self.lock: + self.active += 1 + self.max_active = max(self.max_active, self.active) + try: + time.sleep(0.01) + return bytes([offset]) * length + finally: + with self.lock: + self.active -= 1 + + reader = RangeReader() + adapter = _HdfsReaderAdapter(reader) + file_io = FileIO.get("file:///tmp", {}) + opens = [] + file_io.new_range_input_stream = lambda path: ( + opens.append(path) or adapter) + ranges = [("hdfs://ns/blob", offset, 4) + for offset in range(0, 64, 8)] + results = file_io.read_ranges_coalesced( + ranges, parallelism=8, max_gap=0) + + self.assertEqual( + [bytes([offset]) * 4 for _, offset, _ in ranges], results) + self.assertEqual(["hdfs://ns/blob"], opens) + self.assertEqual(8, reader.max_active) + def test_reader_adapter_close_releases_underlying(self): from pypaimon.filesystem.hdfs_native_file_io import _HdfsReaderAdapter fr = MagicMock() diff --git a/paimon-python/pypaimon/tests/jindo_file_system_test.py b/paimon-python/pypaimon/tests/jindo_file_system_test.py index ea059a4c7f59..00ae1c6cc7fd 100644 --- a/paimon-python/pypaimon/tests/jindo_file_system_test.py +++ b/paimon-python/pypaimon/tests/jindo_file_system_test.py @@ -31,7 +31,7 @@ class JindoRangeInputStreamTest(unittest.TestCase): - def test_marks_native_stream_as_non_concurrent(self): + def test_returns_native_stream_for_pread(self): handler = MagicMock() handler._normalize_path.return_value = "oss://bucket/blob" native_stream = object() @@ -40,8 +40,7 @@ def test_marks_native_stream_as_non_concurrent(self): result = JindoFileSystemHandler.open_range_input_stream( handler, "blob") - self.assertIs(native_stream, result._stream) - self.assertFalse(result.supports_concurrent_pread) + self.assertIs(native_stream, result) handler._jindo_fs.open.assert_called_once_with( "oss://bucket/blob", "rb") From 7e692e3e10c95f936951ee81851e4d63f7f81ce9 Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Sun, 9 Aug 2026 04:21:38 -0700 Subject: [PATCH 04/16] [python] Guard concurrent range stream reuse --- paimon-python/pypaimon/common/file_io.py | 35 ++++++++++++------- .../filesystem/hdfs_native_file_io.py | 2 ++ .../filesystem/jindo_file_system_handler.py | 5 ++- paimon-python/pypaimon/tests/blob_test.py | 22 +++++++----- .../pypaimon/tests/jindo_file_system_test.py | 15 +++++--- 5 files changed, 53 insertions(+), 26 deletions(-) diff --git a/paimon-python/pypaimon/common/file_io.py b/paimon-python/pypaimon/common/file_io.py index d1da4652cb28..56c077b1df7a 100644 --- a/paimon-python/pypaimon/common/file_io.py +++ b/paimon-python/pypaimon/common/file_io.py @@ -29,31 +29,40 @@ from pypaimon.common.options import Options +def _fileno(stream): + if not hasattr(stream, 'fileno'): + return None + try: + return stream.fileno() + except Exception: + return None + + def supports_pread(stream) -> bool: """Check if the stream supports position-based reads.""" - if hasattr(stream, 'read_at') or hasattr(stream, 'pread'): + # Unlike read_at, Python pread methods have no common argument order. + if hasattr(stream, 'read_at'): return True - if hasattr(stream, 'fileno'): - try: - stream.fileno() - return True - except Exception: - pass - return False + return _fileno(stream) is not None def pread(stream, length: int, offset: int) -> bytes: """Position-based read without changing the stream cursor.""" - if hasattr(stream, 'pread'): - return stream.pread(length, offset) + fd = _fileno(stream) + if fd is not None: + return os.pread(fd, length, offset) if hasattr(stream, 'read_at'): return stream.read_at(length, offset) - return os.pread(stream.fileno(), length, offset) + raise AttributeError("stream does not support positional reads") def supports_concurrent_pread(stream) -> bool: - return (supports_pread(stream) - and getattr(stream, 'supports_concurrent_pread', True)) + if not supports_pread(stream): + return False + concurrent = getattr(stream, 'supports_concurrent_pread', None) + if concurrent is not None: + return bool(concurrent) + return _fileno(stream) is not None # Coalescing bounds: merge same-file ranges whose gap is within GAP, capping a diff --git a/paimon-python/pypaimon/filesystem/hdfs_native_file_io.py b/paimon-python/pypaimon/filesystem/hdfs_native_file_io.py index b5f479cba2fc..4ca3c9c77fbf 100644 --- a/paimon-python/pypaimon/filesystem/hdfs_native_file_io.py +++ b/paimon-python/pypaimon/filesystem/hdfs_native_file_io.py @@ -107,6 +107,8 @@ class _HdfsReaderAdapter: is closed — hdfs-native's own FileReader.__exit__ is a no-op. """ + supports_concurrent_pread = True + def __init__(self, fr): self._fr = fr self._closed = False diff --git a/paimon-python/pypaimon/filesystem/jindo_file_system_handler.py b/paimon-python/pypaimon/filesystem/jindo_file_system_handler.py index 17e1ade2de69..f1ebb5f10ce9 100644 --- a/paimon-python/pypaimon/filesystem/jindo_file_system_handler.py +++ b/paimon-python/pypaimon/filesystem/jindo_file_system_handler.py @@ -114,6 +114,8 @@ def create_jindo_oss_filesystem(root_uri: str, catalog_options: Options): class JindoInputFile: + supports_concurrent_pread = True + def __init__(self, jindo_stream): self._stream = jindo_stream self._closed = False @@ -329,7 +331,8 @@ def open_input_file(self, path: str): return PythonFile(JindoInputFile(jindo_stream), mode="r") def open_range_input_stream(self, path: str): - return self._jindo_fs.open(self._normalize_path(path), "rb") + stream = self._jindo_fs.open(self._normalize_path(path), "rb") + return JindoInputFile(stream) def open_output_stream(self, path: str, metadata): normalized = self._normalize_path(path) diff --git a/paimon-python/pypaimon/tests/blob_test.py b/paimon-python/pypaimon/tests/blob_test.py index f50f0aafd4f2..efb02a971e49 100644 --- a/paimon-python/pypaimon/tests/blob_test.py +++ b/paimon-python/pypaimon/tests/blob_test.py @@ -2423,7 +2423,9 @@ def new_range_input_stream(path): stream = original_open(path) class TrackingStream: - def pread(self, length, offset): + supports_concurrent_pread = True + + def read_at(self, length, offset): range_reads.append((path, offset, length)) return os.pread(stream.fileno(), length, offset) @@ -3697,7 +3699,9 @@ def new_range_input_stream(file_path): stream = original_open(file_path) class TrackingStream: - def pread(self, length, offset): + supports_concurrent_pread = True + + def read_at(self, length, offset): reads.append((file_path, offset, length)) return os.pread(stream.fileno(), length, offset) @@ -3756,7 +3760,9 @@ def new_range_input_stream(file_path): stream = original_open(file_path) class TrackingStream: - def pread(self, length, offset): + supports_concurrent_pread = True + + def read_at(self, length, offset): reads.append((offset, length)) return os.pread(stream.fileno(), length, offset) @@ -3794,7 +3800,9 @@ def test_shared_stream_failure_reopens_range(self): original_read = file_io.read_file_range class FailingStream: - def pread(self, length, offset): + supports_concurrent_pread = True + + def read_at(self, length, offset): raise IOError("shared stream failed") def close(self): @@ -3862,7 +3870,7 @@ def new_range_input_stream(_): self.assertGreater(len(streams), 1) self.assertLessEqual(len(streams), 8) - def test_non_thread_safe_pread_uses_bounded_stream_pool(self): + def test_unmarked_read_at_uses_bounded_stream_pool(self): from pypaimon.common.file_io import FileIO data = bytes(range(256)) * 64 @@ -3870,14 +3878,12 @@ def test_non_thread_safe_pread_uses_bounded_stream_pool(self): streams = [] class PositionalStream: - supports_concurrent_pread = False - def __init__(self): self.reading = False self.reads = 0 self.closed = False - def pread(self, length, offset): + def read_at(self, length, offset): if self.reading: raise AssertionError("one stream was used concurrently") self.reading = True diff --git a/paimon-python/pypaimon/tests/jindo_file_system_test.py b/paimon-python/pypaimon/tests/jindo_file_system_test.py index 00ae1c6cc7fd..2ae2a28e112e 100644 --- a/paimon-python/pypaimon/tests/jindo_file_system_test.py +++ b/paimon-python/pypaimon/tests/jindo_file_system_test.py @@ -25,22 +25,29 @@ from pyarrow.fs import PyFileSystem from pypaimon.common.options import Options from pypaimon.common.options.config import OssOptions -from pypaimon.filesystem.jindo_file_system_handler import JindoFileSystemHandler, JINDO_AVAILABLE +from pypaimon.filesystem.jindo_file_system_handler import ( + JindoFileSystemHandler, JindoInputFile, JINDO_AVAILABLE) from pypaimon.filesystem.pyarrow_file_io import PyArrowFileIO class JindoRangeInputStreamTest(unittest.TestCase): - def test_returns_native_stream_for_pread(self): + def test_wraps_native_stream_for_concurrent_pread(self): handler = MagicMock() handler._normalize_path.return_value = "oss://bucket/blob" - native_stream = object() + native_stream = MagicMock() + native_stream.closed = False + native_stream.pread.return_value = b"data" handler._jindo_fs.open.return_value = native_stream result = JindoFileSystemHandler.open_range_input_stream( handler, "blob") - self.assertIs(native_stream, result) + self.assertIsInstance(result, JindoInputFile) + self.assertIs(native_stream, result._stream) + self.assertTrue(result.supports_concurrent_pread) + self.assertEqual(b"data", result.read_at(4, 7)) + native_stream.pread.assert_called_once_with(4, 7) handler._jindo_fs.open.assert_called_once_with( "oss://bucket/blob", "rb") From 0ee000dc12e41a9b69c39c1ed0c6e32d20575cab Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Sun, 9 Aug 2026 05:25:57 -0700 Subject: [PATCH 05/16] [python] Address shared range stream review --- .../catalog/rest/rest_token_file_io.py | 3 - paimon-python/pypaimon/common/file_io.py | 60 ++++++++--- .../pypaimon/filesystem/caching_file_io.py | 22 +--- .../filesystem/hdfs_native_file_io.py | 7 +- .../filesystem/jindo_file_system_handler.py | 63 ++++++----- .../pypaimon/filesystem/pyarrow_file_io.py | 7 +- .../pypaimon/filesystem/resolving_file_io.py | 3 - paimon-python/pypaimon/tests/blob_test.py | 100 +++++++++++++++--- .../pypaimon/tests/caching_file_io_test.py | 15 --- .../pypaimon/tests/hdfs_native_test.py | 20 +++- .../pypaimon/tests/jindo_file_system_test.py | 48 ++++++--- .../pypaimon/tests/resolving_file_io_test.py | 11 -- .../tests/rest/rest_token_file_io_test.py | 18 ---- 13 files changed, 228 insertions(+), 149 deletions(-) diff --git a/paimon-python/pypaimon/catalog/rest/rest_token_file_io.py b/paimon-python/pypaimon/catalog/rest/rest_token_file_io.py index c89455f64fec..42dabb268cb2 100644 --- a/paimon-python/pypaimon/catalog/rest/rest_token_file_io.py +++ b/paimon-python/pypaimon/catalog/rest/rest_token_file_io.py @@ -143,9 +143,6 @@ def _merge_token_with_catalog_options(self, token: dict) -> dict: def new_input_stream(self, path: str): return self.file_io().new_input_stream(path) - def new_range_input_stream(self, path: str): - return self.file_io().new_range_input_stream(path) - def new_output_stream(self, path: str): return self.file_io().new_output_stream(path) diff --git a/paimon-python/pypaimon/common/file_io.py b/paimon-python/pypaimon/common/file_io.py index 56c077b1df7a..b8f1265c914f 100644 --- a/paimon-python/pypaimon/common/file_io.py +++ b/paimon-python/pypaimon/common/file_io.py @@ -28,6 +28,8 @@ from pypaimon.common.options import Options +_LOG = logging.getLogger(__name__) + def _fileno(stream): if not hasattr(stream, 'fileno'): @@ -113,15 +115,6 @@ class FileIO(ABC): def new_input_stream(self, path: str): pass - def new_range_input_stream(self, path: str): - """Open a stream for shared positional range reads. - - Implementations may return a lower-level stream whose positional read - avoids cursor emulation. The default preserves third-party FileIO - compatibility. - """ - return self.new_input_stream(path) - @abstractmethod def new_output_stream(self, path: str): pass @@ -289,9 +282,10 @@ def __init__(self, file_io, path, task_count, max_streams): self._remaining = task_count self._max_streams = min( max_streams, task_count, _MAX_EXCLUSIVE_RANGE_STREAMS) + self._close_error = None def _open(self): - return self._file_io.new_range_input_stream(self._path) + return self._file_io.new_input_stream(self._path) def _detect(self): with self._condition: @@ -300,16 +294,23 @@ def _detect(self): if self._concurrent is not None: return self._detecting = True + stream = None try: stream = self._open() - except Exception: + concurrent = supports_concurrent_pread(stream) + except BaseException: + if stream is not None: + try: + stream.close() + except Exception: + pass with self._condition: self._detecting = False self._condition.notify_all() raise with self._condition: self._streams.append(stream) - self._concurrent = supports_concurrent_pread(stream) + self._concurrent = concurrent if not self._concurrent: self._available.append(stream) self._detecting = False @@ -383,15 +384,24 @@ def task_done(self): self._remaining -= 1 close = self._remaining == 0 if close: - self.close() + self._close_all() - def close(self): + def _close_all(self): with self._condition: streams = self._streams self._streams = [] self._available = [] for stream in streams: - stream.close() + try: + stream.close() + except BaseException as error: + if self._close_error is None: + self._close_error = error + + def close(self): + self._close_all() + if self._close_error is not None: + raise self._close_error streams = { path: _RangeStreamPool( @@ -435,12 +445,30 @@ def _run(task): finally: streams[path].task_done() + failed = False try: with ThreadPoolExecutor(workers) as pool: list(pool.map(_run, tasks)) + except BaseException: + failed = True + raise finally: + close_error = None for stream in streams.values(): - stream.close() + try: + stream.close() + except BaseException as error: + if close_error is None: + close_error = error + if close_error is not None: + if failed: + _LOG.warning( + "Failed to close a range input stream", + exc_info=(type(close_error), close_error, + close_error.__traceback__), + ) + else: + raise close_error return results def read_blobs_concurrent(self, blobs, parallelism): diff --git a/paimon-python/pypaimon/filesystem/caching_file_io.py b/paimon-python/pypaimon/filesystem/caching_file_io.py index b6e31ca3b0ea..8aa64e227f19 100644 --- a/paimon-python/pypaimon/filesystem/caching_file_io.py +++ b/paimon-python/pypaimon/filesystem/caching_file_io.py @@ -195,18 +195,14 @@ def put_file_size(self, file_path: str, size: int) -> None: class CachingInputStream: """Wraps a remote stream with block-level caching.""" - supports_concurrent_pread = False - - def __init__(self, file_io, file_path: str, cache, range_reads=False): + def __init__(self, file_io, file_path: str, cache): self._file_io = file_io self._stream = None self._file_path = file_path self._file_size = -1 self._cache = cache - self._range_reads = range_reads self._pos = 0 self._io_lock = threading.Lock() - self._stream_lock = threading.Lock() self._remote_supports_pread = None def _get_file_size(self) -> int: @@ -313,14 +309,7 @@ def _read_fully(self, stream, size: int) -> bytes: def _get_remote_stream(self): if self._stream is None: - with self._stream_lock: - if self._stream is None: - if self._range_reads: - self._stream = self._file_io.new_range_input_stream( - self._file_path) - else: - self._stream = self._file_io.new_input_stream( - self._file_path) + self._stream = self._file_io.new_input_stream(self._file_path) return self._stream def close(self): @@ -408,13 +397,6 @@ def new_input_stream(self, path: str): return self._delegate.new_input_stream(path) return CachingInputStream(self._delegate, path, self._cache) - def new_range_input_stream(self, path: str): - file_type = FileType.classify(path) - if self._cache is None or file_type not in self._whitelist or FileType.is_mutable(path): - return self._delegate.new_range_input_stream(path) - return CachingInputStream( - self._delegate, path, self._cache, range_reads=True) - def new_output_stream(self, path: str): return self._delegate.new_output_stream(path) diff --git a/paimon-python/pypaimon/filesystem/hdfs_native_file_io.py b/paimon-python/pypaimon/filesystem/hdfs_native_file_io.py index 4ca3c9c77fbf..d61157d49f95 100644 --- a/paimon-python/pypaimon/filesystem/hdfs_native_file_io.py +++ b/paimon-python/pypaimon/filesystem/hdfs_native_file_io.py @@ -120,7 +120,12 @@ def read1(self, size: int = -1) -> bytes: return self.read(size) def read_at(self, nbytes: int, offset: int) -> bytes: - return self._fr.read_range(offset, nbytes) + if offset < 0: + raise ValueError("offset must be non-negative") + file_size = len(self._fr) + if nbytes <= 0 or offset >= file_size: + return b'' + return self._fr.read_range(offset, min(nbytes, file_size - offset)) def seek(self, pos: int, whence: int = 0) -> int: self._fr.seek(pos, whence) diff --git a/paimon-python/pypaimon/filesystem/jindo_file_system_handler.py b/paimon-python/pypaimon/filesystem/jindo_file_system_handler.py index f1ebb5f10ce9..500c5b0a919d 100644 --- a/paimon-python/pypaimon/filesystem/jindo_file_system_handler.py +++ b/paimon-python/pypaimon/filesystem/jindo_file_system_handler.py @@ -16,6 +16,7 @@ # under the License. import logging +import threading import pyarrow as pa from pyarrow import PythonFile @@ -116,42 +117,51 @@ def create_jindo_oss_filesystem(root_uri: str, catalog_options: Options): class JindoInputFile: supports_concurrent_pread = True - def __init__(self, jindo_stream): - self._stream = jindo_stream + def __init__(self, stream_factory): + self._stream_factory = stream_factory + self._stream = None + self._lock = threading.Lock() self._closed = False @property def closed(self): - if hasattr(self._stream, 'closed'): - return self._stream.closed return self._closed - def read(self, nbytes: int = -1): - if self.closed: + def _get_stream(self): + if self._closed: raise ValueError("I/O operation on closed file") + if self._stream is None: + with self._lock: + if self._closed: + raise ValueError("I/O operation on closed file") + if self._stream is None: + self._stream = self._stream_factory() + return self._stream + + def read(self, nbytes: int = -1): + stream = self._get_stream() if nbytes is None or nbytes < 0: - return self._stream.read() - return self._stream.read(nbytes) + return stream.read() + return stream.read(nbytes) def seek(self, position: int, whence: int = 0): - if self.closed: - raise ValueError("I/O operation on closed file") - self._stream.seek(position, whence) + return self._get_stream().seek(position, whence) def tell(self) -> int: - if self.closed: - raise ValueError("I/O operation on closed file") - return self._stream.tell() + return self._get_stream().tell() def read_at(self, nbytes: int, offset: int): - if self.closed: - raise ValueError("I/O operation on closed file") - return self._stream.pread(nbytes, offset) + return self._get_stream().pread(nbytes, offset) def close(self): - if not self._closed: - self._stream.close() + with self._lock: + if self._closed: + return self._closed = True + stream = self._stream + self._stream = None + if stream is not None: + stream.close() def __enter__(self): return self @@ -321,18 +331,15 @@ def copy_file(self, src: str, dest: str): self._jindo_fs.copy_file(src_norm, dst_norm) def open_input_stream(self, path: str): - normalized = self._normalize_path(path) - jindo_stream = self._jindo_fs.open(normalized, "rb") - return PythonFile(JindoInputFile(jindo_stream), mode="r") + return PythonFile(self.new_input_stream(path), mode="r") def open_input_file(self, path: str): - normalized = self._normalize_path(path) - jindo_stream = self._jindo_fs.open(normalized, "rb") - return PythonFile(JindoInputFile(jindo_stream), mode="r") + return PythonFile(self.new_input_stream(path), mode="r") - def open_range_input_stream(self, path: str): - stream = self._jindo_fs.open(self._normalize_path(path), "rb") - return JindoInputFile(stream) + def new_input_stream(self, path: str): + normalized = self._normalize_path(path) + return JindoInputFile( + lambda: self._jindo_fs.open(normalized, "rb")) def open_output_stream(self, path: str, metadata): normalized = self._normalize_path(path) diff --git a/paimon-python/pypaimon/filesystem/pyarrow_file_io.py b/paimon-python/pypaimon/filesystem/pyarrow_file_io.py index f171f8cf7e2e..464c5c02c762 100644 --- a/paimon-python/pypaimon/filesystem/pyarrow_file_io.py +++ b/paimon-python/pypaimon/filesystem/pyarrow_file_io.py @@ -355,14 +355,11 @@ def _get_ticket_cache_path() -> Optional[str]: return _kerberos.get_ticket_cache_path() def new_input_stream(self, path: str): + if self._use_jindo: + return self.filesystem.handler.new_input_stream(path) path_str = self.to_filesystem_path(path) return self.filesystem.open_input_file(path_str) - def new_range_input_stream(self, path: str): - if self._use_jindo: - return self.filesystem.handler.open_range_input_stream(path) - return self.new_input_stream(path) - def new_output_stream(self, path: str): path_str = self.to_filesystem_path(path) diff --git a/paimon-python/pypaimon/filesystem/resolving_file_io.py b/paimon-python/pypaimon/filesystem/resolving_file_io.py index a1a1cecb1e4e..576d160c8155 100644 --- a/paimon-python/pypaimon/filesystem/resolving_file_io.py +++ b/paimon-python/pypaimon/filesystem/resolving_file_io.py @@ -66,9 +66,6 @@ def is_object_store(self) -> bool: def new_input_stream(self, path: str): return self._get_fileio(path).new_input_stream(path) - def new_range_input_stream(self, path: str): - return self._get_fileio(path).new_range_input_stream(path) - def new_output_stream(self, path: str): return self._get_fileio(path).new_output_stream(path) diff --git a/paimon-python/pypaimon/tests/blob_test.py b/paimon-python/pypaimon/tests/blob_test.py index efb02a971e49..9439ca2b50f5 100644 --- a/paimon-python/pypaimon/tests/blob_test.py +++ b/paimon-python/pypaimon/tests/blob_test.py @@ -2413,18 +2413,27 @@ def test_array_blob_parallelism_uses_concurrent_resolver(self): calls = [] range_reads = [] original_read = file_io.read_blobs_concurrent - original_open = file_io.new_range_input_stream + original_open = file_io.new_input_stream def read_blobs_concurrent(blobs, parallelism): calls.append((list(blobs), parallelism)) return original_read(blobs, parallelism) - def new_range_input_stream(path): + def new_input_stream(path): stream = original_open(path) class TrackingStream: supports_concurrent_pread = True + def read(self, length=-1): + return stream.read(length) + + def seek(self, offset, whence=0): + return stream.seek(offset, whence) + + def tell(self): + return stream.tell() + def read_at(self, length, offset): range_reads.append((path, offset, length)) return os.pread(stream.fileno(), length, offset) @@ -2435,7 +2444,7 @@ def close(self): return TrackingStream() file_io.read_blobs_concurrent = read_blobs_concurrent - file_io.new_range_input_stream = new_range_input_stream + file_io.new_input_stream = new_input_stream reader = FormatBlobReader( file_io=file_io, file_path=blob_file_path, @@ -3693,9 +3702,9 @@ def test_sparse_views_preserve_coalesced_read(self): output.write(data) file_io = FileIO.get(f"file://{tmp_dir}", {}) reads = [] - original_open = file_io.new_range_input_stream + original_open = file_io.new_input_stream - def new_range_input_stream(file_path): + def new_input_stream(file_path): stream = original_open(file_path) class TrackingStream: @@ -3710,7 +3719,7 @@ def close(self): return TrackingStream() - file_io.new_range_input_stream = new_range_input_stream + file_io.new_input_stream = new_input_stream got = file_io.read_ranges_coalesced_views( [(path, 0, 10), (path, 1000, 10)], parallelism=4, @@ -3750,12 +3759,12 @@ def test_fetch_bodies_reuses_one_stream_for_same_uri(self): output.write(payload) file_io = FileIO.get(f"file://{tmp_dir}", {}) - original_open = file_io.new_range_input_stream + original_open = file_io.new_input_stream opens = [] reads = [] closes = [] - def new_range_input_stream(file_path): + def new_input_stream(file_path): opens.append(file_path) stream = original_open(file_path) @@ -3772,7 +3781,7 @@ def close(self): return TrackingStream() - file_io.new_range_input_stream = new_range_input_stream + file_io.new_input_stream = new_input_stream cells = [ BlobDescriptor( path, i * span_gap, len(payload)).serialize() @@ -3797,7 +3806,6 @@ def test_shared_stream_failure_reopens_range(self): output.write(data) file_io = FileIO.get(f"file://{tmp_dir}", {}) fallbacks = [] - original_read = file_io.read_file_range class FailingStream: supports_concurrent_pread = True @@ -3810,9 +3818,9 @@ def close(self): def read_file_range(file_path, offset, length): fallbacks.append((file_path, offset, length)) - return original_read(file_path, offset, length) + return data[offset:offset + length] - file_io.new_range_input_stream = lambda _: FailingStream() + file_io.new_input_stream = lambda _: FailingStream() file_io.read_file_range = read_file_range ranges = [(path, 0, 4), (path, 16, 4)] @@ -3854,12 +3862,12 @@ def close(self): streams = [] - def new_range_input_stream(_): + def new_input_stream(_): stream = SerialStream() streams.append(stream) return stream - file_io.new_range_input_stream = new_range_input_stream + file_io.new_input_stream = new_input_stream ranges = [("blob", i * 4, 2) for i in range(16)] self.assertEqual( @@ -3897,12 +3905,12 @@ def read_at(self, length, offset): def close(self): self.closed = True - def new_range_input_stream(_): + def new_input_stream(_): stream = PositionalStream() streams.append(stream) return stream - file_io.new_range_input_stream = new_range_input_stream + file_io.new_input_stream = new_input_stream ranges = [("blob", i * 128, 16) for i in range(64)] self.assertEqual( @@ -3916,6 +3924,66 @@ def new_range_input_stream(_): self.assertEqual(64, sum(stream.reads for stream in streams)) self.assertTrue(all(stream.closed for stream in streams)) + def test_closes_all_streams_before_raising_close_error(self): + from pypaimon.common.file_io import FileIO + + data = bytes(range(128)) + file_io = FileIO.get("file:///tmp", {}) + streams = [] + + class CloseStream: + def __init__(self, index): + self.index = index + self.closed = False + + def read_at(self, length, offset): + time.sleep(0.01) + return data[offset:offset + length] + + def close(self): + self.closed = True + if self.index == 0: + raise IOError("first close failed") + + def new_input_stream(_): + stream = CloseStream(len(streams)) + streams.append(stream) + return stream + + file_io.new_input_stream = new_input_stream + ranges = [("blob", offset, 4) for offset in range(0, 64, 8)] + + with self.assertRaisesRegex(IOError, "first close failed"): + file_io.read_ranges_coalesced( + ranges, parallelism=4, max_gap=0) + + self.assertGreater(len(streams), 1) + self.assertTrue(all(stream.closed for stream in streams)) + + def test_close_error_does_not_mask_read_error(self): + from pypaimon.common.file_io import FileIO + + file_io = FileIO.get("file:///tmp", {}) + + class FailingStream: + supports_concurrent_pread = True + + def read_at(self, length, offset): + raise IOError("shared read failed") + + def close(self): + raise IOError("close failed") + + def fail_fallback(path, offset, length): + raise IOError("fallback read failed") + + file_io.new_input_stream = lambda _: FailingStream() + file_io.read_file_range = fail_fallback + + with self.assertRaisesRegex(IOError, "fallback read failed"): + file_io.read_ranges_coalesced( + [("blob", 0, 4)], parallelism=1, max_gap=0) + class ReadFileRangeTest(unittest.TestCase): """read_file_range must accept length == -1 (read to EOF) -- the valid diff --git a/paimon-python/pypaimon/tests/caching_file_io_test.py b/paimon-python/pypaimon/tests/caching_file_io_test.py index ca5be3ba5b30..12288ef1203b 100644 --- a/paimon-python/pypaimon/tests/caching_file_io_test.py +++ b/paimon-python/pypaimon/tests/caching_file_io_test.py @@ -289,21 +289,6 @@ def test_manifest_file_is_cached(self): with caching_io.new_input_stream("manifest-abc") as s: self.assertEqual(data, s.read()) - def test_cached_range_stream_uses_delegate_range_stream(self): - data = b"manifest data" - delegate = self._make_delegate({"manifest-abc": data}) - delegate.new_range_input_stream.side_effect = ( - lambda path: io.BytesIO(data)) - cache = LocalDiskCacheManager( - self.cache_dir, 2 ** 63 - 1, block_size=64) - caching_io = CachingFileIO(delegate, cache) - - with caching_io.new_range_input_stream("manifest-abc") as stream: - self.assertEqual(data, stream.read_at(len(data), 0)) - - delegate.new_range_input_stream.assert_called_once_with("manifest-abc") - delegate.new_input_stream.assert_not_called() - def test_global_index_file_is_cached(self): data = b"index data" delegate = self._make_delegate({"global-index-uuid.index": data}) diff --git a/paimon-python/pypaimon/tests/hdfs_native_test.py b/paimon-python/pypaimon/tests/hdfs_native_test.py index d82739183822..7e5b9e33d0d6 100644 --- a/paimon-python/pypaimon/tests/hdfs_native_test.py +++ b/paimon-python/pypaimon/tests/hdfs_native_test.py @@ -467,11 +467,26 @@ def test_reader_adapter_read_negative_reads_all(self): def test_reader_adapter_read_at(self): from pypaimon.filesystem.hdfs_native_file_io import _HdfsReaderAdapter fr = MagicMock() + fr.__len__.return_value = 100 fr.read_range.return_value = b"data" adapter = _HdfsReaderAdapter(fr) self.assertEqual(adapter.read_at(4, 7), b"data") fr.read_range.assert_called_once_with(7, 4) + def test_reader_adapter_read_at_clamps_to_eof(self): + from pypaimon.filesystem.hdfs_native_file_io import _HdfsReaderAdapter + fr = MagicMock() + fr.__len__.return_value = 10 + fr.read_range.return_value = b"89" + adapter = _HdfsReaderAdapter(fr) + + self.assertEqual(adapter.read_at(4, 8), b"89") + fr.read_range.assert_called_once_with(8, 2) + self.assertEqual(adapter.read_at(4, 10), b"") + self.assertEqual(1, fr.read_range.call_count) + with self.assertRaisesRegex(ValueError, "non-negative"): + adapter.read_at(1, -1) + def test_reader_adapter_read_at_is_concurrent(self): from pypaimon.filesystem.hdfs_native_file_io import _HdfsReaderAdapter @@ -481,6 +496,9 @@ def __init__(self): self.active = 0 self.max_active = 0 + def __len__(self): + return 64 + def read_range(self, offset, length): with self.lock: self.active += 1 @@ -496,7 +514,7 @@ def read_range(self, offset, length): adapter = _HdfsReaderAdapter(reader) file_io = FileIO.get("file:///tmp", {}) opens = [] - file_io.new_range_input_stream = lambda path: ( + file_io.new_input_stream = lambda path: ( opens.append(path) or adapter) ranges = [("hdfs://ns/blob", offset, 4) for offset in range(0, 64, 8)] diff --git a/paimon-python/pypaimon/tests/jindo_file_system_test.py b/paimon-python/pypaimon/tests/jindo_file_system_test.py index 2ae2a28e112e..f15c61c5ce64 100644 --- a/paimon-python/pypaimon/tests/jindo_file_system_test.py +++ b/paimon-python/pypaimon/tests/jindo_file_system_test.py @@ -30,40 +30,64 @@ from pypaimon.filesystem.pyarrow_file_io import PyArrowFileIO -class JindoRangeInputStreamTest(unittest.TestCase): +class JindoInputStreamTest(unittest.TestCase): - def test_wraps_native_stream_for_concurrent_pread(self): + def test_native_stream_is_opened_lazily_for_pread(self): handler = MagicMock() handler._normalize_path.return_value = "oss://bucket/blob" native_stream = MagicMock() - native_stream.closed = False native_stream.pread.return_value = b"data" handler._jindo_fs.open.return_value = native_stream - result = JindoFileSystemHandler.open_range_input_stream( + result = JindoFileSystemHandler.new_input_stream( handler, "blob") self.assertIsInstance(result, JindoInputFile) - self.assertIs(native_stream, result._stream) self.assertTrue(result.supports_concurrent_pread) + handler._jindo_fs.open.assert_not_called() self.assertEqual(b"data", result.read_at(4, 7)) native_stream.pread.assert_called_once_with(4, 7) handler._jindo_fs.open.assert_called_once_with( "oss://bucket/blob", "rb") + result.close() + native_stream.close.assert_called_once() - def test_pyarrow_file_io_exposes_native_jindo_stream(self): + def test_regular_stream_methods_use_one_lazy_native_stream(self): + native_stream = MagicMock() + native_stream.read.return_value = b"data" + native_stream.tell.return_value = 4 + stream = JindoInputFile(lambda: native_stream) + + self.assertEqual(b"data", stream.read(4)) + stream.seek(2) + self.assertEqual(4, stream.tell()) + + native_stream.read.assert_called_once_with(4) + native_stream.seek.assert_called_once_with(2, 0) + native_stream.tell.assert_called_once() + + def test_close_before_read_does_not_open_native_stream(self): + stream_factory = MagicMock() + stream = JindoInputFile(stream_factory) + + stream.close() + + stream_factory.assert_not_called() + with self.assertRaisesRegex(ValueError, "closed file"): + stream.read(1) + + def test_pyarrow_file_io_exposes_jindo_adapter_directly(self): file_io = object.__new__(PyArrowFileIO) file_io._use_jindo = True file_io.filesystem = MagicMock() - native_stream = object() - file_io.filesystem.handler.open_range_input_stream.return_value = ( - native_stream) + stream = object() + file_io.filesystem.handler.new_input_stream.return_value = stream self.assertIs( - native_stream, - file_io.new_range_input_stream("oss://bucket/blob"), + stream, + file_io.new_input_stream("oss://bucket/blob"), ) - file_io.filesystem.handler.open_range_input_stream.assert_called_once_with( + file_io.filesystem.handler.new_input_stream.assert_called_once_with( "oss://bucket/blob") diff --git a/paimon-python/pypaimon/tests/resolving_file_io_test.py b/paimon-python/pypaimon/tests/resolving_file_io_test.py index d0694eac7afe..c3dec4ebe5c5 100644 --- a/paimon-python/pypaimon/tests/resolving_file_io_test.py +++ b/paimon-python/pypaimon/tests/resolving_file_io_test.py @@ -19,7 +19,6 @@ import shutil import tempfile import unittest -from unittest.mock import MagicMock from pypaimon.common.file_io import FileIO from pypaimon.common.options import Options @@ -86,16 +85,6 @@ def test_is_object_store_with_local_warehouse(self): resolving = ResolvingFileIO(opts) self.assertFalse(resolving.is_object_store()) - def test_range_stream_is_forwarded(self): - resolving = ResolvingFileIO(Options({})) - delegate = MagicMock() - expected = object() - delegate.new_range_input_stream.return_value = expected - resolving._get_fileio = MagicMock(return_value=delegate) - - self.assertIs(expected, resolving.new_range_input_stream("file:///x")) - delegate.new_range_input_stream.assert_called_once_with("file:///x") - class ResolvingFileIOReadWriteTest(unittest.TestCase): """End-to-end read/write tests using ResolvingFileIO with local filesystem.""" diff --git a/paimon-python/pypaimon/tests/rest/rest_token_file_io_test.py b/paimon-python/pypaimon/tests/rest/rest_token_file_io_test.py index 78d929cdfd13..dbb919eed28b 100644 --- a/paimon-python/pypaimon/tests/rest/rest_token_file_io_test.py +++ b/paimon-python/pypaimon/tests/rest/rest_token_file_io_test.py @@ -140,24 +140,6 @@ def test_new_output_stream_behavior_matches_parent(self): read_content = stream.read() self.assertEqual(read_content, test_content) - def test_range_stream_is_forwarded(self): - with patch.object(RESTTokenFileIO, 'try_to_refresh_token'): - file_io = RESTTokenFileIO( - self.identifier, - self.warehouse_path, - self.catalog_options, - ) - delegate = MagicMock() - expected = object() - delegate.new_range_input_stream.return_value = expected - with patch.object(file_io, 'file_io', return_value=delegate): - self.assertIs( - expected, - file_io.new_range_input_stream("file:///blob"), - ) - delegate.new_range_input_stream.assert_called_once_with( - "file:///blob") - def test_pickle_serialization(self): with patch.object(RESTTokenFileIO, 'try_to_refresh_token'): original_file_io = RESTTokenFileIO( From eaceac6e0accc5177a04191138ede6b0704aeb32 Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Sun, 9 Aug 2026 05:28:09 -0700 Subject: [PATCH 06/16] [python] Use exclusive Jindo range streams --- .../pypaimon/filesystem/jindo_file_system_handler.py | 3 ++- paimon-python/pypaimon/tests/blob_test.py | 4 +++- paimon-python/pypaimon/tests/jindo_file_system_test.py | 2 +- 3 files changed, 6 insertions(+), 3 deletions(-) diff --git a/paimon-python/pypaimon/filesystem/jindo_file_system_handler.py b/paimon-python/pypaimon/filesystem/jindo_file_system_handler.py index 500c5b0a919d..73306c8ffc2f 100644 --- a/paimon-python/pypaimon/filesystem/jindo_file_system_handler.py +++ b/paimon-python/pypaimon/filesystem/jindo_file_system_handler.py @@ -115,7 +115,8 @@ def create_jindo_oss_filesystem(root_uri: str, catalog_options: Options): class JindoInputFile: - supports_concurrent_pread = True + # Jindo stream handles are not documented as thread-safe. + supports_concurrent_pread = False def __init__(self, stream_factory): self._stream_factory = stream_factory diff --git a/paimon-python/pypaimon/tests/blob_test.py b/paimon-python/pypaimon/tests/blob_test.py index 9439ca2b50f5..7350f7285862 100644 --- a/paimon-python/pypaimon/tests/blob_test.py +++ b/paimon-python/pypaimon/tests/blob_test.py @@ -3878,7 +3878,7 @@ def new_input_stream(_): self.assertGreater(len(streams), 1) self.assertLessEqual(len(streams), 8) - def test_unmarked_read_at_uses_bounded_stream_pool(self): + def test_non_concurrent_read_at_uses_bounded_stream_pool(self): from pypaimon.common.file_io import FileIO data = bytes(range(256)) * 64 @@ -3886,6 +3886,8 @@ def test_unmarked_read_at_uses_bounded_stream_pool(self): streams = [] class PositionalStream: + supports_concurrent_pread = False + def __init__(self): self.reading = False self.reads = 0 diff --git a/paimon-python/pypaimon/tests/jindo_file_system_test.py b/paimon-python/pypaimon/tests/jindo_file_system_test.py index f15c61c5ce64..12eed05d4db8 100644 --- a/paimon-python/pypaimon/tests/jindo_file_system_test.py +++ b/paimon-python/pypaimon/tests/jindo_file_system_test.py @@ -43,7 +43,7 @@ def test_native_stream_is_opened_lazily_for_pread(self): handler, "blob") self.assertIsInstance(result, JindoInputFile) - self.assertTrue(result.supports_concurrent_pread) + self.assertFalse(result.supports_concurrent_pread) handler._jindo_fs.open.assert_not_called() self.assertEqual(b"data", result.read_at(4, 7)) native_stream.pread.assert_called_once_with(4, 7) From 901dbee6600e4d7585ca9af18e8c37ce138f2a9c Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Sun, 9 Aug 2026 06:06:55 -0700 Subject: [PATCH 07/16] [python] Bound pooled range streams globally --- paimon-python/pypaimon/common/file_io.py | 51 +++++++++++------ paimon-python/pypaimon/tests/blob_test.py | 70 +++++++++++++++++++++++ 2 files changed, 102 insertions(+), 19 deletions(-) diff --git a/paimon-python/pypaimon/common/file_io.py b/paimon-python/pypaimon/common/file_io.py index b8f1265c914f..b6b062e7cb49 100644 --- a/paimon-python/pypaimon/common/file_io.py +++ b/paimon-python/pypaimon/common/file_io.py @@ -267,11 +267,14 @@ def _read_ranges_coalesced(self, ranges, parallelism, max_gap, max_span, return results workers = max(1, min(parallelism, len(tasks))) + stream_budget = threading.BoundedSemaphore(workers) class _RangeStreamPool: - def __init__(self, file_io, path, task_count, max_streams): + def __init__(self, file_io, path, task_count, max_streams, + budget): self._file_io = file_io self._path = path + self._budget = budget self._condition = threading.Condition() self._streams = [] self._available = [] @@ -285,7 +288,25 @@ def __init__(self, file_io, path, task_count, max_streams): self._close_error = None def _open(self): - return self._file_io.new_input_stream(self._path) + self._budget.acquire() + try: + return self._file_io.new_input_stream(self._path) + except BaseException: + self._budget.release() + raise + + def _record_close_error(self, error): + with self._condition: + if self._close_error is None: + self._close_error = error + + def _close_stream(self, stream): + try: + stream.close() + except BaseException as error: + self._record_close_error(error) + finally: + self._budget.release() def _detect(self): with self._condition: @@ -300,10 +321,7 @@ def _detect(self): concurrent = supports_concurrent_pread(stream) except BaseException: if stream is not None: - try: - stream.close() - except Exception: - pass + self._close_stream(stream) with self._condition: self._detecting = False self._condition.notify_all() @@ -331,7 +349,7 @@ def _acquire(self): self._condition.wait() try: stream = self._open() - except Exception: + except BaseException: with self._condition: self._opening -= 1 self._condition.notify_all() @@ -348,13 +366,10 @@ def _return(self, stream): self._condition.notify() def _discard(self, stream): - try: - stream.close() - except Exception: - pass with self._condition: self._streams.remove(stream) self._condition.notify_all() + self._close_stream(stream) def read(self, offset, length): stream, exclusive = self._acquire() @@ -392,20 +407,18 @@ def _close_all(self): self._streams = [] self._available = [] for stream in streams: - try: - stream.close() - except BaseException as error: - if self._close_error is None: - self._close_error = error + self._close_stream(stream) def close(self): self._close_all() - if self._close_error is not None: - raise self._close_error + with self._condition: + close_error = self._close_error + if close_error is not None: + raise close_error streams = { path: _RangeStreamPool( - self, path, len(path_tasks), workers) + self, path, len(path_tasks), workers, stream_budget) for path, path_tasks in tasks_by_path.items() } diff --git a/paimon-python/pypaimon/tests/blob_test.py b/paimon-python/pypaimon/tests/blob_test.py index 7350f7285862..384809658513 100644 --- a/paimon-python/pypaimon/tests/blob_test.py +++ b/paimon-python/pypaimon/tests/blob_test.py @@ -22,6 +22,7 @@ import shutil import struct import tempfile +import threading import time import unittest import zlib @@ -3926,6 +3927,48 @@ def new_input_stream(_): self.assertEqual(64, sum(stream.reads for stream in streams)) self.assertTrue(all(stream.closed for stream in streams)) + def test_stream_budget_is_shared_across_paths(self): + from pypaimon.common.file_io import FileIO + + file_io = FileIO.get("file:///tmp", {}) + lock = threading.Lock() + open_streams = 0 + max_open_streams = 0 + + class PositionalStream: + supports_concurrent_pread = False + + def read_at(self, length, offset): + time.sleep(0.03 if offset == 0 else 0.001) + return bytes([offset]) * length + + def close(self): + nonlocal open_streams + with lock: + open_streams -= 1 + + def new_input_stream(_): + nonlocal open_streams, max_open_streams + with lock: + open_streams += 1 + max_open_streams = max(max_open_streams, open_streams) + return PositionalStream() + + file_io.new_input_stream = new_input_stream + ranges = [ + ("blob-%d" % path, offset, 4) + for path in range(4) + for offset in range(0, 32, 8) + ] + + self.assertEqual( + [bytes([offset]) * length for _, offset, length in ranges], + file_io.read_ranges_coalesced( + ranges, parallelism=4, max_gap=0), + ) + self.assertLessEqual(max_open_streams, 4) + self.assertEqual(0, open_streams) + def test_closes_all_streams_before_raising_close_error(self): from pypaimon.common.file_io import FileIO @@ -3986,6 +4029,33 @@ def fail_fallback(path, offset, length): file_io.read_ranges_coalesced( [("blob", 0, 4)], parallelism=1, max_gap=0) + def test_discard_close_error_is_propagated_after_fallback(self): + from pypaimon.common.file_io import FileIO + + file_io = FileIO.get("file:///tmp", {}) + fallbacks = [] + + class FailingStream: + supports_concurrent_pread = False + + def read_at(self, length, offset): + raise IOError("pooled read failed") + + def close(self): + raise IOError("discard close failed") + + def read_file_range(path, offset, length): + fallbacks.append((path, offset, length)) + return b"ok" + + file_io.new_input_stream = lambda _: FailingStream() + file_io.read_file_range = read_file_range + + with self.assertRaisesRegex(IOError, "discard close failed"): + file_io.read_ranges_coalesced( + [("blob", 0, 2)], parallelism=1, max_gap=0) + self.assertEqual([("blob", 0, 2)], fallbacks) + class ReadFileRangeTest(unittest.TestCase): """read_file_range must accept length == -1 (read to EOF) -- the valid From 74dd5e9252db22242dd926c4374f4c18a0c8de37 Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Sun, 9 Aug 2026 06:15:04 -0700 Subject: [PATCH 08/16] [python] Reclaim idle range streams across paths --- paimon-python/pypaimon/common/file_io.py | 59 +++++++++++++++++++++++- 1 file changed, 58 insertions(+), 1 deletion(-) diff --git a/paimon-python/pypaimon/common/file_io.py b/paimon-python/pypaimon/common/file_io.py index b6b062e7cb49..b0721eb7eff5 100644 --- a/paimon-python/pypaimon/common/file_io.py +++ b/paimon-python/pypaimon/common/file_io.py @@ -267,7 +267,52 @@ def _read_ranges_coalesced(self, ranges, parallelism, max_gap, max_span, return results workers = max(1, min(parallelism, len(tasks))) - stream_budget = threading.BoundedSemaphore(workers) + + class _RangeStreamBudget: + def __init__(self, limit): + self._limit = limit + self._condition = threading.Condition() + self._open_streams = 0 + self._idle_generation = 0 + self._pools = [] + + def register(self, pools): + self._pools = list(pools) + + def acquire(self): + while True: + with self._condition: + if self._open_streams < self._limit: + self._open_streams += 1 + return + generation = self._idle_generation + + reclaimed = any( + pool._evict_available_stream() + for pool in self._pools + ) + if reclaimed: + continue + + with self._condition: + if (self._open_streams >= self._limit + and generation == self._idle_generation): + self._condition.wait() + + def release(self): + with self._condition: + if self._open_streams <= 0: + raise RuntimeError( + "range stream budget released too many times") + self._open_streams -= 1 + self._condition.notify() + + def notify_idle(self): + with self._condition: + self._idle_generation += 1 + self._condition.notify() + + stream_budget = _RangeStreamBudget(workers) class _RangeStreamPool: def __init__(self, file_io, path, task_count, max_streams, @@ -364,6 +409,17 @@ def _return(self, stream): with self._condition: self._available.append(stream) self._condition.notify() + self._budget.notify_idle() + + def _evict_available_stream(self): + with self._condition: + if not self._available: + return False + stream = self._available.pop() + self._streams.remove(stream) + self._condition.notify_all() + self._close_stream(stream) + return True def _discard(self, stream): with self._condition: @@ -421,6 +477,7 @@ def close(self): self, path, len(path_tasks), workers, stream_budget) for path, path_tasks in tasks_by_path.items() } + stream_budget.register(streams.values()) def _read(path, offset, length): try: From 78d582190ca78eaef27d04d138c75ebe2d28787c Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Sun, 9 Aug 2026 06:18:51 -0700 Subject: [PATCH 09/16] [python] Isolate unknown-length range reads --- paimon-python/pypaimon/common/file_io.py | 43 +++++++++++++++++++---- paimon-python/pypaimon/tests/blob_test.py | 36 +++++++++++++++++++ 2 files changed, 72 insertions(+), 7 deletions(-) diff --git a/paimon-python/pypaimon/common/file_io.py b/paimon-python/pypaimon/common/file_io.py index b0721eb7eff5..457b860a1891 100644 --- a/paimon-python/pypaimon/common/file_io.py +++ b/paimon-python/pypaimon/common/file_io.py @@ -326,6 +326,7 @@ def __init__(self, file_io, path, task_count, max_streams, self._opening = 0 self._detecting = False self._concurrent = None + self._shared_users = 0 self._seek_lock = threading.Lock() self._remaining = task_count self._max_streams = min( @@ -380,11 +381,14 @@ def _detect(self): self._condition.notify_all() def _acquire(self): - self._detect() - with self._condition: - if self._concurrent: - return self._streams[0], False - while True: + while True: + self._detect() + with self._condition: + if self._concurrent is None: + continue + if self._concurrent: + self._shared_users += 1 + return self._streams[0], False if self._available: return self._available.pop(), True if (len(self._streams) + self._opening @@ -411,11 +415,25 @@ def _return(self, stream): self._condition.notify() self._budget.notify_idle() + def _release_shared(self): + with self._condition: + self._shared_users -= 1 + idle = self._shared_users == 0 + if idle: + self._condition.notify_all() + if idle: + self._budget.notify_idle() + def _evict_available_stream(self): with self._condition: - if not self._available: + if self._available: + stream = self._available.pop() + elif (self._concurrent and self._shared_users == 0 + and self._streams): + stream = self._streams[0] + self._concurrent = None + else: return False - stream = self._available.pop() self._streams.remove(stream) self._condition.notify_all() self._close_stream(stream) @@ -449,6 +467,8 @@ def read(self, offset, length): self._discard(stream) else: self._return(stream) + else: + self._release_shared() def task_done(self): with self._condition: @@ -479,7 +499,16 @@ def close(self): } stream_budget.register(streams.values()) + def _read_independent(path, offset, length): + stream_budget.acquire() + try: + return self.read_file_range(path, offset, length) + finally: + stream_budget.release() + def _read(path, offset, length): + if length < 0: + return _read_independent(path, offset, length) try: return streams[path].read(offset, length) except Exception: diff --git a/paimon-python/pypaimon/tests/blob_test.py b/paimon-python/pypaimon/tests/blob_test.py index 384809658513..ae2b94a7192c 100644 --- a/paimon-python/pypaimon/tests/blob_test.py +++ b/paimon-python/pypaimon/tests/blob_test.py @@ -3832,6 +3832,42 @@ def read_file_range(file_path, offset, length): ) self.assertEqual(2, len(fallbacks)) + def test_unknown_length_does_not_use_shared_stream(self): + from pypaimon.common.file_io import FileIO + + file_io = FileIO.get("file:///tmp", {}) + independent_reads = [] + + class SharedStream: + supports_concurrent_pread = True + + def read_at(self, length, offset): + return b"known" + + def seek(self, offset): + raise AssertionError( + "unknown-length read used the shared stream") + + def close(self): + pass + + def read_file_range(path, offset, length): + independent_reads.append((path, offset, length)) + return b"tail" + + file_io.new_input_stream = lambda _: SharedStream() + file_io.read_file_range = read_file_range + + self.assertEqual( + [b"known", b"tail"], + file_io.read_ranges_coalesced( + [("blob", 0, 5), ("blob", 5, -1)], + parallelism=1, + max_gap=0, + ), + ) + self.assertEqual([("blob", 5, -1)], independent_reads) + def test_non_positional_streams_are_exclusive(self): from pypaimon.common.file_io import FileIO From a843fd6ec7c3b57e4a3813192f55e08f3f9d4052 Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Sun, 9 Aug 2026 06:24:23 -0700 Subject: [PATCH 10/16] [python] Bound fallback range streams --- paimon-python/pypaimon/common/file_io.py | 2 +- paimon-python/pypaimon/tests/blob_test.py | 60 +++++++++++++++++++++++ 2 files changed, 61 insertions(+), 1 deletion(-) diff --git a/paimon-python/pypaimon/common/file_io.py b/paimon-python/pypaimon/common/file_io.py index 457b860a1891..dd6b9981c535 100644 --- a/paimon-python/pypaimon/common/file_io.py +++ b/paimon-python/pypaimon/common/file_io.py @@ -514,7 +514,7 @@ def _read(path, offset, length): except Exception: # Preserve the previous independent-open behavior as a retry # when a shared stream becomes unusable. - return self.read_file_range(path, offset, length) + return _read_independent(path, offset, length) def _run(task): kind, payload = task diff --git a/paimon-python/pypaimon/tests/blob_test.py b/paimon-python/pypaimon/tests/blob_test.py index ae2b94a7192c..d7f0d3a9d6dc 100644 --- a/paimon-python/pypaimon/tests/blob_test.py +++ b/paimon-python/pypaimon/tests/blob_test.py @@ -3832,6 +3832,66 @@ def read_file_range(file_path, offset, length): ) self.assertEqual(2, len(fallbacks)) + def test_fallback_reads_share_global_stream_budget(self): + from pypaimon.common.file_io import FileIO + + parallelism = 8 + file_io = FileIO.get("file:///tmp", {}) + barrier = threading.Barrier(parallelism) + lock = threading.Lock() + open_streams = 0 + max_open_streams = 0 + + def opened(): + nonlocal open_streams, max_open_streams + with lock: + open_streams += 1 + max_open_streams = max(max_open_streams, open_streams) + + def closed(): + nonlocal open_streams + with lock: + open_streams -= 1 + + class FailingStream: + supports_concurrent_pread = True + + def __init__(self): + self.closed = False + opened() + + def read_at(self, length, offset): + barrier.wait() + raise IOError("pooled read failed") + + def close(self): + if not self.closed: + self.closed = True + closed() + + def read_file_range(path, offset, length): + opened() + try: + time.sleep(0.01) + return b"ok" + finally: + closed() + + file_io.new_input_stream = lambda _: FailingStream() + file_io.read_file_range = read_file_range + ranges = [ + ("blob-%d" % index, 0, 2) + for index in range(parallelism) + ] + + self.assertEqual( + [b"ok"] * parallelism, + file_io.read_ranges_coalesced( + ranges, parallelism=parallelism, max_gap=0), + ) + self.assertLessEqual(max_open_streams, parallelism) + self.assertEqual(0, open_streams) + def test_unknown_length_does_not_use_shared_stream(self): from pypaimon.common.file_io import FileIO From 730db9a8bb8602481497a22a2d71d96e6a8f50ee Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Sun, 9 Aug 2026 06:41:01 -0700 Subject: [PATCH 11/16] [python] Simplify range stream reuse with lanes --- paimon-python/pypaimon/common/file_io.py | 410 +++++------------- paimon-python/pypaimon/tests/blob_test.py | 77 ++-- .../pypaimon/tests/hdfs_native_test.py | 30 +- 3 files changed, 177 insertions(+), 340 deletions(-) diff --git a/paimon-python/pypaimon/common/file_io.py b/paimon-python/pypaimon/common/file_io.py index dd6b9981c535..9536cca18337 100644 --- a/paimon-python/pypaimon/common/file_io.py +++ b/paimon-python/pypaimon/common/file_io.py @@ -17,7 +17,6 @@ import logging import os -import threading import uuid from abc import ABC, abstractmethod from pathlib import Path @@ -58,21 +57,12 @@ def pread(stream, length: int, offset: int) -> bytes: raise AttributeError("stream does not support positional reads") -def supports_concurrent_pread(stream) -> bool: - if not supports_pread(stream): - return False - concurrent = getattr(stream, 'supports_concurrent_pread', None) - if concurrent is not None: - return bool(concurrent) - return _fileno(stream) is not None - - # Coalescing bounds: merge same-file ranges whose gap is within GAP, capping a # merged read at SPAN so threads stay busy and memory stays bounded. _COALESCE_GAP = 1 << 20 _COALESCE_SPAN = 8 << 20 _COALESCE_VIEW_MAX_RETAINED_AMPLIFICATION = 2.0 -_MAX_EXCLUSIVE_RANGE_STREAMS = 16 +_MAX_RANGE_LANES_PER_PATH = 16 def create_temp_path(path: str) -> str: @@ -206,9 +196,9 @@ def read_ranges_coalesced(self, ranges, parallelism, max_gap=_COALESCE_GAP, max_span=_COALESCE_SPAN): """Read ``ranges`` (each ``None`` or ``(path, offset, length)``), returning bytes in the same order. Same-file nearby ranges are merged into one read - to cut round trips, then sliced. All spans for one path share an input - stream; reads run on a thread pool. Negative length (read to EOF) is read - on its own, never merged. + to cut round trips, then sliced. Each worker lane reuses one exclusive + stream for consecutive spans of the same path. Negative length (read to + EOF) is read on its own, never merged. A failed read propagates and aborts the whole batch (unlike a per-row ``file.open()`` loop that fails one row at a time). @@ -259,315 +249,131 @@ def _read_ranges_coalesced(self, ranges, parallelism, max_gap, max_span, for singleton in singletons: tasks_by_path.setdefault(singleton[1], []).append( ("one", singleton)) - tasks = [ - task for path_tasks in tasks_by_path.values() - for task in path_tasks - ] - if not tasks: + task_count = sum(len(path_tasks) + for path_tasks in tasks_by_path.values()) + if task_count == 0: return results - workers = max(1, min(parallelism, len(tasks))) - - class _RangeStreamBudget: - def __init__(self, limit): - self._limit = limit - self._condition = threading.Condition() - self._open_streams = 0 - self._idle_generation = 0 - self._pools = [] - - def register(self, pools): - self._pools = list(pools) - - def acquire(self): - while True: - with self._condition: - if self._open_streams < self._limit: - self._open_streams += 1 - return - generation = self._idle_generation - - reclaimed = any( - pool._evict_available_stream() - for pool in self._pools - ) - if reclaimed: - continue - - with self._condition: - if (self._open_streams >= self._limit - and generation == self._idle_generation): - self._condition.wait() - - def release(self): - with self._condition: - if self._open_streams <= 0: - raise RuntimeError( - "range stream budget released too many times") - self._open_streams -= 1 - self._condition.notify() - - def notify_idle(self): - with self._condition: - self._idle_generation += 1 - self._condition.notify() - - stream_budget = _RangeStreamBudget(workers) - - class _RangeStreamPool: - def __init__(self, file_io, path, task_count, max_streams, - budget): + workers = max(1, min(parallelism, task_count)) + + lanes = [[] for _ in range(workers)] + lane_loads = [0] * workers + for path_tasks in tasks_by_path.values(): + proportional_lanes = max( + 1, + (workers * len(path_tasks) + task_count - 1) // task_count, + ) + path_lanes = min( + len(path_tasks), proportional_lanes, + _MAX_RANGE_LANES_PER_PATH) + selected = sorted( + range(workers), key=lane_loads.__getitem__)[:path_lanes] + for index, task in enumerate(path_tasks): + lane = selected[index % path_lanes] + lanes[lane].append(task) + lane_loads[lane] += 1 + lanes = [lane for lane in lanes if lane] + + class _RangeLane: + def __init__(self, file_io): self._file_io = file_io - self._path = path - self._budget = budget - self._condition = threading.Condition() - self._streams = [] - self._available = [] - self._opening = 0 - self._detecting = False - self._concurrent = None - self._shared_users = 0 - self._seek_lock = threading.Lock() - self._remaining = task_count - self._max_streams = min( - max_streams, task_count, _MAX_EXCLUSIVE_RANGE_STREAMS) + self._path = None + self._stream = None self._close_error = None - def _open(self): - self._budget.acquire() + def _close_current(self): + stream = self._stream + self._stream = None + self._path = None + if stream is None: + return try: - return self._file_io.new_input_stream(self._path) - except BaseException: - self._budget.release() - raise - - def _record_close_error(self, error): - with self._condition: + stream.close() + except BaseException as error: if self._close_error is None: self._close_error = error - def _close_stream(self, stream): - try: - stream.close() - except BaseException as error: - self._record_close_error(error) - finally: - self._budget.release() - - def _detect(self): - with self._condition: - while self._concurrent is None and self._detecting: - self._condition.wait() - if self._concurrent is not None: - return - self._detecting = True - stream = None - try: - stream = self._open() - concurrent = supports_concurrent_pread(stream) - except BaseException: - if stream is not None: - self._close_stream(stream) - with self._condition: - self._detecting = False - self._condition.notify_all() - raise - with self._condition: - self._streams.append(stream) - self._concurrent = concurrent - if not self._concurrent: - self._available.append(stream) - self._detecting = False - self._condition.notify_all() - - def _acquire(self): - while True: - self._detect() - with self._condition: - if self._concurrent is None: - continue - if self._concurrent: - self._shared_users += 1 - return self._streams[0], False - if self._available: - return self._available.pop(), True - if (len(self._streams) + self._opening - < self._max_streams): - self._opening += 1 - break - self._condition.wait() - try: - stream = self._open() - except BaseException: - with self._condition: - self._opening -= 1 - self._condition.notify_all() - raise - with self._condition: - self._streams.append(stream) - self._opening -= 1 - self._condition.notify_all() - return stream, True - - def _return(self, stream): - with self._condition: - self._available.append(stream) - self._condition.notify() - self._budget.notify_idle() - - def _release_shared(self): - with self._condition: - self._shared_users -= 1 - idle = self._shared_users == 0 - if idle: - self._condition.notify_all() - if idle: - self._budget.notify_idle() - - def _evict_available_stream(self): - with self._condition: - if self._available: - stream = self._available.pop() - elif (self._concurrent and self._shared_users == 0 - and self._streams): - stream = self._streams[0] - self._concurrent = None - else: - return False - self._streams.remove(stream) - self._condition.notify_all() - self._close_stream(stream) - return True - - def _discard(self, stream): - with self._condition: - self._streams.remove(stream) - self._condition.notify_all() - self._close_stream(stream) - - def read(self, offset, length): - stream, exclusive = self._acquire() - failed = False + def _stream_for(self, path): + if self._stream is not None and self._path == path: + return self._stream + self._close_current() + self._stream = self._file_io.new_input_stream(path) + self._path = path + return self._stream + + def read(self, path, offset, length): try: + stream = self._stream_for(path) if length >= 0 and supports_pread(stream): return pread(stream, length, offset) - if not exclusive: - with self._seek_lock: - stream.seek(offset) - return (stream.read() if length < 0 - else stream.read(length)) stream.seek(offset) - return stream.read() if length < 0 else stream.read(length) + return (stream.read() if length < 0 + else stream.read(length)) except Exception: - failed = True - raise - finally: - if exclusive: - if failed: - self._discard(stream) - else: - self._return(stream) - else: - self._release_shared() - - def task_done(self): - with self._condition: - self._remaining -= 1 - close = self._remaining == 0 - if close: - self._close_all() - - def _close_all(self): - with self._condition: - streams = self._streams - self._streams = [] - self._available = [] - for stream in streams: - self._close_stream(stream) + self._close_current() + return self._file_io.read_file_range( + path, offset, length) def close(self): - self._close_all() - with self._condition: - close_error = self._close_error - if close_error is not None: - raise close_error - - streams = { - path: _RangeStreamPool( - self, path, len(path_tasks), workers, stream_budget) - for path, path_tasks in tasks_by_path.items() - } - stream_budget.register(streams.values()) - - def _read_independent(path, offset, length): - stream_budget.acquire() - try: - return self.read_file_range(path, offset, length) - finally: - stream_budget.release() - - def _read(path, offset, length): - if length < 0: - return _read_independent(path, offset, length) - try: - return streams[path].read(offset, length) - except Exception: - # Preserve the previous independent-open behavior as a retry - # when a shared stream becomes unusable. - return _read_independent(path, offset, length) + self._close_current() + if self._close_error is not None: + raise self._close_error - def _run(task): + def _run_task(reader, task): kind, payload = task - path = payload[0] if kind == "span" else payload[1] + if kind == "span": + path, span_off, span_len, members = payload + buf = reader.read(path, span_off, span_len) + if return_views: + buf = memoryview(buf) + useful = sum(length for _, _, length in members) + share_buffer = ( + max_retained_amplification <= 0 + or span_len <= useful * max_retained_amplification + ) + for idx, off, length in members: + start = off - span_off + value = buf[start:start + length] + if return_views and not share_buffer: + value = memoryview(bytes(value)) + results[idx] = value + else: + idx, path, offset, length = payload + result = reader.read(path, offset, length) + results[idx] = ( + memoryview(result) if return_views else result) + + def _run_lane(lane): + reader = _RangeLane(self) + read_error = None try: - if kind == "span": - path, span_off, span_len, members = payload - buf = _read(path, span_off, span_len) - if return_views: - buf = memoryview(buf) - useful = sum(length for _, _, length in members) - share_buffer = ( - max_retained_amplification <= 0 - or span_len <= useful * max_retained_amplification - ) - for idx, off, length in members: - s = off - span_off - value = buf[s:s + length] - if return_views and not share_buffer: - value = memoryview(bytes(value)) - results[idx] = value - else: - idx, path, off, length = payload - result = _read(path, off, length) - results[idx] = ( - memoryview(result) if return_views else result) - finally: - streams[path].task_done() - - failed = False - try: - with ThreadPoolExecutor(workers) as pool: - list(pool.map(_run, tasks)) - except BaseException: - failed = True - raise - finally: + for task in lane: + _run_task(reader, task) + except BaseException as error: + read_error = error close_error = None - for stream in streams.values(): - try: - stream.close() - except BaseException as error: - if close_error is None: - close_error = error + try: + reader.close() + except BaseException as error: + close_error = error + return read_error, close_error + + with ThreadPoolExecutor(len(lanes)) as pool: + outcomes = list(pool.map(_run_lane, lanes)) + read_error = next( + (error for error, _ in outcomes if error is not None), None) + close_error = next( + (error for _, error in outcomes if error is not None), None) + if read_error is not None: if close_error is not None: - if failed: - _LOG.warning( - "Failed to close a range input stream", - exc_info=(type(close_error), close_error, - close_error.__traceback__), - ) - else: - raise close_error + _LOG.warning( + "Failed to close a range input stream", + exc_info=(type(close_error), close_error, + close_error.__traceback__), + ) + raise read_error + if close_error is not None: + raise close_error return results def read_blobs_concurrent(self, blobs, parallelism): diff --git a/paimon-python/pypaimon/tests/blob_test.py b/paimon-python/pypaimon/tests/blob_test.py index d7f0d3a9d6dc..092835b499cc 100644 --- a/paimon-python/pypaimon/tests/blob_test.py +++ b/paimon-python/pypaimon/tests/blob_test.py @@ -3744,7 +3744,7 @@ def close(self): self.assertEqual(reads, [(path, 0, 1010)]) self.assertIs(shared[0].obj, shared[1].obj) - def test_fetch_bodies_reuses_one_stream_for_same_uri(self): + def test_fetch_bodies_reuses_bounded_streams_for_same_uri(self): from pypaimon.common.file_io import FileIO from pypaimon.multimodal.query import ScanQuery from pypaimon.table.row.blob import BlobDescriptor @@ -3793,11 +3793,11 @@ def close(self): file_io, {"image": cells}, ["image"], parallelism=16) self.assertEqual(payloads, bodies["image"]) - self.assertEqual([path], opens) + self.assertEqual([path] * 16, opens) self.assertEqual(span_count, len(reads)) - self.assertEqual([path], closes) + self.assertEqual([path] * 16, closes) - def test_shared_stream_failure_reopens_range(self): + def test_lane_stream_failure_reopens_range(self): from pypaimon.common.file_io import FileIO data = bytes(range(64)) @@ -3832,7 +3832,7 @@ def read_file_range(file_path, offset, length): ) self.assertEqual(2, len(fallbacks)) - def test_fallback_reads_share_global_stream_budget(self): + def test_fallback_reads_do_not_exceed_parallelism(self): from pypaimon.common.file_io import FileIO parallelism = 8 @@ -3892,31 +3892,41 @@ def read_file_range(path, offset, length): self.assertLessEqual(max_open_streams, parallelism) self.assertEqual(0, open_streams) - def test_unknown_length_does_not_use_shared_stream(self): + def test_known_and_unknown_lengths_share_exclusive_lane(self): from pypaimon.common.file_io import FileIO file_io = FileIO.get("file:///tmp", {}) - independent_reads = [] + operations = [] + streams = [] - class SharedStream: - supports_concurrent_pread = True + class LaneStream: + def __init__(self): + self.position = 0 + self.closed = False def read_at(self, length, offset): + operations.append(("read_at", offset, length)) return b"known" def seek(self, offset): - raise AssertionError( - "unknown-length read used the shared stream") + operations.append(("seek", offset)) + self.position = offset + + def read(self): + operations.append(("read", self.position)) + return b"tail" def close(self): - pass + self.closed = True - def read_file_range(path, offset, length): - independent_reads.append((path, offset, length)) - return b"tail" + def new_input_stream(_): + stream = LaneStream() + streams.append(stream) + return stream - file_io.new_input_stream = lambda _: SharedStream() - file_io.read_file_range = read_file_range + file_io.new_input_stream = new_input_stream + file_io.read_file_range = lambda *args: self.fail( + "exclusive lane unexpectedly used fallback") self.assertEqual( [b"known", b"tail"], @@ -3926,7 +3936,13 @@ def read_file_range(path, offset, length): max_gap=0, ), ) - self.assertEqual([("blob", 5, -1)], independent_reads) + self.assertEqual([ + ("read_at", 0, 5), + ("seek", 5), + ("read", 5), + ], operations) + self.assertEqual(1, len(streams)) + self.assertTrue(streams[0].closed) def test_non_positional_streams_are_exclusive(self): from pypaimon.common.file_io import FileIO @@ -3975,7 +3991,7 @@ def new_input_stream(_): self.assertGreater(len(streams), 1) self.assertLessEqual(len(streams), 8) - def test_non_concurrent_read_at_uses_bounded_stream_pool(self): + def test_same_path_uses_bounded_exclusive_lanes(self): from pypaimon.common.file_io import FileIO data = bytes(range(256)) * 64 @@ -4023,13 +4039,14 @@ def new_input_stream(_): self.assertEqual(64, sum(stream.reads for stream in streams)) self.assertTrue(all(stream.closed for stream in streams)) - def test_stream_budget_is_shared_across_paths(self): + def test_stream_count_is_bounded_across_paths(self): from pypaimon.common.file_io import FileIO file_io = FileIO.get("file:///tmp", {}) lock = threading.Lock() open_streams = 0 max_open_streams = 0 + total_streams = 0 class PositionalStream: supports_concurrent_pread = False @@ -4044,9 +4061,10 @@ def close(self): open_streams -= 1 def new_input_stream(_): - nonlocal open_streams, max_open_streams + nonlocal open_streams, max_open_streams, total_streams with lock: open_streams += 1 + total_streams += 1 max_open_streams = max(max_open_streams, open_streams) return PositionalStream() @@ -4063,6 +4081,7 @@ def new_input_stream(_): ranges, parallelism=4, max_gap=0), ) self.assertLessEqual(max_open_streams, 4) + self.assertEqual(4, total_streams) self.assertEqual(0, open_streams) def test_closes_all_streams_before_raising_close_error(self): @@ -4107,10 +4126,13 @@ def test_close_error_does_not_mask_read_error(self): file_io = FileIO.get("file:///tmp", {}) class FailingStream: - supports_concurrent_pread = True + def __init__(self, path): + self.path = path def read_at(self, length, offset): - raise IOError("shared read failed") + if self.path == "read-error": + raise IOError("shared read failed") + return b"ok" def close(self): raise IOError("close failed") @@ -4118,14 +4140,17 @@ def close(self): def fail_fallback(path, offset, length): raise IOError("fallback read failed") - file_io.new_input_stream = lambda _: FailingStream() + file_io.new_input_stream = FailingStream file_io.read_file_range = fail_fallback with self.assertRaisesRegex(IOError, "fallback read failed"): file_io.read_ranges_coalesced( - [("blob", 0, 4)], parallelism=1, max_gap=0) + [("close-error", 0, 2), ("read-error", 0, 2)], + parallelism=2, + max_gap=0, + ) - def test_discard_close_error_is_propagated_after_fallback(self): + def test_failed_stream_close_error_is_propagated_after_fallback(self): from pypaimon.common.file_io import FileIO file_io = FileIO.get("file:///tmp", {}) diff --git a/paimon-python/pypaimon/tests/hdfs_native_test.py b/paimon-python/pypaimon/tests/hdfs_native_test.py index 7e5b9e33d0d6..025ea87235b1 100644 --- a/paimon-python/pypaimon/tests/hdfs_native_test.py +++ b/paimon-python/pypaimon/tests/hdfs_native_test.py @@ -490,32 +490,38 @@ def test_reader_adapter_read_at_clamps_to_eof(self): def test_reader_adapter_read_at_is_concurrent(self): from pypaimon.filesystem.hdfs_native_file_io import _HdfsReaderAdapter - class RangeReader: + class Concurrency: def __init__(self): self.lock = threading.Lock() self.active = 0 self.max_active = 0 + concurrency = Concurrency() + + class RangeReader: def __len__(self): return 64 def read_range(self, offset, length): - with self.lock: - self.active += 1 - self.max_active = max(self.max_active, self.active) + with concurrency.lock: + concurrency.active += 1 + concurrency.max_active = max( + concurrency.max_active, concurrency.active) try: time.sleep(0.01) return bytes([offset]) * length finally: - with self.lock: - self.active -= 1 + with concurrency.lock: + concurrency.active -= 1 - reader = RangeReader() - adapter = _HdfsReaderAdapter(reader) file_io = FileIO.get("file:///tmp", {}) opens = [] - file_io.new_input_stream = lambda path: ( - opens.append(path) or adapter) + + def new_input_stream(path): + opens.append(path) + return _HdfsReaderAdapter(RangeReader()) + + file_io.new_input_stream = new_input_stream ranges = [("hdfs://ns/blob", offset, 4) for offset in range(0, 64, 8)] results = file_io.read_ranges_coalesced( @@ -523,8 +529,8 @@ def read_range(self, offset, length): self.assertEqual( [bytes([offset]) * 4 for _, offset, _ in ranges], results) - self.assertEqual(["hdfs://ns/blob"], opens) - self.assertEqual(8, reader.max_active) + self.assertEqual(["hdfs://ns/blob"] * 8, opens) + self.assertEqual(8, concurrency.max_active) def test_reader_adapter_close_releases_underlying(self): from pypaimon.filesystem.hdfs_native_file_io import _HdfsReaderAdapter From 4bc6f5b65dede3664cd3167b52097d08da06c57c Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Sun, 9 Aug 2026 19:50:53 -0700 Subject: [PATCH 12/16] [python] Narrow blob stream reuse scope --- paimon-python/pypaimon/common/file_io.py | 32 +++------ .../filesystem/hdfs_native_file_io.py | 10 --- .../filesystem/jindo_file_system_handler.py | 62 +++++++--------- .../pypaimon/filesystem/pyarrow_file_io.py | 2 - paimon-python/pypaimon/tests/blob_test.py | 71 +------------------ .../pypaimon/tests/hdfs_native_test.py | 70 ------------------ .../pypaimon/tests/jindo_file_system_test.py | 66 +---------------- 7 files changed, 38 insertions(+), 275 deletions(-) diff --git a/paimon-python/pypaimon/common/file_io.py b/paimon-python/pypaimon/common/file_io.py index 9536cca18337..5d2e66392d75 100644 --- a/paimon-python/pypaimon/common/file_io.py +++ b/paimon-python/pypaimon/common/file_io.py @@ -30,31 +30,24 @@ _LOG = logging.getLogger(__name__) -def _fileno(stream): - if not hasattr(stream, 'fileno'): - return None - try: - return stream.fileno() - except Exception: - return None - - def supports_pread(stream) -> bool: - """Check if the stream supports position-based reads.""" - # Unlike read_at, Python pread methods have no common argument order. + """Check if the stream supports position-based reads (thread-safe I/O).""" if hasattr(stream, 'read_at'): return True - return _fileno(stream) is not None + if hasattr(stream, 'fileno'): + try: + stream.fileno() + return True + except Exception: + pass + return False def pread(stream, length: int, offset: int) -> bytes: - """Position-based read without changing the stream cursor.""" - fd = _fileno(stream) - if fd is not None: - return os.pread(fd, length, offset) + """Position-based read without changing the stream cursor. Thread-safe.""" if hasattr(stream, 'read_at'): return stream.read_at(length, offset) - raise AttributeError("stream does not support positional reads") + return os.pread(stream.fileno(), length, offset) # Coalescing bounds: merge same-file ranges whose gap is within GAP, capping a @@ -62,7 +55,6 @@ def pread(stream, length: int, offset: int) -> bytes: _COALESCE_GAP = 1 << 20 _COALESCE_SPAN = 8 << 20 _COALESCE_VIEW_MAX_RETAINED_AMPLIFICATION = 2.0 -_MAX_RANGE_LANES_PER_PATH = 16 def create_temp_path(path: str) -> str: @@ -263,9 +255,7 @@ def _read_ranges_coalesced(self, ranges, parallelism, max_gap, max_span, 1, (workers * len(path_tasks) + task_count - 1) // task_count, ) - path_lanes = min( - len(path_tasks), proportional_lanes, - _MAX_RANGE_LANES_PER_PATH) + path_lanes = min(len(path_tasks), proportional_lanes) selected = sorted( range(workers), key=lane_loads.__getitem__)[:path_lanes] for index, task in enumerate(path_tasks): diff --git a/paimon-python/pypaimon/filesystem/hdfs_native_file_io.py b/paimon-python/pypaimon/filesystem/hdfs_native_file_io.py index d61157d49f95..78cad2140e28 100644 --- a/paimon-python/pypaimon/filesystem/hdfs_native_file_io.py +++ b/paimon-python/pypaimon/filesystem/hdfs_native_file_io.py @@ -107,8 +107,6 @@ class _HdfsReaderAdapter: is closed — hdfs-native's own FileReader.__exit__ is a no-op. """ - supports_concurrent_pread = True - def __init__(self, fr): self._fr = fr self._closed = False @@ -119,14 +117,6 @@ def read(self, size: int = -1) -> bytes: def read1(self, size: int = -1) -> bytes: return self.read(size) - def read_at(self, nbytes: int, offset: int) -> bytes: - if offset < 0: - raise ValueError("offset must be non-negative") - file_size = len(self._fr) - if nbytes <= 0 or offset >= file_size: - return b'' - return self._fr.read_range(offset, min(nbytes, file_size - offset)) - def seek(self, pos: int, whence: int = 0) -> int: self._fr.seek(pos, whence) return self._fr.tell() diff --git a/paimon-python/pypaimon/filesystem/jindo_file_system_handler.py b/paimon-python/pypaimon/filesystem/jindo_file_system_handler.py index 73306c8ffc2f..efbfe7f1899d 100644 --- a/paimon-python/pypaimon/filesystem/jindo_file_system_handler.py +++ b/paimon-python/pypaimon/filesystem/jindo_file_system_handler.py @@ -16,7 +16,6 @@ # under the License. import logging -import threading import pyarrow as pa from pyarrow import PythonFile @@ -115,54 +114,42 @@ def create_jindo_oss_filesystem(root_uri: str, catalog_options: Options): class JindoInputFile: - # Jindo stream handles are not documented as thread-safe. - supports_concurrent_pread = False - - def __init__(self, stream_factory): - self._stream_factory = stream_factory - self._stream = None - self._lock = threading.Lock() + def __init__(self, jindo_stream): + self._stream = jindo_stream self._closed = False @property def closed(self): + if hasattr(self._stream, 'closed'): + return self._stream.closed return self._closed - def _get_stream(self): - if self._closed: - raise ValueError("I/O operation on closed file") - if self._stream is None: - with self._lock: - if self._closed: - raise ValueError("I/O operation on closed file") - if self._stream is None: - self._stream = self._stream_factory() - return self._stream - def read(self, nbytes: int = -1): - stream = self._get_stream() + if self.closed: + raise ValueError("I/O operation on closed file") if nbytes is None or nbytes < 0: - return stream.read() - return stream.read(nbytes) + return self._stream.read() + return self._stream.read(nbytes) def seek(self, position: int, whence: int = 0): - return self._get_stream().seek(position, whence) + if self.closed: + raise ValueError("I/O operation on closed file") + self._stream.seek(position, whence) def tell(self) -> int: - return self._get_stream().tell() + if self.closed: + raise ValueError("I/O operation on closed file") + return self._stream.tell() def read_at(self, nbytes: int, offset: int): - return self._get_stream().pread(nbytes, offset) + if self.closed: + raise ValueError("I/O operation on closed file") + return self._stream.pread(nbytes, offset) def close(self): - with self._lock: - if self._closed: - return + if not self._closed: + self._stream.close() self._closed = True - stream = self._stream - self._stream = None - if stream is not None: - stream.close() def __enter__(self): return self @@ -332,15 +319,14 @@ def copy_file(self, src: str, dest: str): self._jindo_fs.copy_file(src_norm, dst_norm) def open_input_stream(self, path: str): - return PythonFile(self.new_input_stream(path), mode="r") + normalized = self._normalize_path(path) + jindo_stream = self._jindo_fs.open(normalized, "rb") + return PythonFile(JindoInputFile(jindo_stream), mode="r") def open_input_file(self, path: str): - return PythonFile(self.new_input_stream(path), mode="r") - - def new_input_stream(self, path: str): normalized = self._normalize_path(path) - return JindoInputFile( - lambda: self._jindo_fs.open(normalized, "rb")) + jindo_stream = self._jindo_fs.open(normalized, "rb") + return PythonFile(JindoInputFile(jindo_stream), mode="r") def open_output_stream(self, path: str, metadata): normalized = self._normalize_path(path) diff --git a/paimon-python/pypaimon/filesystem/pyarrow_file_io.py b/paimon-python/pypaimon/filesystem/pyarrow_file_io.py index 464c5c02c762..4423c9559e30 100644 --- a/paimon-python/pypaimon/filesystem/pyarrow_file_io.py +++ b/paimon-python/pypaimon/filesystem/pyarrow_file_io.py @@ -355,8 +355,6 @@ def _get_ticket_cache_path() -> Optional[str]: return _kerberos.get_ticket_cache_path() def new_input_stream(self, path: str): - if self._use_jindo: - return self.filesystem.handler.new_input_stream(path) path_str = self.to_filesystem_path(path) return self.filesystem.open_input_file(path_str) diff --git a/paimon-python/pypaimon/tests/blob_test.py b/paimon-python/pypaimon/tests/blob_test.py index 092835b499cc..c01438c61e0b 100644 --- a/paimon-python/pypaimon/tests/blob_test.py +++ b/paimon-python/pypaimon/tests/blob_test.py @@ -2424,8 +2424,6 @@ def new_input_stream(path): stream = original_open(path) class TrackingStream: - supports_concurrent_pread = True - def read(self, length=-1): return stream.read(length) @@ -3709,8 +3707,6 @@ def new_input_stream(file_path): stream = original_open(file_path) class TrackingStream: - supports_concurrent_pread = True - def read_at(self, length, offset): reads.append((file_path, offset, length)) return os.pread(stream.fileno(), length, offset) @@ -3744,59 +3740,6 @@ def close(self): self.assertEqual(reads, [(path, 0, 1010)]) self.assertIs(shared[0].obj, shared[1].obj) - def test_fetch_bodies_reuses_bounded_streams_for_same_uri(self): - from pypaimon.common.file_io import FileIO - from pypaimon.multimodal.query import ScanQuery - from pypaimon.table.row.blob import BlobDescriptor - - span_count = 64 - span_gap = 2 << 20 - payloads = [("blob-%02d" % i).encode() for i in range(span_count)] - with tempfile.TemporaryDirectory() as tmp_dir: - path = os.path.join(tmp_dir, "blobs.bin") - with open(path, "wb") as output: - for i, payload in enumerate(payloads): - output.seek(i * span_gap) - output.write(payload) - - file_io = FileIO.get(f"file://{tmp_dir}", {}) - original_open = file_io.new_input_stream - opens = [] - reads = [] - closes = [] - - def new_input_stream(file_path): - opens.append(file_path) - stream = original_open(file_path) - - class TrackingStream: - supports_concurrent_pread = True - - def read_at(self, length, offset): - reads.append((offset, length)) - return os.pread(stream.fileno(), length, offset) - - def close(self): - closes.append(file_path) - stream.close() - - return TrackingStream() - - file_io.new_input_stream = new_input_stream - cells = [ - BlobDescriptor( - path, i * span_gap, len(payload)).serialize() - for i, payload in enumerate(payloads) - ] - - bodies = ScanQuery._fetch_bodies( - file_io, {"image": cells}, ["image"], parallelism=16) - - self.assertEqual(payloads, bodies["image"]) - self.assertEqual([path] * 16, opens) - self.assertEqual(span_count, len(reads)) - self.assertEqual([path] * 16, closes) - def test_lane_stream_failure_reopens_range(self): from pypaimon.common.file_io import FileIO @@ -3809,8 +3752,6 @@ def test_lane_stream_failure_reopens_range(self): fallbacks = [] class FailingStream: - supports_concurrent_pread = True - def read_at(self, length, offset): raise IOError("shared stream failed") @@ -3854,8 +3795,6 @@ def closed(): open_streams -= 1 class FailingStream: - supports_concurrent_pread = True - def __init__(self): self.closed = False opened() @@ -3991,7 +3930,7 @@ def new_input_stream(_): self.assertGreater(len(streams), 1) self.assertLessEqual(len(streams), 8) - def test_same_path_uses_bounded_exclusive_lanes(self): + def test_same_path_uses_requested_exclusive_lanes(self): from pypaimon.common.file_io import FileIO data = bytes(range(256)) * 64 @@ -3999,8 +3938,6 @@ def test_same_path_uses_bounded_exclusive_lanes(self): streams = [] class PositionalStream: - supports_concurrent_pread = False - def __init__(self): self.reading = False self.reads = 0 @@ -4035,7 +3972,7 @@ def new_input_stream(_): ranges, parallelism=32, max_gap=0), ) self.assertGreater(len(streams), 1) - self.assertLessEqual(len(streams), 16) + self.assertLessEqual(len(streams), 32) self.assertEqual(64, sum(stream.reads for stream in streams)) self.assertTrue(all(stream.closed for stream in streams)) @@ -4049,8 +3986,6 @@ def test_stream_count_is_bounded_across_paths(self): total_streams = 0 class PositionalStream: - supports_concurrent_pread = False - def read_at(self, length, offset): time.sleep(0.03 if offset == 0 else 0.001) return bytes([offset]) * length @@ -4157,8 +4092,6 @@ def test_failed_stream_close_error_is_propagated_after_fallback(self): fallbacks = [] class FailingStream: - supports_concurrent_pread = False - def read_at(self, length, offset): raise IOError("pooled read failed") diff --git a/paimon-python/pypaimon/tests/hdfs_native_test.py b/paimon-python/pypaimon/tests/hdfs_native_test.py index 025ea87235b1..957ee8f123dd 100644 --- a/paimon-python/pypaimon/tests/hdfs_native_test.py +++ b/paimon-python/pypaimon/tests/hdfs_native_test.py @@ -18,8 +18,6 @@ import os import sys import tempfile -import threading -import time import types import unittest from unittest.mock import MagicMock, patch @@ -464,74 +462,6 @@ def test_reader_adapter_read_negative_reads_all(self): self.assertEqual(adapter.read(), b"all-content") fr.read.assert_called_once_with(-1) - def test_reader_adapter_read_at(self): - from pypaimon.filesystem.hdfs_native_file_io import _HdfsReaderAdapter - fr = MagicMock() - fr.__len__.return_value = 100 - fr.read_range.return_value = b"data" - adapter = _HdfsReaderAdapter(fr) - self.assertEqual(adapter.read_at(4, 7), b"data") - fr.read_range.assert_called_once_with(7, 4) - - def test_reader_adapter_read_at_clamps_to_eof(self): - from pypaimon.filesystem.hdfs_native_file_io import _HdfsReaderAdapter - fr = MagicMock() - fr.__len__.return_value = 10 - fr.read_range.return_value = b"89" - adapter = _HdfsReaderAdapter(fr) - - self.assertEqual(adapter.read_at(4, 8), b"89") - fr.read_range.assert_called_once_with(8, 2) - self.assertEqual(adapter.read_at(4, 10), b"") - self.assertEqual(1, fr.read_range.call_count) - with self.assertRaisesRegex(ValueError, "non-negative"): - adapter.read_at(1, -1) - - def test_reader_adapter_read_at_is_concurrent(self): - from pypaimon.filesystem.hdfs_native_file_io import _HdfsReaderAdapter - - class Concurrency: - def __init__(self): - self.lock = threading.Lock() - self.active = 0 - self.max_active = 0 - - concurrency = Concurrency() - - class RangeReader: - def __len__(self): - return 64 - - def read_range(self, offset, length): - with concurrency.lock: - concurrency.active += 1 - concurrency.max_active = max( - concurrency.max_active, concurrency.active) - try: - time.sleep(0.01) - return bytes([offset]) * length - finally: - with concurrency.lock: - concurrency.active -= 1 - - file_io = FileIO.get("file:///tmp", {}) - opens = [] - - def new_input_stream(path): - opens.append(path) - return _HdfsReaderAdapter(RangeReader()) - - file_io.new_input_stream = new_input_stream - ranges = [("hdfs://ns/blob", offset, 4) - for offset in range(0, 64, 8)] - results = file_io.read_ranges_coalesced( - ranges, parallelism=8, max_gap=0) - - self.assertEqual( - [bytes([offset]) * 4 for _, offset, _ in ranges], results) - self.assertEqual(["hdfs://ns/blob"] * 8, opens) - self.assertEqual(8, concurrency.max_active) - def test_reader_adapter_close_releases_underlying(self): from pypaimon.filesystem.hdfs_native_file_io import _HdfsReaderAdapter fr = MagicMock() diff --git a/paimon-python/pypaimon/tests/jindo_file_system_test.py b/paimon-python/pypaimon/tests/jindo_file_system_test.py index 12eed05d4db8..9bd45aeddaad 100644 --- a/paimon-python/pypaimon/tests/jindo_file_system_test.py +++ b/paimon-python/pypaimon/tests/jindo_file_system_test.py @@ -18,77 +18,13 @@ import os import unittest import uuid -from unittest.mock import MagicMock import pyarrow.fs as pafs from pyarrow.fs import PyFileSystem from pypaimon.common.options import Options from pypaimon.common.options.config import OssOptions -from pypaimon.filesystem.jindo_file_system_handler import ( - JindoFileSystemHandler, JindoInputFile, JINDO_AVAILABLE) -from pypaimon.filesystem.pyarrow_file_io import PyArrowFileIO - - -class JindoInputStreamTest(unittest.TestCase): - - def test_native_stream_is_opened_lazily_for_pread(self): - handler = MagicMock() - handler._normalize_path.return_value = "oss://bucket/blob" - native_stream = MagicMock() - native_stream.pread.return_value = b"data" - handler._jindo_fs.open.return_value = native_stream - - result = JindoFileSystemHandler.new_input_stream( - handler, "blob") - - self.assertIsInstance(result, JindoInputFile) - self.assertFalse(result.supports_concurrent_pread) - handler._jindo_fs.open.assert_not_called() - self.assertEqual(b"data", result.read_at(4, 7)) - native_stream.pread.assert_called_once_with(4, 7) - handler._jindo_fs.open.assert_called_once_with( - "oss://bucket/blob", "rb") - result.close() - native_stream.close.assert_called_once() - - def test_regular_stream_methods_use_one_lazy_native_stream(self): - native_stream = MagicMock() - native_stream.read.return_value = b"data" - native_stream.tell.return_value = 4 - stream = JindoInputFile(lambda: native_stream) - - self.assertEqual(b"data", stream.read(4)) - stream.seek(2) - self.assertEqual(4, stream.tell()) - - native_stream.read.assert_called_once_with(4) - native_stream.seek.assert_called_once_with(2, 0) - native_stream.tell.assert_called_once() - - def test_close_before_read_does_not_open_native_stream(self): - stream_factory = MagicMock() - stream = JindoInputFile(stream_factory) - - stream.close() - - stream_factory.assert_not_called() - with self.assertRaisesRegex(ValueError, "closed file"): - stream.read(1) - - def test_pyarrow_file_io_exposes_jindo_adapter_directly(self): - file_io = object.__new__(PyArrowFileIO) - file_io._use_jindo = True - file_io.filesystem = MagicMock() - stream = object() - file_io.filesystem.handler.new_input_stream.return_value = stream - - self.assertIs( - stream, - file_io.new_input_stream("oss://bucket/blob"), - ) - file_io.filesystem.handler.new_input_stream.assert_called_once_with( - "oss://bucket/blob") +from pypaimon.filesystem.jindo_file_system_handler import JindoFileSystemHandler, JINDO_AVAILABLE class JindoFileSystemTest(unittest.TestCase): From 33c3a3762add3e414f7b87174dd2dd79b9aa3263 Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Sun, 9 Aug 2026 20:58:26 -0700 Subject: [PATCH 13/16] python: bound range lanes per blob path --- paimon-python/pypaimon/common/file_io.py | 5 ++++- paimon-python/pypaimon/tests/blob_test.py | 6 +++--- 2 files changed, 7 insertions(+), 4 deletions(-) diff --git a/paimon-python/pypaimon/common/file_io.py b/paimon-python/pypaimon/common/file_io.py index 5d2e66392d75..1c2381912ce6 100644 --- a/paimon-python/pypaimon/common/file_io.py +++ b/paimon-python/pypaimon/common/file_io.py @@ -55,6 +55,7 @@ def pread(stream, length: int, offset: int) -> bytes: _COALESCE_GAP = 1 << 20 _COALESCE_SPAN = 8 << 20 _COALESCE_VIEW_MAX_RETAINED_AMPLIFICATION = 2.0 +_MAX_RANGE_LANES_PER_PATH = 16 def create_temp_path(path: str) -> str: @@ -255,7 +256,9 @@ def _read_ranges_coalesced(self, ranges, parallelism, max_gap, max_span, 1, (workers * len(path_tasks) + task_count - 1) // task_count, ) - path_lanes = min(len(path_tasks), proportional_lanes) + path_lanes = min( + len(path_tasks), proportional_lanes, + _MAX_RANGE_LANES_PER_PATH) selected = sorted( range(workers), key=lane_loads.__getitem__)[:path_lanes] for index, task in enumerate(path_tasks): diff --git a/paimon-python/pypaimon/tests/blob_test.py b/paimon-python/pypaimon/tests/blob_test.py index c01438c61e0b..322c7c95f871 100644 --- a/paimon-python/pypaimon/tests/blob_test.py +++ b/paimon-python/pypaimon/tests/blob_test.py @@ -3930,7 +3930,7 @@ def new_input_stream(_): self.assertGreater(len(streams), 1) self.assertLessEqual(len(streams), 8) - def test_same_path_uses_requested_exclusive_lanes(self): + def test_same_path_reuses_bounded_exclusive_lanes(self): from pypaimon.common.file_io import FileIO data = bytes(range(256)) * 64 @@ -3969,10 +3969,10 @@ def new_input_stream(_): [data[offset:offset + length] for _, offset, length in ranges], file_io.read_ranges_coalesced( - ranges, parallelism=32, max_gap=0), + ranges, parallelism=64, max_gap=0), ) self.assertGreater(len(streams), 1) - self.assertLessEqual(len(streams), 32) + self.assertLessEqual(len(streams), 16) self.assertEqual(64, sum(stream.reads for stream in streams)) self.assertTrue(all(stream.closed for stream in streams)) From 38a232dfdedc2dce9f865178de5fe80b5b738a97 Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Sun, 9 Aug 2026 22:23:59 -0700 Subject: [PATCH 14/16] python: rebalance capped blob range lanes --- paimon-python/pypaimon/common/file_io.py | 35 ++++++++++++++---- paimon-python/pypaimon/tests/blob_test.py | 43 +++++++++++++++++++++-- 2 files changed, 68 insertions(+), 10 deletions(-) diff --git a/paimon-python/pypaimon/common/file_io.py b/paimon-python/pypaimon/common/file_io.py index 1c2381912ce6..091d8820b719 100644 --- a/paimon-python/pypaimon/common/file_io.py +++ b/paimon-python/pypaimon/common/file_io.py @@ -55,6 +55,7 @@ def pread(stream, length: int, offset: int) -> bytes: _COALESCE_GAP = 1 << 20 _COALESCE_SPAN = 8 << 20 _COALESCE_VIEW_MAX_RETAINED_AMPLIFICATION = 2.0 +# Bound per-object opens; 16 cuts them by 75% for default 64-range batches. _MAX_RANGE_LANES_PER_PATH = 16 @@ -251,14 +252,34 @@ def _read_ranges_coalesced(self, ranges, parallelism, max_gap, max_span, lanes = [[] for _ in range(workers)] lane_loads = [0] * workers - for path_tasks in tasks_by_path.values(): - proportional_lanes = max( - 1, - (workers * len(path_tasks) + task_count - 1) // task_count, + path_task_groups = list(tasks_by_path.values()) + path_capacities = [ + min(len(path_tasks), _MAX_RANGE_LANES_PER_PATH) + for path_tasks in path_task_groups + ] + path_lane_counts = [1] * len(path_task_groups) + remaining_lanes = max( + 0, + min(workers, sum(path_capacities)) - len(path_task_groups), + ) + for _ in range(remaining_lanes): + candidates = [ + index for index in range(len(path_task_groups)) + if path_lane_counts[index] < path_capacities[index] + ] + if not candidates: + break + index = max( + candidates, + key=lambda value: ( + len(path_task_groups[value]) + / path_lane_counts[value] + ), ) - path_lanes = min( - len(path_tasks), proportional_lanes, - _MAX_RANGE_LANES_PER_PATH) + path_lane_counts[index] += 1 + + for path_tasks, path_lanes in zip( + path_task_groups, path_lane_counts): selected = sorted( range(workers), key=lane_loads.__getitem__)[:path_lanes] for index, task in enumerate(path_tasks): diff --git a/paimon-python/pypaimon/tests/blob_test.py b/paimon-python/pypaimon/tests/blob_test.py index 322c7c95f871..1dedbb8a2c04 100644 --- a/paimon-python/pypaimon/tests/blob_test.py +++ b/paimon-python/pypaimon/tests/blob_test.py @@ -3800,7 +3800,7 @@ def __init__(self): opened() def read_at(self, length, offset): - barrier.wait() + barrier.wait(timeout=5) raise IOError("pooled read failed") def close(self): @@ -3830,6 +3830,7 @@ def read_file_range(path, offset, length): ) self.assertLessEqual(max_open_streams, parallelism) self.assertEqual(0, open_streams) + self.assertFalse(barrier.broken) def test_known_and_unknown_lengths_share_exclusive_lane(self): from pypaimon.common.file_io import FileIO @@ -3971,11 +3972,47 @@ def new_input_stream(_): file_io.read_ranges_coalesced( ranges, parallelism=64, max_gap=0), ) - self.assertGreater(len(streams), 1) - self.assertLessEqual(len(streams), 16) + self.assertEqual(16, len(streams)) + self.assertTrue(all(stream.reads == 4 for stream in streams)) self.assertEqual(64, sum(stream.reads for stream in streams)) self.assertTrue(all(stream.closed for stream in streams)) + def test_skewed_paths_redistribute_capped_lanes(self): + from collections import Counter + + from pypaimon.common.file_io import FileIO + + file_io = FileIO.get("file:///tmp", {}) + streams = Counter() + lock = threading.Lock() + + class PositionalStream: + def __init__(self, path): + self.path = path + + def read_at(self, length, offset): + return self.path[0].encode() * length + + def close(self): + pass + + def new_input_stream(path): + with lock: + streams[path] += 1 + return PositionalStream(path) + + file_io.new_input_stream = new_input_stream + ranges = ( + [("hot", index * 2, 1) for index in range(9900)] + + [("cold", index * 2, 1) for index in range(100)] + ) + + result = file_io.read_ranges_coalesced( + ranges, parallelism=64, max_gap=0) + + self.assertEqual([b"h"] * 9900 + [b"c"] * 100, result) + self.assertEqual(Counter({"hot": 16, "cold": 16}), streams) + def test_stream_count_is_bounded_across_paths(self): from pypaimon.common.file_io import FileIO From 3c62c78a28c17cf240c185b2bcfdb2ed7a9bd85a Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Sun, 9 Aug 2026 22:55:56 -0700 Subject: [PATCH 15/16] python: preserve capped path lane memberships --- paimon-python/pypaimon/common/file_io.py | 15 ++++++- paimon-python/pypaimon/tests/blob_test.py | 50 +++++++++++++++++++++++ 2 files changed, 63 insertions(+), 2 deletions(-) diff --git a/paimon-python/pypaimon/common/file_io.py b/paimon-python/pypaimon/common/file_io.py index 091d8820b719..001d47611059 100644 --- a/paimon-python/pypaimon/common/file_io.py +++ b/paimon-python/pypaimon/common/file_io.py @@ -257,10 +257,21 @@ def _read_ranges_coalesced(self, ranges, parallelism, max_gap, max_span, min(len(path_tasks), _MAX_RANGE_LANES_PER_PATH) for path_tasks in path_task_groups ] - path_lane_counts = [1] * len(path_task_groups) + path_lane_counts = [ + min( + capacity, + max( + 1, + (workers * len(path_tasks) + task_count - 1) + // task_count, + ), + ) + for path_tasks, capacity in zip( + path_task_groups, path_capacities) + ] remaining_lanes = max( 0, - min(workers, sum(path_capacities)) - len(path_task_groups), + min(workers, sum(path_capacities)) - sum(path_lane_counts), ) for _ in range(remaining_lanes): candidates = [ diff --git a/paimon-python/pypaimon/tests/blob_test.py b/paimon-python/pypaimon/tests/blob_test.py index 1dedbb8a2c04..4e672bc2eeeb 100644 --- a/paimon-python/pypaimon/tests/blob_test.py +++ b/paimon-python/pypaimon/tests/blob_test.py @@ -4013,6 +4013,56 @@ def new_input_stream(path): self.assertEqual([b"h"] * 9900 + [b"c"] * 100, result) self.assertEqual(Counter({"hot": 16, "cold": 16}), streams) + def test_path_memberships_can_exceed_worker_count(self): + from collections import Counter + + from pypaimon.common.file_io import FileIO + + file_io = FileIO.get("file:///tmp", {}) + streams = Counter() + lock = threading.Lock() + active_streams = 0 + max_active_streams = 0 + + class PositionalStream: + def __init__(self, path): + self.path = path + self.closed = False + + def read_at(self, length, offset): + return self.path[0].encode() * length + + def close(self): + nonlocal active_streams + if self.closed: + return + self.closed = True + with lock: + active_streams -= 1 + + def new_input_stream(path): + nonlocal active_streams, max_active_streams + with lock: + streams[path] += 1 + active_streams += 1 + max_active_streams = max( + max_active_streams, active_streams) + return PositionalStream(path) + + file_io.new_input_stream = new_input_stream + ranges = ( + [("hot", index * 2, 1) for index in range(10000)] + + [("cold-%d" % index, 0, 1) for index in range(63)] + ) + + result = file_io.read_ranges_coalesced( + ranges, parallelism=64, max_gap=0) + + self.assertEqual(10063, len(result)) + self.assertEqual(16, streams["hot"]) + self.assertLessEqual(max_active_streams, 64) + self.assertEqual(0, active_streams) + def test_stream_count_is_bounded_across_paths(self): from pypaimon.common.file_io import FileIO From 03d9a56f48f789c2ce77cd9c372c7de2b44f486c Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Sun, 9 Aug 2026 23:54:58 -0700 Subject: [PATCH 16/16] [python] Balance blob range lanes by IO cost --- paimon-python/pypaimon/common/file_io.py | 38 ++++++++++++------ paimon-python/pypaimon/tests/blob_test.py | 48 +++++++++++++++++++++-- 2 files changed, 70 insertions(+), 16 deletions(-) diff --git a/paimon-python/pypaimon/common/file_io.py b/paimon-python/pypaimon/common/file_io.py index 001d47611059..10406849b341 100644 --- a/paimon-python/pypaimon/common/file_io.py +++ b/paimon-python/pypaimon/common/file_io.py @@ -57,6 +57,7 @@ def pread(stream, length: int, offset: int) -> bytes: _COALESCE_VIEW_MAX_RETAINED_AMPLIFICATION = 2.0 # Bound per-object opens; 16 cuts them by 75% for default 64-range batches. _MAX_RANGE_LANES_PER_PATH = 16 +_RANGE_REQUEST_WEIGHT = 1 << 20 def create_temp_path(path: str) -> str: @@ -250,9 +251,19 @@ def _read_ranges_coalesced(self, ranges, parallelism, max_gap, max_span, workers = max(1, min(parallelism, task_count)) + def _task_weight(task): + kind, payload = task + length = payload[2] if kind == "span" else payload[3] + return _RANGE_REQUEST_WEIGHT + max(0, length) + lanes = [[] for _ in range(workers)] lane_loads = [0] * workers path_task_groups = list(tasks_by_path.values()) + path_loads = [ + sum(_task_weight(task) for task in path_tasks) + for path_tasks in path_task_groups + ] + total_load = sum(path_loads) path_capacities = [ min(len(path_tasks), _MAX_RANGE_LANES_PER_PATH) for path_tasks in path_task_groups @@ -262,12 +273,10 @@ def _read_ranges_coalesced(self, ranges, parallelism, max_gap, max_span, capacity, max( 1, - (workers * len(path_tasks) + task_count - 1) - // task_count, + (workers * path_load + total_load - 1) // total_load, ), ) - for path_tasks, capacity in zip( - path_task_groups, path_capacities) + for path_load, capacity in zip(path_loads, path_capacities) ] remaining_lanes = max( 0, @@ -283,8 +292,7 @@ def _read_ranges_coalesced(self, ranges, parallelism, max_gap, max_span, index = max( candidates, key=lambda value: ( - len(path_task_groups[value]) - / path_lane_counts[value] + path_loads[value] / path_lane_counts[value] ), ) path_lane_counts[index] += 1 @@ -293,10 +301,10 @@ def _read_ranges_coalesced(self, ranges, parallelism, max_gap, max_span, path_task_groups, path_lane_counts): selected = sorted( range(workers), key=lane_loads.__getitem__)[:path_lanes] - for index, task in enumerate(path_tasks): - lane = selected[index % path_lanes] + for task in sorted(path_tasks, key=_task_weight, reverse=True): + lane = min(selected, key=lane_loads.__getitem__) lanes[lane].append(task) - lane_loads[lane] += 1 + lane_loads[lane] += _task_weight(task) lanes = [lane for lane in lanes if lane] class _RangeLane: @@ -311,17 +319,21 @@ def _close_current(self): self._stream = None self._path = None if stream is None: - return + return None try: stream.close() except BaseException as error: if self._close_error is None: self._close_error = error + return error + return None def _stream_for(self, path): if self._stream is not None and self._path == path: return self._stream - self._close_current() + close_error = self._close_current() + if close_error is not None: + raise close_error self._stream = self._file_io.new_input_stream(path) self._path = path return self._stream @@ -334,8 +346,10 @@ def read(self, path, offset, length): stream.seek(offset) return (stream.read() if length < 0 else stream.read(length)) - except Exception: + except Exception as read_error: self._close_current() + if self._close_error is not None: + raise read_error return self._file_io.read_file_range( path, offset, length) diff --git a/paimon-python/pypaimon/tests/blob_test.py b/paimon-python/pypaimon/tests/blob_test.py index 4e672bc2eeeb..2c7ea1beda56 100644 --- a/paimon-python/pypaimon/tests/blob_test.py +++ b/paimon-python/pypaimon/tests/blob_test.py @@ -3977,6 +3977,46 @@ def new_input_stream(_): self.assertEqual(64, sum(stream.reads for stream in streams)) self.assertTrue(all(stream.closed for stream in streams)) + def test_same_path_lanes_balance_estimated_io(self): + from pypaimon.common.file_io import FileIO + + large = 8 << 20 + file_io = FileIO.get("file:///tmp", {}) + streams = [] + + class PositionalStream: + def __init__(self): + self.lengths = [] + + def read_at(self, length, offset): + self.lengths.append(length) + return b"x" + + def close(self): + pass + + def new_input_stream(_): + stream = PositionalStream() + streams.append(stream) + return stream + + file_io.new_input_stream = new_input_stream + ranges = [] + offset = 0 + for index in range(256): + length = large if index % 16 == 0 else 1 + ranges.append(("blob", offset, length)) + offset += length + 1 + + file_io.read_ranges_coalesced( + ranges, parallelism=64, max_gap=0) + + self.assertEqual(16, len(streams)) + self.assertEqual( + [1] * 16, + sorted(stream.lengths.count(large) for stream in streams), + ) + def test_skewed_paths_redistribute_capped_lanes(self): from collections import Counter @@ -4165,14 +4205,14 @@ def fail_fallback(path, offset, length): file_io.new_input_stream = FailingStream file_io.read_file_range = fail_fallback - with self.assertRaisesRegex(IOError, "fallback read failed"): + with self.assertRaisesRegex(IOError, "shared read failed"): file_io.read_ranges_coalesced( [("close-error", 0, 2), ("read-error", 0, 2)], parallelism=2, max_gap=0, ) - def test_failed_stream_close_error_is_propagated_after_fallback(self): + def test_failed_stream_close_stops_before_fallback(self): from pypaimon.common.file_io import FileIO file_io = FileIO.get("file:///tmp", {}) @@ -4192,10 +4232,10 @@ def read_file_range(path, offset, length): file_io.new_input_stream = lambda _: FailingStream() file_io.read_file_range = read_file_range - with self.assertRaisesRegex(IOError, "discard close failed"): + with self.assertRaisesRegex(IOError, "pooled read failed"): file_io.read_ranges_coalesced( [("blob", 0, 2)], parallelism=1, max_gap=0) - self.assertEqual([("blob", 0, 2)], fallbacks) + self.assertEqual([], fallbacks) class ReadFileRangeTest(unittest.TestCase):