XiaoHongbo-Hope commented on code in PR #8406:
URL: https://github.com/apache/paimon/pull/8406#discussion_r3511128169
##########
paimon-python/pypaimon/common/file_io.py:
##########
@@ -48,6 +48,36 @@ def pread(stream, length: int, offset: int) -> bytes:
return os.pread(stream.fileno(), length, offset)
+def read_file_range(file_io, path, offset, length):
+ """Read a byte range. Thread-safe. ``length < 0`` = read to EOF (pread
can't
+ express that, so seek + read)."""
+ stream = file_io.new_input_stream(path)
+ try:
+ 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)
+ finally:
+ stream.close()
+
+
+def read_blobs_concurrent(file_io, blobs, parallelism):
+ """Read a list of Blobs concurrently via ThreadPool + pread."""
+ from concurrent.futures import ThreadPoolExecutor
+
+ def _read_one(blob):
+ if blob is None:
+ return None
+ return blob.to_data()
+
+ non_null = [b for b in blobs if b is not None]
+ if not non_null:
+ return [None] * len(blobs)
+ workers = min(parallelism, len(non_null))
+ with ThreadPoolExecutor(workers) as pool:
+ return list(pool.map(_read_one, blobs))
Review Comment:
> Do we need to do some IO merging? For example, if the current number of
blob reads is greater than the concurrent number, we can look at the same
file's blob and merge them together for reading if the intervals are not far
apart.
Great idea, updated.
--
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]
