danielcweeks commented on code in PR #15581:
URL: https://github.com/apache/iceberg/pull/15581#discussion_r3075945680
##########
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:
I think we can do this better by creating a wrapper object for the original
`FileRanges` without impacting the existing public api. Rather than creating
new FileRanges and inserting them, just create a inner wrapper class (e.g.
`CoalescedFileRanges`) that contains the ordered list of file ranges that
should be combined. Then we can dispatch the read requests for the total
combined range and populate internal buffers.
--
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]
