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

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

@@ -0,0 +1,46 @@
+/*
+ * 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.nio.ByteBuffer;
+import java.util.concurrent.CompletableFuture;
+
+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) {
+    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,10 @@
 
 import java.io.Closeable;
 import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.function.IntFunction;
+import org.apache.iceberg.util.VectoredReadUtils;
 
 /**
  * {@code RangeReadable} is an interface that allows for implementations of {@link InputFile}
@@ -77,4 +81,41 @@ default void readFully(long position, byte[] buffer) throws IOException {
   default int readTail(byte[] buffer) throws IOException {
     return readTail(buffer, 0, buffer.length);
   }
+
+  /**
+   * Is the {@link #readVectored(List, IntFunction)} method available?
+   *
+   * @param allocate the allocator to use for allocating ByteBuffers
+   * @return True if the operation is considered available for this allocator.
+   */
+  default boolean readVectoredAvailable(IntFunction<ByteBuffer> allocate) {
+    return true;
+  }
+
+  /**
+   * 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 the any of ranges are invalid, or they overlap.
+   */
+  default void readVectored(List<FileRange> ranges, IntFunction<ByteBuffer> allocate)
+      throws IOException {
+    List<FileRange> validatedRanges = VectoredReadUtils.validateAndSortRanges(ranges);
+    for (FileRange range : validatedRanges) {
+      ByteBuffer buffer = allocate.apply(range.length());
+      readFully(range.offset(), buffer.array());
+      range.byteBuffer().complete(buffer);
+    }
+  }
 }

api/src/main/java/org/apache/iceberg/util/VectoredReadUtils.java

@@ -0,0 +1,120 @@
+/*
+ * 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.util;
+
+import java.io.EOFException;
+import java.util.Comparator;
+import java.util.List;
+import org.apache.iceberg.io.FileRange;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Utils class for vectoredReads, to help with things like range validation. Most of the code in
+ * this class is written by @mukundthakur, and taken from
+ * /hadoop-common/src/main/java/org/apache/hadoop/fs/VectoredReadUtils.java (thank you!).
+ */
+public final class VectoredReadUtils {
+  private VectoredReadUtils() {}
+
+  private static final Logger LOG = LoggerFactory.getLogger(VectoredReadUtils.class);
+
+  /**
+   * Validate a list of ranges (including overlapping checks) and return the sorted list.
+   *
+   * <p>Two ranges overlap when the start offset of second is less than the end offset of first. End
+   * offset is calculated as start offset + length.
+   *
+   * @param input input list
+   * @return a new sorted list.
+   * @throws IllegalArgumentException if there are overlapping ranges or a range element is invalid
+   *     (other than with negative offset)
+   * @throws EOFException if the last range extends beyond the end of the file supplied or a range
+   *     offset is negative
+   */
+  public static List<FileRange> validateAndSortRanges(final List<FileRange> input)
+      throws EOFException {
+
+    Preconditions.checkNotNull(input, "Null input list");
+
+    if (input.isEmpty()) {
+      // this may seem a pathological case, but it was valid
+      // before and somehow Spark can call it through parquet.
+      LOG.debug("Empty input list");
+      return input;
+    }
+
+    final List<FileRange> sortedRanges;
+
+    if (input.size() == 1) {
+      validateRangeRequest(input.get(0));
+      sortedRanges = input;
+    } else {
+      sortedRanges = sortRangeList(input);
+      FileRange prev = null;
+      for (final FileRange current : sortedRanges) {
+        validateRangeRequest(current);
+        if (prev != null) {
+          Preconditions.checkArgument(
+              current.offset() >= prev.offset() + prev.length(),
+              "Overlapping ranges %s and %s",
+              prev,
+              current);
+        }
+        prev = current;
+      }
+    }
+
+    return sortedRanges;
+  }
+
+  /**
+   * Validate a single range.
+   *
+   * @param range range to validate.
+   * @return the range.
+   * @throws IllegalArgumentException the range length is negative or other invalid condition is met
+   *     other than the those which raise EOFException or NullPointerException.
+   * @throws EOFException the range offset is negative
+   * @throws NullPointerException if the range is null.
+   */
+  public static FileRange validateRangeRequest(FileRange range) throws EOFException {
+    Preconditions.checkNotNull(range, "range is null");
+
+    Preconditions.checkArgument(range.length() >= 0, "length is negative in %s", range);
+    if (range.offset() < 0) {
+      throw new EOFException("position is negative in range " + range);
+    }
+    return range;
+  }
+
+  /**
+   * Sort the input ranges by offset; no validation is done.
+   *
+   * @param input input ranges.
+   * @return a new list of the ranges, sorted by offset.
+   */
+  public static List<FileRange> sortRangeList(List<FileRange> input) {
+    final List<FileRange> l = Lists.newArrayList(input);
+    l.sort(Comparator.comparingLong(FileRange::offset));
+    return l;
+  }
+}

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

@@ -21,6 +21,11 @@
 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 +34,15 @@
 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.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 +100,9 @@ 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 +135,64 @@ public void seek(long newPos) throws IOException {
     }
   }
 
+  private static class ParquetRangeReadableInputStreamAdapter<
+          T extends org.apache.iceberg.io.SeekableInputStream & RangeReadable>
+      extends DelegatingSeekableInputStream implements RangeReadable {
+    private final T delegate;
+
+    private 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> result = new CompletableFuture<>();
+                parquetFileRange.setDataReadFuture(result);
+                return new FileRange(
+                    parquetFileRange.getDataReadFuture(),
+                    parquetFileRange.getOffset(),
+                    parquetFileRange.getLength());
+              })
+          .collect(Collectors.toList());
+    }
+  }
+
   private static class ParquetOutputStreamAdapter extends DelegatingPositionOutputStream {
     private final org.apache.iceberg.io.PositionOutputStream delegate;