From 8fcec7745f58563c91faedd7c9d402fd916a2860 Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Mon, 14 Sep 2026 06:49:13 -0700 Subject: [PATCH 1/4] [python] Configure coalesced range read bounds --- paimon-python/README.md | 16 +++++ paimon-python/pypaimon/common/file_io.py | 19 +++++- .../pypaimon/common/options/config.py | 17 ++++++ paimon-python/pypaimon/tests/blob_test.py | 61 +++++++++++++++++++ 4 files changed, 111 insertions(+), 2 deletions(-) diff --git a/paimon-python/README.md b/paimon-python/README.md index 2af01d988c54..ac6b870ceac0 100644 --- a/paimon-python/README.md +++ b/paimon-python/README.md @@ -133,6 +133,22 @@ and precomputed primary-key global-index results still use the Python planner. Continuous streaming and write planning also retain their Python entrypoints. Native planning remains optional and is disabled by default. +# Coalesced BLOB reads + +FileIO merges nearby BLOB ranges before reading. Set +`file-io.read-coalesce.max-gap` and `file-io.read-coalesce.max-span` in the +catalog or connection options to tune the 1 MiB and 8 MiB defaults: + +```python +import pypaimon.multimodal as pmm + +connection = pmm.connect(options={ + "warehouse": "/tmp/warehouse", + "file-io.read-coalesce.max-gap": "64 kb", + "file-io.read-coalesce.max-span": "16 mb", +}) +``` + # Load LeRobot Dataset v3 Install the optional dependency, then import a local directory, FileIO URI, or diff --git a/paimon-python/pypaimon/common/file_io.py b/paimon-python/pypaimon/common/file_io.py index 10406849b341..dea4b4855e30 100644 --- a/paimon-python/pypaimon/common/file_io.py +++ b/paimon-python/pypaimon/common/file_io.py @@ -188,7 +188,7 @@ def read_file_range(self, path, offset, length): stream.close() def read_ranges_coalesced(self, ranges, parallelism, - max_gap=_COALESCE_GAP, max_span=_COALESCE_SPAN): + max_gap=None, max_span=None): """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. Each worker lane reuses one exclusive @@ -198,12 +198,13 @@ def read_ranges_coalesced(self, ranges, parallelism, A failed read propagates and aborts the whole batch (unlike a per-row ``file.open()`` loop that fails one row at a time). """ + max_gap, max_span = self._resolve_coalesce_limits(max_gap, max_span) return self._read_ranges_coalesced( ranges, parallelism, max_gap, max_span, max_retained_amplification=0, return_views=False) def read_ranges_coalesced_views(self, ranges, parallelism, - max_gap=_COALESCE_GAP, max_span=_COALESCE_SPAN, + max_gap=None, max_span=None, max_retained_amplification=( _COALESCE_VIEW_MAX_RETAINED_AMPLIFICATION)): """Read coalesced ranges as zero-copy ``memoryview`` slices. @@ -216,10 +217,24 @@ def read_ranges_coalesced_views(self, ranges, parallelism, excessive gap bytes; set ``max_retained_amplification`` to a non-positive value to always share the merged buffer. """ + max_gap, max_span = self._resolve_coalesce_limits(max_gap, max_span) return self._read_ranges_coalesced( ranges, parallelism, max_gap, max_span, max_retained_amplification, return_views=True) + def _resolve_coalesce_limits(self, max_gap, max_span): + from pypaimon.common.options.config import FileIOOptions + properties = getattr(self, "properties", None) + if max_gap is None: + max_gap = (properties.get(FileIOOptions.READ_COALESCE_MAX_GAP) + .get_bytes() if isinstance(properties, Options) + else _COALESCE_GAP) + if max_span is None: + max_span = (properties.get(FileIOOptions.READ_COALESCE_MAX_SPAN) + .get_bytes() if isinstance(properties, Options) + else _COALESCE_SPAN) + return max_gap, max_span + def _read_ranges_coalesced(self, ranges, parallelism, max_gap, max_span, max_retained_amplification, return_views): from concurrent.futures import ThreadPoolExecutor diff --git a/paimon-python/pypaimon/common/options/config.py b/paimon-python/pypaimon/common/options/config.py index 5684ab62c3c6..9742a41e443e 100644 --- a/paimon-python/pypaimon/common/options/config.py +++ b/paimon-python/pypaimon/common/options/config.py @@ -134,6 +134,23 @@ class CatalogOptions: BLOB_FILE_IO_DEFAULT_CACHE_SIZE = 2 ** 31 - 1 +class FileIOOptions: + READ_COALESCE_MAX_GAP = ( + ConfigOptions.key("file-io.read-coalesce.max-gap") + .memory_type() + .default_value(MemorySize.of_mebi_bytes(1)) + .with_description( + "Maximum gap between same-file ranges merged into one read." + ) + ) + READ_COALESCE_MAX_SPAN = ( + ConfigOptions.key("file-io.read-coalesce.max-span") + .memory_type() + .default_value(MemorySize.of_mebi_bytes(8)) + .with_description("Maximum span for merging same-file ranges.") + ) + + class HdfsOptions: HDFS_CLIENT_IMPL = ( ConfigOptions.key("hdfs.client.impl") diff --git a/paimon-python/pypaimon/tests/blob_test.py b/paimon-python/pypaimon/tests/blob_test.py index 456d8dc2b2a1..bbe7b2172ad3 100644 --- a/paimon-python/pypaimon/tests/blob_test.py +++ b/paimon-python/pypaimon/tests/blob_test.py @@ -4135,6 +4135,67 @@ def test_read_ranges_coalesced(self): self.assertEqual(got[4], data[100:]) # length -1 => read to EOF self.assertIsNone(got[5]) # None offset/length => skipped + def test_coalesce_limits_from_file_io_options(self): + from pypaimon.common.options.config import FileIOOptions + self.assertEqual(1 << 20, Options({}).get( + FileIOOptions.READ_COALESCE_MAX_GAP).get_bytes()) + self.assertEqual(8 << 20, Options({}).get( + FileIOOptions.READ_COALESCE_MAX_SPAN).get_bytes()) + data = bytes(range(256)) + with tempfile.TemporaryDirectory() as tmp_dir: + path = os.path.join(tmp_dir, "f.bin") + with open(path, "wb") as output: + output.write(data) + file_io = FileIO.get( + "file://" + tmp_dir, + Options({ + "file-io.read-coalesce.max-gap": "64 b", + "file-io.read-coalesce.max-span": "100 b", + }), + ) + reads = [] + original_open = file_io.new_input_stream + + def new_input_stream(file_path): + stream = original_open(file_path) + + class TrackingStream: + def read_at(self, length, offset): + reads.append((offset, length)) + return os.pread(stream.fileno(), length, offset) + + def close(self): + stream.close() + + return TrackingStream() + + file_io.new_input_stream = new_input_stream + ranges = [(path, 0, 10), (path, 60, 10), (path, 120, 10)] + + self.assertEqual( + [data[0:10], data[60:70], data[120:130]], + file_io.read_ranges_coalesced(ranges, parallelism=3), + ) + self.assertEqual([(0, 70), (120, 10)], sorted(reads)) + + reads.clear() + views = file_io.read_ranges_coalesced_views(ranges, parallelism=3) + self.assertEqual([data[0:10], data[60:70], data[120:130]], + [bytes(view) for view in views]) + self.assertEqual([(0, 70), (120, 10)], sorted(reads)) + + reads.clear() + file_io.properties.set( + FileIOOptions.READ_COALESCE_MAX_GAP, "0 b") + file_io.read_ranges_coalesced( + ranges, parallelism=3) + self.assertEqual([(0, 10), (60, 10), (120, 10)], sorted(reads)) + + reads.clear() + file_io.read_ranges_coalesced( + ranges, parallelism=3, max_gap=64, max_span=100) + self.assertEqual([(0, 70), (120, 10)], sorted(reads)) + def test_read_ranges_coalesced_views(self): from pypaimon.common.file_io import FileIO data = bytes(range(256)) * 4 From 501df001ee4357c94d930aabfb1d76d38c6dd6e5 Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Wed, 16 Sep 2026 03:27:05 -0700 Subject: [PATCH 2/4] [python] Name coalesced read limit max-block --- paimon-python/README.md | 4 ++-- paimon-python/pypaimon/common/file_io.py | 2 +- paimon-python/pypaimon/common/options/config.py | 6 +++--- paimon-python/pypaimon/tests/blob_test.py | 4 ++-- 4 files changed, 8 insertions(+), 8 deletions(-) diff --git a/paimon-python/README.md b/paimon-python/README.md index ac6b870ceac0..e4acf903934f 100644 --- a/paimon-python/README.md +++ b/paimon-python/README.md @@ -136,7 +136,7 @@ Native planning remains optional and is disabled by default. # Coalesced BLOB reads FileIO merges nearby BLOB ranges before reading. Set -`file-io.read-coalesce.max-gap` and `file-io.read-coalesce.max-span` in the +`file-io.read-coalesce.max-gap` and `file-io.read-coalesce.max-block` in the catalog or connection options to tune the 1 MiB and 8 MiB defaults: ```python @@ -145,7 +145,7 @@ import pypaimon.multimodal as pmm connection = pmm.connect(options={ "warehouse": "/tmp/warehouse", "file-io.read-coalesce.max-gap": "64 kb", - "file-io.read-coalesce.max-span": "16 mb", + "file-io.read-coalesce.max-block": "16 mb", }) ``` diff --git a/paimon-python/pypaimon/common/file_io.py b/paimon-python/pypaimon/common/file_io.py index dea4b4855e30..30abe967c296 100644 --- a/paimon-python/pypaimon/common/file_io.py +++ b/paimon-python/pypaimon/common/file_io.py @@ -230,7 +230,7 @@ def _resolve_coalesce_limits(self, max_gap, max_span): .get_bytes() if isinstance(properties, Options) else _COALESCE_GAP) if max_span is None: - max_span = (properties.get(FileIOOptions.READ_COALESCE_MAX_SPAN) + max_span = (properties.get(FileIOOptions.READ_COALESCE_MAX_BLOCK) .get_bytes() if isinstance(properties, Options) else _COALESCE_SPAN) return max_gap, max_span diff --git a/paimon-python/pypaimon/common/options/config.py b/paimon-python/pypaimon/common/options/config.py index 9742a41e443e..5ef4858a4596 100644 --- a/paimon-python/pypaimon/common/options/config.py +++ b/paimon-python/pypaimon/common/options/config.py @@ -143,11 +143,11 @@ class FileIOOptions: "Maximum gap between same-file ranges merged into one read." ) ) - READ_COALESCE_MAX_SPAN = ( - ConfigOptions.key("file-io.read-coalesce.max-span") + READ_COALESCE_MAX_BLOCK = ( + ConfigOptions.key("file-io.read-coalesce.max-block") .memory_type() .default_value(MemorySize.of_mebi_bytes(8)) - .with_description("Maximum span for merging same-file ranges.") + .with_description("Maximum size of a merged same-file read range.") ) diff --git a/paimon-python/pypaimon/tests/blob_test.py b/paimon-python/pypaimon/tests/blob_test.py index bbe7b2172ad3..67ff5763bf07 100644 --- a/paimon-python/pypaimon/tests/blob_test.py +++ b/paimon-python/pypaimon/tests/blob_test.py @@ -4140,7 +4140,7 @@ def test_coalesce_limits_from_file_io_options(self): self.assertEqual(1 << 20, Options({}).get( FileIOOptions.READ_COALESCE_MAX_GAP).get_bytes()) self.assertEqual(8 << 20, Options({}).get( - FileIOOptions.READ_COALESCE_MAX_SPAN).get_bytes()) + FileIOOptions.READ_COALESCE_MAX_BLOCK).get_bytes()) data = bytes(range(256)) with tempfile.TemporaryDirectory() as tmp_dir: path = os.path.join(tmp_dir, "f.bin") @@ -4150,7 +4150,7 @@ def test_coalesce_limits_from_file_io_options(self): "file://" + tmp_dir, Options({ "file-io.read-coalesce.max-gap": "64 b", - "file-io.read-coalesce.max-span": "100 b", + "file-io.read-coalesce.max-block": "100 b", }), ) reads = [] From 81a67aadf5dde342eb91565f29ec9d9bb34095e5 Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Wed, 16 Sep 2026 03:35:09 -0700 Subject: [PATCH 3/4] [python] Resolve coalesced read limits from FileIO options --- paimon-python/pypaimon/common/file_io.py | 31 ++++------ paimon-python/pypaimon/tests/blob_test.py | 75 ++++++++++++----------- 2 files changed, 49 insertions(+), 57 deletions(-) diff --git a/paimon-python/pypaimon/common/file_io.py b/paimon-python/pypaimon/common/file_io.py index 30abe967c296..4759ca26baa3 100644 --- a/paimon-python/pypaimon/common/file_io.py +++ b/paimon-python/pypaimon/common/file_io.py @@ -50,10 +50,6 @@ def pread(stream, length: int, offset: int) -> bytes: return os.pread(stream.fileno(), length, offset) -# 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 # Bound per-object opens; 16 cuts them by 75% for default 64-range batches. _MAX_RANGE_LANES_PER_PATH = 16 @@ -187,8 +183,7 @@ def read_file_range(self, path, offset, length): finally: stream.close() - def read_ranges_coalesced(self, ranges, parallelism, - max_gap=None, max_span=None): + def read_ranges_coalesced(self, ranges, parallelism): """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. Each worker lane reuses one exclusive @@ -198,13 +193,12 @@ def read_ranges_coalesced(self, ranges, parallelism, A failed read propagates and aborts the whole batch (unlike a per-row ``file.open()`` loop that fails one row at a time). """ - max_gap, max_span = self._resolve_coalesce_limits(max_gap, max_span) + max_gap, max_span = self._resolve_coalesce_limits() return self._read_ranges_coalesced( ranges, parallelism, max_gap, max_span, max_retained_amplification=0, return_views=False) - def read_ranges_coalesced_views(self, ranges, parallelism, - max_gap=None, max_span=None, + def read_ranges_coalesced_views(self, ranges, parallelism, *, max_retained_amplification=( _COALESCE_VIEW_MAX_RETAINED_AMPLIFICATION)): """Read coalesced ranges as zero-copy ``memoryview`` slices. @@ -217,23 +211,20 @@ def read_ranges_coalesced_views(self, ranges, parallelism, excessive gap bytes; set ``max_retained_amplification`` to a non-positive value to always share the merged buffer. """ - max_gap, max_span = self._resolve_coalesce_limits(max_gap, max_span) + max_gap, max_span = self._resolve_coalesce_limits() return self._read_ranges_coalesced( ranges, parallelism, max_gap, max_span, max_retained_amplification, return_views=True) - def _resolve_coalesce_limits(self, max_gap, max_span): + def _resolve_coalesce_limits(self): from pypaimon.common.options.config import FileIOOptions properties = getattr(self, "properties", None) - if max_gap is None: - max_gap = (properties.get(FileIOOptions.READ_COALESCE_MAX_GAP) - .get_bytes() if isinstance(properties, Options) - else _COALESCE_GAP) - if max_span is None: - max_span = (properties.get(FileIOOptions.READ_COALESCE_MAX_BLOCK) - .get_bytes() if isinstance(properties, Options) - else _COALESCE_SPAN) - return max_gap, max_span + if not isinstance(properties, Options): + properties = Options({}) + return ( + properties.get(FileIOOptions.READ_COALESCE_MAX_GAP).get_bytes(), + properties.get(FileIOOptions.READ_COALESCE_MAX_BLOCK).get_bytes(), + ) def _read_ranges_coalesced(self, ranges, parallelism, max_gap, max_span, max_retained_amplification, return_views): diff --git a/paimon-python/pypaimon/tests/blob_test.py b/paimon-python/pypaimon/tests/blob_test.py index 67ff5763bf07..00a721105e39 100644 --- a/paimon-python/pypaimon/tests/blob_test.py +++ b/paimon-python/pypaimon/tests/blob_test.py @@ -4192,8 +4192,9 @@ def close(self): self.assertEqual([(0, 10), (60, 10), (120, 10)], sorted(reads)) reads.clear() - file_io.read_ranges_coalesced( - ranges, parallelism=3, max_gap=64, max_span=100) + file_io.properties.set( + FileIOOptions.READ_COALESCE_MAX_GAP, "64 b") + file_io.read_ranges_coalesced(ranges, parallelism=3) self.assertEqual([(0, 70), (120, 10)], sorted(reads)) def test_read_ranges_coalesced_views(self): @@ -4203,11 +4204,13 @@ def test_read_ranges_coalesced_views(self): path = os.path.join(tmp_dir, "f.bin") with open(path, 'wb') as output: output.write(data) - file_io = FileIO.get(f"file://{tmp_dir}", {}) + file_io = FileIO.get( + f"file://{tmp_dir}", + Options({"file-io.read-coalesce.max-gap": "100 b"})) ranges = [(path, 0, 10), (path, 10, 10), None, (path, 500, 20), (path, 100, -1)] got = file_io.read_ranges_coalesced_views( - ranges, parallelism=4, max_gap=100) + ranges, parallelism=4) self.assertIsInstance(got[0], memoryview) self.assertIsInstance(got[1], memoryview) @@ -4231,7 +4234,10 @@ def test_sparse_views_preserve_coalesced_read(self): path = os.path.join(tmp_dir, "f.bin") with open(path, 'wb') as output: output.write(data) - file_io = FileIO.get(f"file://{tmp_dir}", {}) + file_io = FileIO.get( + f"file://{tmp_dir}", + Options({"file-io.read-coalesce.max-gap": "1000 b", + "file-io.read-coalesce.max-block": "1 mb"})) reads = [] original_open = file_io.new_input_stream @@ -4252,8 +4258,6 @@ def close(self): got = file_io.read_ranges_coalesced_views( [(path, 0, 10), (path, 1000, 10)], parallelism=4, - max_gap=1000, - max_span=1 << 20, ) self.assertEqual(reads, [(path, 0, 1010)]) @@ -4265,8 +4269,6 @@ def close(self): shared = file_io.read_ranges_coalesced_views( [(path, 0, 10), (path, 1000, 10)], parallelism=4, - max_gap=1000, - max_span=1 << 20, max_retained_amplification=0, ) self.assertEqual(reads, [(path, 0, 1010)]) @@ -4280,7 +4282,9 @@ def test_lane_stream_failure_reopens_range(self): 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}", {}) + file_io = FileIO.get( + f"file://{tmp_dir}", + Options({"file-io.read-coalesce.max-gap": "0 b"})) fallbacks = [] class FailingStream: @@ -4300,8 +4304,7 @@ def read_file_range(file_path, offset, length): self.assertEqual( [data[0:4], data[16:20]], - file_io.read_ranges_coalesced( - ranges, parallelism=2, max_gap=0), + file_io.read_ranges_coalesced(ranges, parallelism=2), ) self.assertEqual(2, len(fallbacks)) @@ -4358,7 +4361,7 @@ def read_file_range(path, offset, length): self.assertEqual( [b"ok"] * parallelism, file_io.read_ranges_coalesced( - ranges, parallelism=parallelism, max_gap=0), + ranges, parallelism=parallelism), ) self.assertLessEqual(max_open_streams, parallelism) self.assertEqual(0, open_streams) @@ -4405,7 +4408,6 @@ def new_input_stream(_): file_io.read_ranges_coalesced( [("blob", 0, 5), ("blob", 5, -1)], parallelism=1, - max_gap=0, ), ) self.assertEqual([ @@ -4420,7 +4422,8 @@ def test_non_positional_streams_are_exclusive(self): from pypaimon.common.file_io import FileIO data = bytes(range(128)) - file_io = FileIO.get("file:///tmp", {}) + file_io = FileIO.get( + "file:///tmp", Options({"file-io.read-coalesce.max-gap": "0 b"})) class SerialStream: def __init__(self): @@ -4457,8 +4460,7 @@ def new_input_stream(_): self.assertEqual( [data[i * 4:i * 4 + 2] for i in range(16)], - file_io.read_ranges_coalesced( - ranges, parallelism=8, max_gap=0), + file_io.read_ranges_coalesced(ranges, parallelism=8), ) self.assertGreater(len(streams), 1) self.assertLessEqual(len(streams), 8) @@ -4467,7 +4469,8 @@ def test_same_path_reuses_bounded_exclusive_lanes(self): from pypaimon.common.file_io import FileIO data = bytes(range(256)) * 64 - file_io = FileIO.get("file:///tmp", {}) + file_io = FileIO.get( + "file:///tmp", Options({"file-io.read-coalesce.max-gap": "0 b"})) streams = [] class PositionalStream: @@ -4501,8 +4504,7 @@ def new_input_stream(_): self.assertEqual( [data[offset:offset + length] for _, offset, length in ranges], - file_io.read_ranges_coalesced( - ranges, parallelism=64, max_gap=0), + file_io.read_ranges_coalesced(ranges, parallelism=64), ) self.assertEqual(16, len(streams)) self.assertTrue(all(stream.reads == 4 for stream in streams)) @@ -4513,7 +4515,8 @@ 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", {}) + file_io = FileIO.get( + "file:///tmp", Options({"file-io.read-coalesce.max-gap": "0 b"})) streams = [] class PositionalStream: @@ -4540,8 +4543,7 @@ def new_input_stream(_): ranges.append(("blob", offset, length)) offset += length + 1 - file_io.read_ranges_coalesced( - ranges, parallelism=64, max_gap=0) + file_io.read_ranges_coalesced(ranges, parallelism=64) self.assertEqual(16, len(streams)) self.assertEqual( @@ -4554,7 +4556,8 @@ def test_skewed_paths_redistribute_capped_lanes(self): from pypaimon.common.file_io import FileIO - file_io = FileIO.get("file:///tmp", {}) + file_io = FileIO.get( + "file:///tmp", Options({"file-io.read-coalesce.max-gap": "0 b"})) streams = Counter() lock = threading.Lock() @@ -4579,8 +4582,7 @@ def new_input_stream(path): + [("cold", index * 2, 1) for index in range(100)] ) - result = file_io.read_ranges_coalesced( - ranges, parallelism=64, max_gap=0) + result = file_io.read_ranges_coalesced(ranges, parallelism=64) self.assertEqual([b"h"] * 9900 + [b"c"] * 100, result) self.assertEqual(Counter({"hot": 16, "cold": 16}), streams) @@ -4590,7 +4592,8 @@ def test_path_memberships_can_exceed_worker_count(self): from pypaimon.common.file_io import FileIO - file_io = FileIO.get("file:///tmp", {}) + file_io = FileIO.get( + "file:///tmp", Options({"file-io.read-coalesce.max-gap": "0 b"})) streams = Counter() lock = threading.Lock() active_streams = 0 @@ -4627,8 +4630,7 @@ def new_input_stream(path): + [("cold-%d" % index, 0, 1) for index in range(63)] ) - result = file_io.read_ranges_coalesced( - ranges, parallelism=64, max_gap=0) + result = file_io.read_ranges_coalesced(ranges, parallelism=64) self.assertEqual(10063, len(result)) self.assertEqual(16, streams["hot"]) @@ -4638,7 +4640,8 @@ def new_input_stream(path): def test_stream_count_is_bounded_across_paths(self): from pypaimon.common.file_io import FileIO - file_io = FileIO.get("file:///tmp", {}) + file_io = FileIO.get( + "file:///tmp", Options({"file-io.read-coalesce.max-gap": "0 b"})) lock = threading.Lock() open_streams = 0 max_open_streams = 0 @@ -4671,8 +4674,7 @@ def new_input_stream(_): self.assertEqual( [bytes([offset]) * length for _, offset, length in ranges], - file_io.read_ranges_coalesced( - ranges, parallelism=4, max_gap=0), + file_io.read_ranges_coalesced(ranges, parallelism=4), ) self.assertLessEqual(max_open_streams, 4) self.assertEqual(4, total_streams) @@ -4682,7 +4684,8 @@ 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", {}) + file_io = FileIO.get( + "file:///tmp", Options({"file-io.read-coalesce.max-gap": "0 b"})) streams = [] class CloseStream: @@ -4708,8 +4711,7 @@ def 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) + file_io.read_ranges_coalesced(ranges, parallelism=4) self.assertGreater(len(streams), 1) self.assertTrue(all(stream.closed for stream in streams)) @@ -4741,7 +4743,6 @@ def fail_fallback(path, offset, length): file_io.read_ranges_coalesced( [("close-error", 0, 2), ("read-error", 0, 2)], parallelism=2, - max_gap=0, ) def test_failed_stream_close_stops_before_fallback(self): @@ -4766,7 +4767,7 @@ def read_file_range(path, offset, length): with self.assertRaisesRegex(IOError, "pooled read failed"): file_io.read_ranges_coalesced( - [("blob", 0, 2)], parallelism=1, max_gap=0) + [("blob", 0, 2)], parallelism=1) self.assertEqual([], fallbacks) From d33d79d1f5d730251415fb47f108e481ba2d9399 Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Wed, 16 Sep 2026 20:45:50 +0800 Subject: [PATCH 4/4] [python] Clarify coalesced read block limit --- paimon-python/README.md | 3 +++ paimon-python/pypaimon/common/options/config.py | 5 ++++- 2 files changed, 7 insertions(+), 1 deletion(-) diff --git a/paimon-python/README.md b/paimon-python/README.md index e4acf903934f..be2496547167 100644 --- a/paimon-python/README.md +++ b/paimon-python/README.md @@ -149,6 +149,9 @@ connection = pmm.connect(options={ }) ``` +`max-block` constrains coalescing, but does not split an individual BLOB range. +A single read can therefore exceed this value. + # Load LeRobot Dataset v3 Install the optional dependency, then import a local directory, FileIO URI, or diff --git a/paimon-python/pypaimon/common/options/config.py b/paimon-python/pypaimon/common/options/config.py index 5ef4858a4596..356f1064e4a9 100644 --- a/paimon-python/pypaimon/common/options/config.py +++ b/paimon-python/pypaimon/common/options/config.py @@ -147,7 +147,10 @@ class FileIOOptions: ConfigOptions.key("file-io.read-coalesce.max-block") .memory_type() .default_value(MemorySize.of_mebi_bytes(8)) - .with_description("Maximum size of a merged same-file read range.") + .with_description( + "Maximum span for coalescing same-file ranges, except when an " + "individual range is larger. Individual ranges are not split." + ) )