HADOOP-19131. Add ByteBufferPositionedReadable

...because parquet could use it reading footers.

Change-Id: Ic09116005b76b1f2221a961c923df412718508cc

Author: steveloughran

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FSDataInputStream.java

@@ -262,6 +262,14 @@ public int read(long position, ByteBuffer buf) throws IOException {
         "by " + in.getClass().getCanonicalName());
   }
 
+  /**
+   * Delegate to the underlying stream.
+   * @param position position within file
+   * @param buf the ByteBuffer to receive the results of the read operation.
+   * @throws IOException on a failure from the nested stream.
+   * @throws UnsupportedOperationException if the inner stream does not
+   * support this operation.
+   */
   @Override
   public void readFully(long position, ByteBuffer buf) throws IOException {
     if (in instanceof ByteBufferPositionedReadable) {

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/wrappedio/WrappedOperations.java

@@ -18,20 +18,24 @@
 
 package org.apache.hadoop.io.wrappedio;
 
+import java.io.EOFException;
 import java.io.IOException;
 import java.io.Serializable;
 import java.io.UncheckedIOException;
+import java.nio.ByteBuffer;
 import java.util.HashMap;
 import java.util.Map;
 import javax.annotation.Nullable;
 
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.fs.ByteBufferPositionedReadable;
 import org.apache.hadoop.fs.FSDataInputStream;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.FutureDataInputStreamBuilder;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.StreamCapabilities;
 import org.apache.hadoop.fs.statistics.IOStatistics;
 import org.apache.hadoop.fs.statistics.IOStatisticsContext;
 import org.apache.hadoop.fs.statistics.IOStatisticsLogging;
@@ -60,6 +64,28 @@ public final class WrappedOperations {
   private WrappedOperations() {
   }
 
+  public static boolean hasPathCapability(FileSystem fs, Path path, String capability) {
+    try {
+      return fs.hasPathCapability(path, capability);
+    } catch (IOException e) {
+      return false;
+    }
+  }
+
+  /**
+   * Does an object implement {@link StreamCapabilities} and, if so,
+   * what is the result of the probe for the capability.
+   * @param o object to probe
+   * @param capability capability string
+   * @return true iff the capability is declared available.
+   */
+  public static boolean hasStreamCapability(Object o, String capability) {
+    if (o instanceof StreamCapabilities) {
+      return false;
+    }
+    return ((StreamCapabilities) o).hasCapability(capability);
+  }
+
   /**
    * OpenFile assistant, easy reflection-based access to
    * {@link FileSystem#openFile(Path)}.
@@ -251,31 +277,76 @@ public static Map<String, Long> iostatisticsSnapshotToMap(@Nullable Object sourc
    * This is either a thread-local value or a global empty context.
    * @return instance of {@link IOStatisticsContext}
    */
-  public Object iostatisticsContextGetCurrent() {
+  public static Object iostatisticsContextGetCurrent() {
     return getCurrentIOStatisticsContext();
   }
 
   /**
    * Static probe to check if the thread-level IO statistics enabled.
    * @return true if the thread-level IO statistics are enabled.
    */
-  public boolean iostatisticsContextEnabled() {
+  public static boolean iostatisticsContextEnabled() {
     return IOStatisticsContext.enabled();
   }
 
   /**
    * Reset the context's IOStatistics.
    */
-  public void iostatisticsContextReset() {
+  public static void iostatisticsContextReset() {
     getCurrentIOStatisticsContext().reset();
   }
 
   /**
    * Take a snapshot of the context IOStatistics.
    * @return an instance of {@link IOStatisticsSnapshot}.
    */
-  public Serializable iostatisticsContextSnapshot() {
+  public static Serializable iostatisticsContextSnapshot() {
     return getCurrentIOStatisticsContext().snapshot();
   }
 
+  /**
+   * Delegate to {@link ByteBufferPositionedReadable#read(long, ByteBuffer)}.
+   * @param in input stream
+   * @param position position within file
+   * @param buf the ByteBuffer to receive the results of the read operation.
+   * @throws IOException if there is some error performing the read
+   * @throws EOFException the end of the data was reached before
+   * the read operation completed
+   * @throws UnsupportedOperationException if the input doesn't implement
+   * the interface or, if when invoked, it is raised.
+   * Note: that is the default behaviour of {@link FSDataInputStream#readFully(long, ByteBuffer)}.
+   */
+  public static void byteBufferPositionedReadableReadFully(
+      Object in,
+      long position,
+      ByteBuffer buf)
+      throws IOException {
+    if (!(in instanceof ByteBufferPositionedReadable)) {
+      throw new UnsupportedOperationException("Not a ByteBufferPositionedReadable: " + in);
+    }
+
+    ((ByteBufferPositionedReadable) in).readFully(position, buf);
+  }
+
+  /**
+   * Probe to see if the input stream is an instance of ByteBufferPositionedReadable.
+   * If the stream is an FSDataInputStream, the wrapped stream is checked.
+   * @param in input stream
+   * @return true if the stream implements the interface (including a wrapped stream)
+   * and that it declares the stream capability.
+   */
+  public static boolean byteBufferPositionedReadableReadFullyAvailable(
+      Object in) {
+    if (!(in instanceof ByteBufferPositionedReadable)) {
+      return false;
+    }
+    if (in instanceof FSDataInputStream) {
+      // ask the wrapped stream.
+      return byteBufferPositionedReadableReadFullyAvailable(
+          ((FSDataInputStream) in).getWrappedStream());
+    }
+    // now rely on the input stream implementing path capabilities, which
+    // all the Hadoop FS implementations do.
+    return hasStreamCapability(in, StreamCapabilities.PREADBYTEBUFFER);
+  }
 }