PARQUET-2171. Vector IO through self-contained shim classes.

Takes the vector IO shim code from the hadoop-shim library and
reimplements in through the DynMethods class.

Needs
* lower level tests of behaviour
* review of exception handling/unwrapping.

PARQUET-2171. Vector IO

* tests working
* debug logs added DynMethods
* purge shim library refs from pom.xml
* move all static methods to BindingUtils; might be good
  to split into public/private classes, or move the await for stuff to
  VectorIOBridge and keep the rest hidden.

PARQUET-2171. Vector IO

Copy VectoredIO contract tests to parquet code; adapt.

All tests are working except a probe to verify that the
vectored IO API is directly implemented by the RawLocal stream.
That is odd, as the hasCapabilities probe should be in
Hadoop 3.3.5.

PARQUET-2171. Vector IO

Fix name of hasCapabilities method to look for,
so the test works.

+add test javadocs

PARQUET-2171. Vector IO: checkstyles.

PARQUET-2171. tests all working

Testing found a couple of bugs with the initial Vector IO PR;
fixed ParquetFileReader

+ parameterized the tests which had failed, so as to ensure coverage
+ TestVectorIOBridge uses checksumfs, which does implement vector IO
  (which implies all file:// links get the speedup)
+ improved assertions in the failing tests, which were all reporting
  mismatches in data read expected via actual.

As well as these changes, I've also locally run the entire suite with
a core-site.xml set to turn on vector IO; this is how the regressions
were identified.

Author: steveloughran

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

@@ -109,11 +109,19 @@ public abstract class SeekableInputStream extends InputStream {
 
   /**
    * Read a set of file ranges in a vectored manner.
+   * @throws UnsupportedOperationException if not available in this class/runtime.
    */
   public void readVectored(List<ParquetFileRange> ranges,
-                                    IntFunction<ByteBuffer> allocate) throws IOException {
-    throw new IOException(String.format("Vectored io is not supported for %s .",
-      this.getClass().getCanonicalName()));
+    IntFunction<ByteBuffer> allocate) throws IOException {
+
+    throw new UnsupportedOperationException("Vectored IO is not supported for " + this);
+  }
+
+  /**
+   * Is the {@link #readVectored(List, IntFunction)} method available.
+   */
+  public boolean readVectoredAvailable() {
+    return false;
   }
 
 }

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

@@ -27,10 +27,15 @@
 import java.security.PrivilegedAction;
 import java.util.Arrays;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import static org.apache.parquet.Exceptions.throwIfInstance;
 
 public class DynMethods {
 
+  private static final Logger LOG = LoggerFactory.getLogger(DynMethods.class);
+
   /**
    * Convenience wrapper class around {@link java.lang.reflect.Method}.
    *
@@ -243,7 +248,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;
     }
@@ -281,6 +287,7 @@ public Builder impl(Class<?> targetClass, String methodName, Class<?>... argClas
             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;
     }
@@ -311,6 +318,8 @@ 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;
     }
@@ -327,6 +336,9 @@ 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;
     }
@@ -349,7 +361,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;
     }
@@ -388,6 +401,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/pom.xml

@@ -79,11 +79,11 @@
       <version>${hadoop.version}</version>
       <scope>provided</scope>
     </dependency>
-    <dependency>
+<!--    <dependency>
       <groupId>org.apache.hadoop.extensions</groupId>
       <artifactId>hadoop-api-shim</artifactId>
       <scope>provided</scope>
-    </dependency>
+    </dependency>-->
     <dependency>
       <groupId>org.apache.parquet</groupId>
       <artifactId>parquet-jackson</artifactId>

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

@@ -35,7 +35,6 @@
 import java.io.Closeable;
 import java.io.IOException;
 import java.io.InputStream;
-import java.io.InterruptedIOException;
 import java.io.SequenceInputStream;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
@@ -57,14 +56,12 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.stream.Collectors;
-import java.util.stream.Stream;
 import java.util.zip.CRC32;
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.fs.shim.functional.FutureIO;
 import org.apache.parquet.HadoopReadOptions;
 import org.apache.parquet.ParquetReadOptions;
 import org.apache.parquet.Preconditions;
@@ -108,6 +105,7 @@
 import org.apache.parquet.hadoop.metadata.ParquetMetadata;
 import org.apache.parquet.hadoop.util.HadoopInputFile;
 import org.apache.parquet.hadoop.util.counters.BenchmarkCounter;
+import org.apache.parquet.hadoop.util.vectorio.BindingUtils;
 import org.apache.parquet.internal.column.columnindex.ColumnIndex;
 import org.apache.parquet.internal.column.columnindex.OffsetIndex;
 import org.apache.parquet.internal.filter2.columnindex.ColumnIndexFilter;
@@ -995,9 +993,6 @@ private ColumnChunkPageReadStore internalReadRowGroup(int blockIndex) throws IOE
     }
     // actually read all the chunks
     ChunkListBuilder builder = new ChunkListBuilder(block.getRowCount());
-    for (ConsecutivePartList consecutiveChunks : allParts) {
-      consecutiveChunks.readAll(f, builder);
-    }
     readAllPartsVectoredOrNormal(allParts, builder);
     for (Chunk chunk : builder.build()) {
       readChunkPages(chunk, block, rowGroup);
@@ -1075,14 +1070,16 @@ public ColumnChunkPageReadStore readFilteredRowGroup(int blockIndex, RowRanges r
   }
 
   /**
-   *  Read data in all parts via either vectored IO or serial IO.
+   * Read data in all parts via either vectored IO or serial IO.
    * @param allParts all parts to be read.
    * @param builder used to build chunk list to read the pages for the different columns.
    * @throws IOException any IOE.
    */
   private void readAllPartsVectoredOrNormal(List<ConsecutivePartList> allParts, ChunkListBuilder builder)
     throws IOException {
-    boolean isVectoredIO = options.useHadoopVectoredIO() && partsLengthValidForVectoredIO(allParts);
+    boolean isVectoredIO = options.useHadoopVectoredIO()
+      && f.readVectoredAvailable()
+      && partsLengthValidForVectoredIO(allParts);
     if (isVectoredIO) {
       readVectored(allParts, builder);
     } else {
@@ -1114,9 +1111,10 @@ private boolean partsLengthValidForVectoredIO(List<ConsecutivePartList> allParts
    * @param builder used to build chunk list to read the pages for the different columns.
    * @throws IOException any IOE.
    */
-  private void readVectored(List<ConsecutivePartList> allParts, ChunkListBuilder builder)
-    throws IOException {
-    List<ParquetFileRange> ranges = new ArrayList<>();
+  private void readVectored(List<ConsecutivePartList> allParts,
+    ChunkListBuilder builder) throws IOException {
+
+    List<ParquetFileRange> ranges = new ArrayList<>(allParts.size());
     for (ConsecutivePartList consecutiveChunks : allParts) {
       Preconditions.checkArgument(consecutiveChunks.length < Integer.MAX_VALUE,
         "Invalid length %s for vectored read operation. It must be less than max integer value.",
@@ -1125,6 +1123,7 @@ private void readVectored(List<ConsecutivePartList> allParts, ChunkListBuilder b
     }
     LOG.debug("Doing vectored IO for ranges {}", ranges);
     ByteBufferAllocator allocator = options.getAllocator();
+    //blocking or asynchronous vectored read.
     f.readVectored(ranges, allocator::allocate);
     int k = 0;
     for (ConsecutivePartList consecutivePart :  allParts) {
@@ -1949,8 +1948,10 @@ public void readFromVectoredRange(ParquetFileRange currRange,
       ByteBuffer buffer;
       try {
         LOG.debug("Waiting for vectored read to finish for range {} ", currRange);
-        buffer = FutureIO.awaitFuture(currRange.getDataReadFuture(),
+        buffer = BindingUtils.awaitFuture(currRange.getDataReadFuture(),
           HADOOP_VECTORED_READ_TIMEOUT_SECONDS, TimeUnit.SECONDS);
+        // report in a counter the data we just scanned
+        BenchmarkCounter.incrementBytesRead(currRange.getLength());
       } catch (TimeoutException e) {
         String error = String.format("Timeout while fetching result for %s", currRange);
         LOG.error(error, e);

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

@@ -125,7 +125,8 @@ public class ParquetInputFormat<T> extends FileInputFormat<Void, T> {
   public static final String STATS_FILTERING_ENABLED = "parquet.filter.stats.enabled";
 
   /**
-   * Key to enable/disable vectored io while reading parquet files.
+   * Key to enable/disable vectored io while reading parquet files:
+   * {@value}.
    */
   public static final String HADOOP_VECTORED_IO_ENABLED = "parquet.hadoop.vectored.io.enabled";
 

parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/H1SeekableInputStream.java

@@ -20,6 +20,7 @@
 package org.apache.parquet.hadoop.util;
 
 import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.parquet.hadoop.util.vectorio.VectorIOBridge;
 import org.apache.parquet.io.DelegatingSeekableInputStream;
 import org.apache.parquet.io.ParquetFileRange;
 
@@ -28,8 +29,6 @@
 import java.util.List;
 import java.util.function.IntFunction;
 
-import static org.apache.parquet.hadoop.util.ParquetVectoredIOUtil.readVectoredAndPopulate;
-
 /**
  * SeekableInputStream implementation that implements read(ByteBuffer) for
  * Hadoop 1 FSDataInputStream.
@@ -63,8 +62,13 @@ public void readFully(byte[] bytes, int start, int len) throws IOException {
     stream.readFully(bytes, start, len);
   }
 
+  @Override
+  public boolean readVectoredAvailable() {
+    return VectorIOBridge.bridgeAvailable();
+  }
+
   @Override
   public void readVectored(List<ParquetFileRange> ranges, IntFunction<ByteBuffer> allocate) throws IOException {
-    readVectoredAndPopulate(ranges, allocate, stream);
+    VectorIOBridge.readVectoredRanges(stream, ranges, allocate);
   }
 }

parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/H2SeekableInputStream.java

@@ -20,6 +20,7 @@
 package org.apache.parquet.hadoop.util;
 
 import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.parquet.hadoop.util.vectorio.VectorIOBridge;
 import org.apache.parquet.io.DelegatingSeekableInputStream;
 import org.apache.parquet.io.ParquetFileRange;
 
@@ -29,8 +30,6 @@
 import java.util.List;
 import java.util.function.IntFunction;
 
-import static org.apache.parquet.hadoop.util.ParquetVectoredIOUtil.readVectoredAndPopulate;
-
 /**
  * SeekableInputStream implementation for FSDataInputStream that implements
  * ByteBufferReadable in Hadoop 2.
@@ -88,9 +87,14 @@ public int read(ByteBuffer buf) throws IOException {
     }
   }
 
+  @Override
+  public boolean readVectoredAvailable() {
+    return VectorIOBridge.bridgeAvailable();
+  }
+
   @Override
   public void readVectored(List<ParquetFileRange> ranges, IntFunction<ByteBuffer> allocate) throws IOException {
-    readVectoredAndPopulate(ranges, allocate, stream);
+    VectorIOBridge.readVectoredRanges(stream, ranges, allocate);
   }
 
   public static void readFully(Reader reader, ByteBuffer buf) throws IOException {