mukund-thakur commented on code in PR #15581:
URL: https://github.com/apache/iceberg/pull/15581#discussion_r3082524668
##########
api/src/main/java/org/apache/iceberg/io/FileRange.java:
##########
@@ -18,19 +18,25 @@
*/
package org.apache.iceberg.io;
-import java.io.EOFException;
import java.nio.ByteBuffer;
import java.util.concurrent.CompletableFuture;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
-public class FileRange {
- private final CompletableFuture<ByteBuffer> byteBuffer;
+public class FileRange implements Comparable<FileRange> {
+ private CompletableFuture<ByteBuffer> byteBuffer;
private final long offset;
private final int length;
- public FileRange(CompletableFuture<ByteBuffer> byteBuffer, long offset, int
length)
- throws EOFException {
- Preconditions.checkNotNull(byteBuffer, "byteBuffer can't be null");
+ public FileRange(long offset, int length) {
+ Preconditions.checkArgument(
+ length() >= 0, "Invalid length: %s in range (must be >= 0)", length);
Review Comment:
we should use the the input length and offset in the preconditions. someone
found this a bug https://github.com/apache/iceberg/pull/15926
##########
aws/src/main/java/org/apache/iceberg/aws/s3/S3InputStream.java:
##########
@@ -190,6 +221,116 @@ private InputStream readRange(String range) {
return s3.getObject(requestBuilder.build(),
ResponseTransformer.toInputStream());
}
+ /**
+ * Vectored read implementation that coalesces nearby ranges to reduce S3
requests. Step Through
+ * Process: 1. Coalesce ranges - merge nearby ranges within skipSize
tolerance and create linked
+ * lists to populate 2. Submit S3 requests concurrently for each coalesced
range 3. For each S3
+ * response stream, skip gaps and read into individual range buffers Given
ranges [0-100],
+ * [200-300], [10000-10100] with skipSize=1024: - First two ranges are close
(gap=100 < 1024) so
+ * they coalesce into [0-300] - Third range is far (gap=9700 > 1024) so it
stays separate as
+ * [10000-10100] - Results in 2 S3 requests for [0-300] and [10000-10100] -
Reads the stream into
+ * the buffer, stopping when it reaches the length + offset - Then Skips to
the next linked
+ * buffers start - For coalesced [0-300]: read [0-100], skip 100 bytes, read
[200-300]
+ *
+ * @param ranges the ranges to read and their futures
+ * @param allocate the function to allocate ByteBuffer
+ */
+ @Override
+ public void readVectored(List<FileRange> ranges, IntFunction<ByteBuffer>
allocate) {
+ Map<FileRange, List<FileRange>> linkedRanges = Maps.newHashMap();
+ List<FileRange> coalesced = coalesce(ranges, linkedRanges);
+
+ Tasks.foreach(coalesced)
+ .retry(3)
+ .executeWith(executorService())
+ .onFailure((range, thrown) -> LOG.error("Failed to stream range: {}",
range, thrown))
+ .throwFailureWhenFinished()
+ .run(
+ coalescedRange -> {
+ String range =
+ String.format(
+ "bytes=%s-%s",
+ coalescedRange.offset(),
+ coalescedRange.offset() + coalescedRange.length() - 1);
+ try (InputStream stream = readRange(range)) {
+ long currentPos = coalescedRange.offset();
+
+ for (FileRange linkedRange : linkedRanges.get(coalescedRange))
{
+ // Skip to the start of this linked range
+ long skipBytes = linkedRange.offset() - currentPos;
+ if (skipBytes > 0) {
+ ByteStreams.skipFully(stream, skipBytes);
+ }
+
+ // Read into the linked range buffer
+ ByteBuffer buffer = allocate.apply(linkedRange.length());
+ ByteStreams.readFully(stream, buffer.array(), 0,
linkedRange.length());
+ linkedRange.byteBuffer().complete(buffer);
+
+ currentPos = linkedRange.offset() + linkedRange.length();
+ }
+ } catch (IOException e) {
+ LOG.error("Failed to stream range: {}", range, e);
+ throw new UncheckedIOException(e);
+ }
+ });
+ }
+
+ /**
+ * Merges nearby ranges to reduce S3 requests. Ranges are merged if the gap
between them is
+ * smaller than skipSize.
+ *
+ * <p>With skipSize=1024, ranges [0-100], [200-300], [5000-5100] become: -
[0-300] containing
+ * [0-100] and [200-300] (gap=100 < 1024) - [5000-5100] stays separate
(gap=4700 > 1024)
+ *
+ * @param ranges the ranges to coalesce
+ * @param linkedRanges map to store linked ranges for each coalesced range
+ * @return a new list of coalesced ranges
+ */
+ private List<FileRange> coalesce(
Review Comment:
yes agree on this as we did something similar in Hadoop S3A implementation.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
