core: Adding read vector to range readable interface and adding mappe…

api/src/main/java/org/apache/iceberg/io/FileRange.java

@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+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;
+  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");
+    Preconditions.checkArgument(
+        length() >= 0, "Invalid length: %s in range (must be >= 0)", length);
+    Preconditions.checkArgument(
+        offset() >= 0, "Invalid offset: %s in range (must be >= 0)", offset);
+
+    this.byteBuffer = byteBuffer;
+    this.offset = offset;
+    this.length = length;
+  }
+
+  public CompletableFuture<ByteBuffer> byteBuffer() {
+    return byteBuffer;
+  }
+
+  public long offset() {
+    return offset;
+  }
+
+  public int length() {
+    return length;
+  }
+}

api/src/main/java/org/apache/iceberg/io/RangeReadable.java

@@ -20,6 +20,12 @@
 
 import java.io.Closeable;
 import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Comparator;
+import java.util.List;
+import java.util.function.IntFunction;
+import java.util.stream.Collectors;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 
 /**
  * {@code RangeReadable} is an interface that allows for implementations of {@link InputFile}
@@ -77,4 +83,61 @@ default void readFully(long position, byte[] buffer) throws IOException {
   default int readTail(byte[] buffer) throws IOException {
     return readTail(buffer, 0, buffer.length);
   }
+
+  /**
+   * Read fully a list of file ranges asynchronously from this file. As a result of the call, each
+   * range will have FileRange.setData(CompletableFuture) called with a future that when complete
+   * will have a ByteBuffer with the data from the file's range.
+   *
+   * <p>The position returned by getPos() after readVectored() is undefined.
+   *
+   * <p>If a file is changed while the readVectored() operation is in progress, the output is
+   * undefined. Some ranges may have old data, some may have new and some may have both.
+   *
+   * <p>While a readVectored() operation is in progress, normal read api calls may block.
+   *
+   * @param ranges the byte ranges to read
+   * @param allocate the function to allocate ByteBuffer
+   * @throws IOException any IOE.
+   * @throws IllegalArgumentException if any of ranges are invalid, or they overlap.
+   */
+  default void readVectored(List<FileRange> ranges, IntFunction<ByteBuffer> allocate)
+      throws IOException {
+    List<FileRange> validatedRanges = sortRanges(ranges);
+    for (FileRange range : validatedRanges) {
+      ByteBuffer buffer = allocate.apply(range.length());
+      readFully(range.offset(), buffer.array());
+      range.byteBuffer().complete(buffer);
+    }
+  }
+
+  static List<FileRange> sortRanges(final List<FileRange> input) {
+    Preconditions.checkNotNull(input, "Input list can't be null");
+
+    final List<FileRange> sortedRanges;
+
+    // 2 because the input size can be 0/1, and then we want to skip sorting.
+    if (input.size() < 2) {
+      sortedRanges = input;
+    } else {
+      sortedRanges =
+          input.stream()
+              .sorted(Comparator.comparingLong(FileRange::offset))
+              .collect(Collectors.toList());
+      FileRange prev = null;
+      for (final FileRange current : sortedRanges) {
+        if (prev != null) {
+          Preconditions.checkArgument(
+              current.offset() >= prev.offset() + prev.length(),
+              "Overlapping ranges %s and %s",
+              prev,
+              current);
+        }
+
+        prev = current;
+      }
+    }
+
+    return sortedRanges;
+  }
 }

parquet/src/main/java/org/apache/iceberg/parquet/Parquet.java

@@ -1340,6 +1340,7 @@ public <D> CloseableIterable<D> build() {
           optionsBuilder.withDecryption(fileDecryptionProperties);
         }
 
+        optionsBuilder.withUseHadoopVectoredIo(true);
         ParquetReadOptions options = optionsBuilder.build();
 
         NameMapping mapping;

parquet/src/main/java/org/apache/iceberg/parquet/ParquetIO.java

@@ -18,9 +18,15 @@
  */
 package org.apache.iceberg.parquet;
 
+import java.io.EOFException;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.IntFunction;
+import java.util.stream.Collectors;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FSDataInputStream;
 import org.apache.hadoop.fs.FSDataOutputStream;
@@ -29,11 +35,16 @@
 import org.apache.iceberg.hadoop.HadoopOutputFile;
 import org.apache.iceberg.io.DelegatingInputStream;
 import org.apache.iceberg.io.DelegatingOutputStream;
+import org.apache.iceberg.io.FileRange;
+import org.apache.iceberg.io.RangeReadable;
+import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
+import org.apache.parquet.bytes.ByteBufferAllocator;
 import org.apache.parquet.hadoop.util.HadoopStreams;
 import org.apache.parquet.io.DelegatingPositionOutputStream;
 import org.apache.parquet.io.DelegatingSeekableInputStream;
 import org.apache.parquet.io.InputFile;
 import org.apache.parquet.io.OutputFile;
+import org.apache.parquet.io.ParquetFileRange;
 import org.apache.parquet.io.PositionOutputStream;
 import org.apache.parquet.io.SeekableInputStream;
 
@@ -91,6 +102,11 @@ static SeekableInputStream stream(org.apache.iceberg.io.SeekableInputStream stre
         return HadoopStreams.wrap((FSDataInputStream) wrapped);
       }
     }
+
+    if (stream instanceof RangeReadable) {
+      return new ParquetRangeReadableInputStreamAdapter(stream);
+    }
+
     return new ParquetInputStreamAdapter(stream);
   }
 
@@ -123,6 +139,73 @@ public void seek(long newPos) throws IOException {
     }
   }
 
+  @VisibleForTesting
+  static class ParquetRangeReadableInputStreamAdapter<
+          T extends org.apache.iceberg.io.SeekableInputStream & RangeReadable>
+      extends DelegatingSeekableInputStream implements RangeReadable {
+    private final T delegate;
+
+    ParquetRangeReadableInputStreamAdapter(T delegate) {
+      super(delegate);
+      this.delegate = delegate;
+    }
+
+    @Override
+    public long getPos() throws IOException {
+      return delegate.getPos();
+    }
+
+    @Override
+    public void seek(long newPos) throws IOException {
+      delegate.seek(newPos);
+    }
+
+    @Override
+    public void readFully(long position, byte[] buffer, int offset, int length) throws IOException {
+      delegate.readFully(position, buffer, offset, length);
+    }
+
+    @Override
+    public int readTail(byte[] buffer, int offset, int length) throws IOException {
+      return delegate.readTail(buffer, offset, length);
+    }
+
+    @Override
+    public boolean readVectoredAvailable(ByteBufferAllocator allocate) {
+      return true;
+    }
+
+    @Override
+    public void readVectored(List<ParquetFileRange> ranges, ByteBufferAllocator allocate)
+        throws IOException {
+      IntFunction<ByteBuffer> delegateAllocate = (allocate::allocate);
+      List<FileRange> delegateRange = convertRanges(ranges);
+      delegate.readVectored(delegateRange, delegateAllocate);
+    }
+
+    private static List<FileRange> convertRanges(List<ParquetFileRange> ranges) {
+      return ranges.stream()
+          .map(
+              parquetFileRange -> {
+                CompletableFuture<ByteBuffer> future = new CompletableFuture<>();
+                parquetFileRange.setDataReadFuture(future);
+                try {
+                  return new FileRange(
+                      parquetFileRange.getDataReadFuture(),
+                      parquetFileRange.getOffset(),
+                      parquetFileRange.getLength());
+                } catch (EOFException e) {
+                  throw new RuntimeIOException(
+                      e,
+                      "Failed to create range file for offset: %s and length: %s",
+                      parquetFileRange.getOffset(),
+                      parquetFileRange.getLength());
+                }
+              })
+          .collect(Collectors.toList());
+    }
+  }
+
   private static class ParquetOutputStreamAdapter extends DelegatingPositionOutputStream {
     private final org.apache.iceberg.io.PositionOutputStream delegate;