JingsongLi commented on code in PR #9133:
URL: https://github.com/apache/paimon/pull/9133#discussion_r3746805972
##########
paimon-python/pypaimon/common/file_io.py:
##########
@@ -53,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
Review Comment:
Could we document and justify this fixed value? `parallelism` already bounds
global concurrency, while this additionally changes single-path behavior: for
example, Daft's default `max_concurrency=64` is silently reduced to 16 for one
object. The current benchmark uses parallelism 8, so it does not establish why
16 is the right resource/performance trade-off.
##########
paimon-python/pypaimon/tests/blob_test.py:
##########
@@ -3711,6 +3740,376 @@ def read_file_range(file_path, offset, length):
self.assertEqual(reads, [(path, 0, 1010)])
self.assertIs(shared[0].obj, shared[1].obj)
+ def test_lane_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 = []
+
+ class FailingStream:
+ def read_at(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 data[offset:offset + length]
+
+ file_io.new_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_fallback_reads_do_not_exceed_parallelism(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:
+ 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_known_and_unknown_lengths_share_exclusive_lane(self):
+ from pypaimon.common.file_io import FileIO
+
+ file_io = FileIO.get("file:///tmp", {})
+ operations = []
+ streams = []
+
+ 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):
+ operations.append(("seek", offset))
+ self.position = offset
+
+ def read(self):
+ operations.append(("read", self.position))
+ return b"tail"
+
+ def close(self):
+ self.closed = True
+
+ def new_input_stream(_):
+ stream = LaneStream()
+ streams.append(stream)
+ return stream
+
+ 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"],
+ file_io.read_ranges_coalesced(
+ [("blob", 0, 5), ("blob", 5, -1)],
+ parallelism=1,
+ max_gap=0,
+ ),
+ )
+ 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
+
+ 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
+
+ streams = []
+
+ def new_input_stream(_):
+ stream = SerialStream()
+ streams.append(stream)
+ return stream
+
+ file_io.new_input_stream = new_input_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),
+ )
+ self.assertGreater(len(streams), 1)
+ self.assertLessEqual(len(streams), 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", {})
+ streams = []
+
+ class PositionalStream:
+ def __init__(self):
+ self.reading = False
+ self.reads = 0
+ self.closed = False
+
+ def read_at(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_input_stream(_):
+ stream = PositionalStream()
+ streams.append(stream)
+ return stream
+
+ file_io.new_input_stream = new_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=64, max_gap=0),
+ )
+ self.assertGreater(len(streams), 1)
+ self.assertLessEqual(len(streams), 16)
Review Comment:
Could this test assert the expected allocation more precisely? `1 <
len(streams) <= 16` would still pass if a regression allocated only two lanes.
For 64 tasks with parallelism 64, the current policy should produce exactly 16
streams, with four reads per stream. It would also be valuable to add a skewed
multi-path case that catches lanes lost after applying the cap.
##########
paimon-python/pypaimon/common/file_io.py:
##########
@@ -232,12 +236,87 @@ 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))
+ task_count = sum(len(path_tasks)
+ for path_tasks in tasks_by_path.values())
+ if task_count == 0:
+ return results
- def _run(task):
+ 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)
Review Comment:
Could we redistribute lanes after applying the per-path cap?
`proportional_lanes` is calculated against the uncapped task distribution, so a
hot path permanently discards the lanes removed by the cap. With
`parallelism=64` and path task counts `[9900, 100]`, this assigns only `16 + 1
= 17` non-empty lanes even though 32 are available under the cap; `[1000, 100,
100, 100]` uses only 31 even though all 64 could be used. A capped/water-fill
allocation (or another rebalancing pass) would avoid this substantial
underutilization.
##########
paimon-python/pypaimon/tests/blob_test.py:
##########
@@ -3711,6 +3740,376 @@ def read_file_range(file_path, offset, length):
self.assertEqual(reads, [(path, 0, 1010)])
self.assertIs(shared[0].obj, shared[1].obj)
+ def test_lane_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 = []
+
+ class FailingStream:
+ def read_at(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 data[offset:offset + length]
+
+ file_io.new_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_fallback_reads_do_not_exceed_parallelism(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:
+ def __init__(self):
+ self.closed = False
+ opened()
+
+ def read_at(self, length, offset):
+ barrier.wait()
Review Comment:
Please give this barrier a timeout and explicitly assert that it was not
broken. If a scheduling regression creates fewer than eight concurrent lanes,
the current test will hang CI indefinitely instead of failing with a useful
diagnostic.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
