This is an automated email from the ASF dual-hosted git repository. XiaoHongbo-Hope pushed a commit to branch release-2.1 in repository https://gitbox.apache.org/repos/asf/paimon.git
commit d6644bbc979af08e343b36fe5daad95d16c651c2 Author: XiaoHongbo <[email protected]> AuthorDate: Thu Sep 17 11:31:27 2026 +0800 [python] Make BLOB I/O merge parameters configurable (#9890) --- paimon-python/README.md | 19 ++++ paimon-python/pypaimon/common/file_io.py | 22 ++-- paimon-python/pypaimon/common/options/config.py | 20 ++++ paimon-python/pypaimon/tests/blob_test.py | 132 +++++++++++++++++------- 4 files changed, 150 insertions(+), 43 deletions(-) diff --git a/paimon-python/README.md b/paimon-python/README.md index c745cdc261..b3541b1bdb 100644 --- a/paimon-python/README.md +++ b/paimon-python/README.md @@ -207,6 +207,25 @@ 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-block` 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-block": "16 mb", +}) +``` + +`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/file_io.py b/paimon-python/pypaimon/common/file_io.py index 720f9ae510..3baf4813a5 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 @@ class FileIO(ABC): finally: stream.close() - def read_ranges_coalesced(self, ranges, parallelism, - max_gap=_COALESCE_GAP, max_span=_COALESCE_SPAN): + 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,12 +193,12 @@ class FileIO(ABC): 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() 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, + def read_ranges_coalesced_views(self, ranges, parallelism, *, max_retained_amplification=( _COALESCE_VIEW_MAX_RETAINED_AMPLIFICATION)): """Read coalesced ranges as zero-copy ``memoryview`` slices. @@ -216,10 +211,21 @@ class FileIO(ABC): 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() return self._read_ranges_coalesced( ranges, parallelism, max_gap, max_span, max_retained_amplification, return_views=True) + def _resolve_coalesce_limits(self): + from pypaimon.common.options.config import FileIOOptions + properties = getattr(self, "properties", None) + 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): from concurrent.futures import ThreadPoolExecutor diff --git a/paimon-python/pypaimon/common/options/config.py b/paimon-python/pypaimon/common/options/config.py index 98a466496b..80095404ff 100644 --- a/paimon-python/pypaimon/common/options/config.py +++ b/paimon-python/pypaimon/common/options/config.py @@ -146,6 +146,26 @@ 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_BLOCK = ( + ConfigOptions.key("file-io.read-coalesce.max-block") + .memory_type() + .default_value(MemorySize.of_mebi_bytes(8)) + .with_description( + "Maximum span for coalescing same-file ranges, except when an " + "individual range is larger. Individual ranges are not split." + ) + ) + + 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 456d8dc2b2..00a721105e 100644 --- a/paimon-python/pypaimon/tests/blob_test.py +++ b/paimon-python/pypaimon/tests/blob_test.py @@ -4135,6 +4135,68 @@ class CoalesceRangesTest(unittest.TestCase): 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_BLOCK).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-block": "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.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): from pypaimon.common.file_io import FileIO data = bytes(range(256)) * 4 @@ -4142,11 +4204,13 @@ class CoalesceRangesTest(unittest.TestCase): 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) @@ -4170,7 +4234,10 @@ class CoalesceRangesTest(unittest.TestCase): 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 @@ -4191,8 +4258,6 @@ class CoalesceRangesTest(unittest.TestCase): 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)]) @@ -4204,8 +4269,6 @@ class CoalesceRangesTest(unittest.TestCase): 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)]) @@ -4219,7 +4282,9 @@ class CoalesceRangesTest(unittest.TestCase): 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: @@ -4239,8 +4304,7 @@ class CoalesceRangesTest(unittest.TestCase): 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)) @@ -4297,7 +4361,7 @@ class CoalesceRangesTest(unittest.TestCase): 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) @@ -4344,7 +4408,6 @@ class CoalesceRangesTest(unittest.TestCase): file_io.read_ranges_coalesced( [("blob", 0, 5), ("blob", 5, -1)], parallelism=1, - max_gap=0, ), ) self.assertEqual([ @@ -4359,7 +4422,8 @@ class CoalesceRangesTest(unittest.TestCase): 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): @@ -4396,8 +4460,7 @@ class CoalesceRangesTest(unittest.TestCase): 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) @@ -4406,7 +4469,8 @@ class CoalesceRangesTest(unittest.TestCase): 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: @@ -4440,8 +4504,7 @@ class CoalesceRangesTest(unittest.TestCase): 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)) @@ -4452,7 +4515,8 @@ class CoalesceRangesTest(unittest.TestCase): 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: @@ -4479,8 +4543,7 @@ class CoalesceRangesTest(unittest.TestCase): 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( @@ -4493,7 +4556,8 @@ class CoalesceRangesTest(unittest.TestCase): 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() @@ -4518,8 +4582,7 @@ class CoalesceRangesTest(unittest.TestCase): + [("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) @@ -4529,7 +4592,8 @@ class CoalesceRangesTest(unittest.TestCase): 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 @@ -4566,8 +4630,7 @@ class CoalesceRangesTest(unittest.TestCase): + [("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"]) @@ -4577,7 +4640,8 @@ class CoalesceRangesTest(unittest.TestCase): 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 @@ -4610,8 +4674,7 @@ class CoalesceRangesTest(unittest.TestCase): 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) @@ -4621,7 +4684,8 @@ class CoalesceRangesTest(unittest.TestCase): 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: @@ -4647,8 +4711,7 @@ class CoalesceRangesTest(unittest.TestCase): 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)) @@ -4680,7 +4743,6 @@ class CoalesceRangesTest(unittest.TestCase): 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): @@ -4705,7 +4767,7 @@ class CoalesceRangesTest(unittest.TestCase): 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)
