JingsongLi commented on code in PR #9133:
URL: https://github.com/apache/paimon/pull/9133#discussion_r3747231629
##########
paimon-python/pypaimon/common/file_io.py:
##########
@@ -232,12 +237,118 @@ 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
+ 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 = [
+ 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)) - sum(path_lane_counts),
+ )
+ 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_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):
+ lane = selected[index % path_lanes]
Review Comment:
[P2] Balance static lanes by estimated I/O cost
The allocation ratios, `lane_loads`, and this round-robin placement all
treat every task as equal, although a merged span can range from a few bytes to
8 MiB and these static lanes cannot steal work. For one path using 16 lanes,
256 non-coalescing spans with an 8 MiB span at indices `0, 16, ..., 240` send
all sixteen large reads to lane 0 via `index % path_lanes`: that lane transfers
128 MiB while the other 15 lanes process only tiny reads. The previous per-task
executor did not pin this periodic size pattern to one worker. Please assign an
estimated weight to each task, such as `span_len` plus a fixed request cost,
and place each task on the least weighted-load selected lane. A
heterogeneous-span regression test should verify that the large reads are
distributed across lanes.
##########
paimon-python/pypaimon/common/file_io.py:
##########
@@ -232,12 +237,118 @@ 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
+ 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 = [
+ 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)) - sum(path_lane_counts),
+ )
+ 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_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):
+ 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 = None
+ self._stream = None
+ self._close_error = None
+
+ def _close_current(self):
+ stream = self._stream
+ self._stream = None
+ self._path = None
+ if stream is None:
+ return
+ try:
+ stream.close()
+ except BaseException as error:
+ if self._close_error is None:
+ self._close_error = error
+
+ 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)
+ stream.seek(offset)
+ return (stream.read() if length < 0
+ else stream.read(length))
+ except Exception:
+ self._close_current()
+ return self._file_io.read_file_range(
Review Comment:
[P2] Stop before fallback when closing the failed handle also fails
The exception path calls `_close_current()`, which clears the stream
reference and records a close error, and then opens a fresh fallback handle
here. A failed `close()` does not guarantee that the underlying descriptor or
connection was released. With `parallelism=1`, a single read failure followed
by a close failure can therefore leave the original handle active while
`read_file_range` opens a second one, violating the global stream bound and
losing the only reference to the leaked handle. The recorded close error also
makes the batch fail at the end, so this fallback I/O cannot salvage the
result. Please abort the lane before opening another handle when close fails,
preserving the read error as primary when applicable, and update
`test_failed_stream_close_error_is_propagated_after_fallback` to assert that no
fallback is attempted.
--
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]
