XiaoHongbo-Hope commented on code in PR #8406:
URL: https://github.com/apache/paimon/pull/8406#discussion_r3511113131
##########
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:
Fixed
--
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]
