PARQUET-2171: Support Hadoop vectored IO

parquet-common/src/main/java/org/apache/parquet/io/ParquetFileRange.java

@@ -0,0 +1,71 @@
+/*
+ * 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.parquet.io;
+
+import java.nio.ByteBuffer;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Class to define a file range for a parquet file and to
+ * hold future data for any ongoing read for that range.
+ */
+public class ParquetFileRange {
+
+ /**
+ * Start position in file.
+ */
+ private final long offset;
+
+ /**
+ * Length of data to be read from position.
+ */
+ private final int length;
+
+ /**
+ * A future object to hold future for ongoing read.
+ */
+ private CompletableFuture<ByteBuffer> dataReadFuture;
+
+ public ParquetFileRange(long offset, int length) {
+ this.offset = offset;
+ this.length = length;
+ }
+
+ public long getOffset() {
+ return offset;
+ }
+
+ public int getLength() {
+ return length;
+ }
+
+ public CompletableFuture<ByteBuffer> getDataReadFuture() {
+ return dataReadFuture;
+ }
+
+ public void setDataReadFuture(CompletableFuture<ByteBuffer> dataReadFuture) {
+ this.dataReadFuture = dataReadFuture;
+ }
+
+ @Override
+ public String toString() {
+ return String.format("range[%,d - %,d)", offset, offset + length);
+ }
+}

parquet-common/src/main/java/org/apache/parquet/io/SeekableInputStream.java

@@ -23,6 +23,8 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.nio.ByteBuffer;
+import java.util.List;
+import org.apache.parquet.bytes.ByteBufferAllocator;
 
 /**
  * {@code SeekableInputStream} is an interface with the methods needed by
@@ -104,4 +106,29 @@ public abstract class SeekableInputStream extends InputStream {
  * fill the buffer, {@code buf.remaining()}
  */
  public abstract void readFully(ByteBuffer buf) throws IOException;
+
+ /**
+ * Read a set of disjoint file ranges in a vectored manner.
+ *
+ * @param ranges a list of non-overlapping file ranges to read
+ * @param allocator the allocator to use for allocating ByteBuffers
+ * @throws UnsupportedOperationException if not available in this class/runtime (default)
+ * @throws EOFException if a range is past the known end of the file.
+ * @throws IOException any IO problem initiating the read operations.
+ * @throws IllegalArgumentException if there are overlapping ranges or
+ * a range element is invalid
+ */
+ public void readVectored(List<ParquetFileRange> ranges, final ByteBufferAllocator allocator) throws IOException {
+
+ throw new UnsupportedOperationException("Vectored IO is not supported for " + this);
+ }
+
+ /**
+ * Is the {@link #readVectored(List, ByteBufferAllocator)} method available?
+ * @param allocator the allocator to use for allocating ByteBuffers
+ * @return True if the operation is considered available for this allocator in the hadoop runtime.
+ */
+ public boolean readVectoredAvailable(final ByteBufferAllocator allocator) {
+ return false;
+ }
 }

parquet-common/src/main/java/org/apache/parquet/util/DynMethods.java

@@ -28,9 +28,13 @@
 import java.security.PrivilegedAction;
 import java.util.Arrays;
 import org.apache.parquet.Preconditions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class DynMethods {
 
+ private static final Logger LOG = LoggerFactory.getLogger(DynMethods.class);
+
  /**
  * Convenience wrapper class around {@link java.lang.reflect.Method}.
  * <p>
@@ -240,7 +244,8 @@ public Builder impl(String className, String methodName, Class<?>... argClasses)
  Class<?> targetClass = Class.forName(className, true, loader);
  impl(targetClass, methodName, argClasses);
  } catch (ClassNotFoundException e) {
- // not the right implementation
+ // class not found on supplied classloader.
+ LOG.debug("failed to load class {}", className, e);
  }
  return this;
  }
@@ -277,6 +282,7 @@ public Builder impl(Class<?> targetClass, String methodName, Class<?>... argClas
  this.method = new UnboundMethod(targetClass.getMethod(methodName, argClasses), name);
  } catch (NoSuchMethodException e) {
  // not the right implementation
+ LOG.debug("failed to load method {} from class {}", methodName, targetClass, e);
  }
  return this;
  }
@@ -307,6 +313,7 @@ public Builder ctorImpl(Class<?> targetClass, Class<?>... argClasses) {
  .buildChecked();
  } catch (NoSuchMethodException e) {
  // not the right implementation
+ LOG.debug("failed to load constructor arity {} from class {}", argClasses.length, targetClass, e);
  }
  return this;
  }
@@ -323,6 +330,7 @@ public Builder ctorImpl(String className, Class<?>... argClasses) {
  .buildChecked();
  } catch (NoSuchMethodException e) {
  // not the right implementation
+ LOG.debug("failed to load constructor arity {} from class {}", argClasses.length, className, e);
  }
  return this;
  }
@@ -345,7 +353,8 @@ public Builder hiddenImpl(String className, String methodName, Class<?>... argCl
  Class<?> targetClass = Class.forName(className, true, loader);
  hiddenImpl(targetClass, methodName, argClasses);
  } catch (ClassNotFoundException e) {
- // not the right implementation
+ // class not found on supplied classloader.
+ LOG.debug("failed to load class {}", className, e);
  }
  return this;
  }
@@ -384,6 +393,7 @@ public Builder hiddenImpl(Class<?> targetClass, String methodName, Class<?>... a
  this.method = new UnboundMethod(hidden, name);
  } catch (SecurityException | NoSuchMethodException e) {
  // unusable or not the right implementation
+ LOG.debug("failed to load method {} from class {}", methodName, targetClass, e);
  }
  return this;
  }

parquet-hadoop/README.md

@@ -501,3 +501,11 @@ If `false`, key material is stored in separate new files, created in the same fo
 **Description:** Length of key encryption keys (KEKs), randomly generated by parquet key management tools. Can be 128, 192 or 256 bits. 
 **Default value:** `128`
 
+---
+
+**Property:** `parquet.hadoop.vectored.io.enabled` 
+**Description:** Flag to enable use of the FileSystem Vector IO API on Hadoop releases which support the feature.
+If `true` then an attempt will be made to dynamically load the relevant classes; 
+if not found then the library will use the classic non-vectored reads: it is safe to enable this option on older releases. 
+**Default value:** `false`
+

parquet-hadoop/src/main/java/org/apache/parquet/HadoopReadOptions.java

@@ -43,6 +43,7 @@ private HadoopReadOptions(
  boolean usePageChecksumVerification,
  boolean useBloomFilter,
  boolean useOffHeapDecryptBuffer,
+ boolean useHadoopVectoredIo,
  FilterCompat.Filter recordFilter,
  MetadataFilter metadataFilter,
  CompressionCodecFactory codecFactory,
@@ -61,6 +62,7 @@ private HadoopReadOptions(
  usePageChecksumVerification,
  useBloomFilter,
  useOffHeapDecryptBuffer,
+ useHadoopVectoredIo,
  recordFilter,
  metadataFilter,
  codecFactory,
@@ -123,6 +125,7 @@ public ParquetReadOptions build() {
  usePageChecksumVerification,
  useBloomFilter,
  useOffHeapDecryptBuffer,
+ useHadoopVectoredIo,
  recordFilter,
  metadataFilter,
  codecFactory,

parquet-hadoop/src/main/java/org/apache/parquet/ParquetReadOptions.java

@@ -55,6 +55,7 @@ public class ParquetReadOptions {
  private static final boolean STATS_FILTERING_ENABLED_DEFAULT = true;
  private static final boolean DICTIONARY_FILTERING_ENABLED_DEFAULT = true;
  private static final boolean COLUMN_INDEX_FILTERING_ENABLED_DEFAULT = true;
+ private static final boolean HADOOP_VECTORED_IO_ENABLED_DEFAULT = false;
  private static final int ALLOCATION_SIZE_DEFAULT = 8388608; // 8MB
  private static final boolean PAGE_VERIFY_CHECKSUM_ENABLED_DEFAULT = false;
  private static final boolean BLOOM_FILTER_ENABLED_DEFAULT = true;
@@ -68,6 +69,7 @@ public class ParquetReadOptions {
  private final boolean usePageChecksumVerification;
  private final boolean useBloomFilter;
  private final boolean useOffHeapDecryptBuffer;
+ private final boolean useHadoopVectoredIo;
  private final FilterCompat.Filter recordFilter;
  private final ParquetMetadataConverter.MetadataFilter metadataFilter;
  private final CompressionCodecFactory codecFactory;
@@ -87,6 +89,7 @@ public class ParquetReadOptions {
  boolean usePageChecksumVerification,
  boolean useBloomFilter,
  boolean useOffHeapDecryptBuffer,
+ boolean useHadoopVectoredIo,
  FilterCompat.Filter recordFilter,
  ParquetMetadataConverter.MetadataFilter metadataFilter,
  CompressionCodecFactory codecFactory,
@@ -104,6 +107,7 @@ public class ParquetReadOptions {
  usePageChecksumVerification,
  useBloomFilter,
  useOffHeapDecryptBuffer,
+ useHadoopVectoredIo,
  recordFilter,
  metadataFilter,
  codecFactory,
@@ -124,6 +128,7 @@ public class ParquetReadOptions {
  boolean usePageChecksumVerification,
  boolean useBloomFilter,
  boolean useOffHeapDecryptBuffer,
+ boolean useHadoopVectoredIo,
  FilterCompat.Filter recordFilter,
  ParquetMetadataConverter.MetadataFilter metadataFilter,
  CompressionCodecFactory codecFactory,
@@ -141,6 +146,7 @@ public class ParquetReadOptions {
  this.usePageChecksumVerification = usePageChecksumVerification;
  this.useBloomFilter = useBloomFilter;
  this.useOffHeapDecryptBuffer = useOffHeapDecryptBuffer;
+ this.useHadoopVectoredIo = useHadoopVectoredIo;
  this.recordFilter = recordFilter;
  this.metadataFilter = metadataFilter;
  this.codecFactory = codecFactory;
@@ -184,6 +190,10 @@ public boolean usePageChecksumVerification() {
  return usePageChecksumVerification;
  }
 
+ public boolean useHadoopVectoredIo() {
+ return useHadoopVectoredIo;
+ }
+
  public FilterCompat.Filter getRecordFilter() {
  return recordFilter;
  }
@@ -242,6 +252,7 @@ public static class Builder {
  protected boolean useStatsFilter = STATS_FILTERING_ENABLED_DEFAULT;
  protected boolean useDictionaryFilter = DICTIONARY_FILTERING_ENABLED_DEFAULT;
  protected boolean useRecordFilter = RECORD_FILTERING_ENABLED_DEFAULT;
+ protected boolean useHadoopVectoredIo = HADOOP_VECTORED_IO_ENABLED_DEFAULT;
  protected boolean useColumnIndexFilter = COLUMN_INDEX_FILTERING_ENABLED_DEFAULT;
  protected boolean usePageChecksumVerification = PAGE_VERIFY_CHECKSUM_ENABLED_DEFAULT;
  protected boolean useBloomFilter = BLOOM_FILTER_ENABLED_DEFAULT;
@@ -320,6 +331,11 @@ public Builder useRecordFilter() {
  return this;
  }
 
+ public Builder withUseHadoopVectoredIo(boolean useHadoopVectoredIo) {
+ this.useHadoopVectoredIo = useHadoopVectoredIo;
+ return this;
+ }
+
  public Builder useColumnIndexFilter(boolean useColumnIndexFilter) {
  this.useColumnIndexFilter = useColumnIndexFilter;
  return this;
@@ -418,6 +434,7 @@ public Builder copy(ParquetReadOptions options) {
  useDictionaryFilter(options.useDictionaryFilter);
  useRecordFilter(options.useRecordFilter);
  withRecordFilter(options.recordFilter);
+ withUseHadoopVectoredIo(options.useHadoopVectoredIo);
  withMetadataFilter(options.metadataFilter);
  withCodecFactory(options.codecFactory);
  withAllocator(options.allocator);
@@ -449,6 +466,7 @@ public ParquetReadOptions build() {
  usePageChecksumVerification,
  useBloomFilter,
  useOffHeapDecryptBuffer,
+ useHadoopVectoredIo,
  recordFilter,
  metadataFilter,
  codecFactory,