This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new e8dbbbe257 [python] Avoid thread-pool overhead for single-lane range
reads (#10795)
e8dbbbe257 is described below
commit e8dbbbe2572227daef167cd3d2790973faa0c8a9
Author: XiaoHongbo <[email protected]>
AuthorDate: Sun Oct 25 10:45:46 2026 +0800
[python] Avoid thread-pool overhead for single-lane range reads (#10795)
---
paimon-python/pypaimon/common/file_io.py | 7 ++--
paimon-python/pypaimon/tests/blob_test.py | 53 +++++++++++++++++++++++++++++++
2 files changed, 58 insertions(+), 2 deletions(-)
diff --git a/paimon-python/pypaimon/common/file_io.py
b/paimon-python/pypaimon/common/file_io.py
index c2d132f7ac..455acd4379 100644
--- a/paimon-python/pypaimon/common/file_io.py
+++ b/paimon-python/pypaimon/common/file_io.py
@@ -404,8 +404,11 @@ class FileIO(ABC):
close_error = error
return read_error, close_error
- with ThreadPoolExecutor(len(lanes)) as pool:
- outcomes = list(pool.map(_run_lane, lanes))
+ if len(lanes) == 1:
+ outcomes = [_run_lane(lanes[0])]
+ else:
+ 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(
diff --git a/paimon-python/pypaimon/tests/blob_test.py
b/paimon-python/pypaimon/tests/blob_test.py
index b693b8d01f..9d0f5a660b 100644
--- a/paimon-python/pypaimon/tests/blob_test.py
+++ b/paimon-python/pypaimon/tests/blob_test.py
@@ -5972,6 +5972,59 @@ class CoalesceRangesTest(unittest.TestCase):
"""read_ranges_coalesced merges same-file adjacent reads into fewer
requests
(JingsongLi's IO-merging suggestion) while returning identical bytes."""
+ def test_single_lane_runs_without_executor(self):
+ from pypaimon.filesystem.local_file_io import LocalFileIO
+ from unittest.mock import patch
+
+ with tempfile.TemporaryDirectory() as directory:
+ path = os.path.join(directory, "data")
+ with open(path, "wb") as stream:
+ stream.write(b"0123456789")
+ file_io = LocalFileIO()
+ # Adjacent requests coalesce to one lane even with parallelism > 1.
+ for parallelism in (1, 8):
+ with self.subTest(parallelism=parallelism), patch(
+ "concurrent.futures.ThreadPoolExecutor",
+ side_effect=AssertionError("Unexpected executor")):
+ ranges = [(path, 4, 2), (path, 2, 2)]
+ self.assertEqual([b"45", b"23"],
+ file_io.read_ranges_coalesced(ranges,
parallelism))
+ views = file_io.read_ranges_coalesced_views(ranges,
parallelism)
+ self.assertEqual([b"45", b"23"], [bytes(view) for view in
views])
+ self.assertTrue(all(isinstance(view, memoryview) for view
in views))
+ with self.assertRaises(FileNotFoundError):
+ file_io.read_ranges_coalesced(
+ [(path + "-missing", 0, 1)], parallelism)
+
+ def test_single_lane_close_errors(self):
+ for parallelism in (1, 8):
+ for return_views in (False, True):
+ for fail_read in (False, True):
+ with self.subTest(parallelism=parallelism,
+ return_views=return_views,
fail_read=fail_read):
+ read_error = IOError("read failed")
+ close_error = IOError("close failed")
+ stream = MagicMock(spec=["read_at", "close"])
+ stream.read_at.return_value = b"2345"
+ if fail_read:
+ stream.read_at.side_effect = read_error
+ stream.close.side_effect = close_error
+ file_io = LocalFileIO()
+ file_io.new_input_stream =
MagicMock(return_value=stream)
+ file_io.read_file_range = MagicMock()
+ read = (file_io.read_ranges_coalesced_views if
return_views
+ else file_io.read_ranges_coalesced)
+ # Both requests merge into one lane, regardless of
parallelism.
+ with patch("concurrent.futures.ThreadPoolExecutor",
+ side_effect=AssertionError("Unexpected
executor")):
+ with self.assertRaises(IOError) as raised:
+ read([("blob", 4, 2), ("blob", 2, 2)],
parallelism)
+ self.assertIs(raised.exception,
+ read_error if fail_read else close_error)
+ stream.read_at.assert_called_once_with(4, 2)
+ stream.close.assert_called_once_with()
+ file_io.read_file_range.assert_not_called()
+
def test_coalesce_ranges_grouping(self):
from pypaimon.common.file_io import _coalesce_ranges
items = [(0, "a", 0, 10), (1, "a", 10, 10), (2, "a", 1000, 10), (3,
"b", 0, 5)]
