HADOOP-19131. Assist reflection IO with WrappedOperations class (#6686)

1. The class WrappedIO has been extended with more filesystem operations

- openFile()
- PathCapabilities
- StreamCapabilities
- ByteBufferPositionedReadable

All these static methods raise UncheckedIOExceptions rather than
checked ones.

2. The adjacent class org.apache.hadoop.io.wrappedio.WrappedStatistics
provides similar access to IOStatistics/IOStatisticsContext classes
and operations.

Allows callers to:
* Get a serializable IOStatisticsSnapshot from an IOStatisticsSource or
  IOStatistics instance
* Save an IOStatisticsSnapshot to file
* Convert an IOStatisticsSnapshot to JSON
* Given an object which may be an IOStatisticsSource, return an object
  whose toString() value is a dynamically generated, human readable summary.
  This is for logging.
* Separate getters to the different sections of IOStatistics.
* Mean values are returned as a Map.Pair<Long, Long> of (samples, sum)
  from which means may be calculated.

There are examples of the dynamic bindings to these classes in:

org.apache.hadoop.io.wrappedio.impl.DynamicWrappedIO
org.apache.hadoop.io.wrappedio.impl.DynamicWrappedStatistics

These use DynMethods and other classes in the package
org.apache.hadoop.util.dynamic which are based on the
Apache Parquet equivalents.
This makes re-implementing these in that library and others
which their own fork of the classes (example: Apache Iceberg)

3. The openFile() option "fs.option.openfile.read.policy" has
added specific file format policies for the core filetypes

* avro
* columnar
* csv
* hbase
* json
* orc
* parquet

S3A chooses the appropriate sequential/random policy as a 

A policy `parquet, columnar, vector, random, adaptive` will use the parquet policy for
any filesystem aware of it, falling back to the first entry in the list which
the specific version of the filesystem recognizes

4. New Path capability fs.capability.virtual.block.locations

Indicates that locations are generated client side
and don't refer to real hosts.

Contributed by Steve Loughran

Author: steveloughran

hadoop-common-project/hadoop-common/dev-support/findbugsExcludeFile.xml

@@ -454,4 +454,10 @@
     <Class name="org.apache.hadoop.ipc.internal.ShadedProtobufHelper" />
     <Bug pattern="AT_OPERATION_SEQUENCE_ON_CONCURRENT_ABSTRACTION" />
   </Match>
+
+  <!-- class cast after an assignableFrom check. -->
+  <Match>
+    <Class name="org.apache.hadoop.util.dynamic.DynMethods" />
+    <Bug pattern="BC_UNCONFIRMED_CAST" />
+  </Match>
 </FindBugsFilter>

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

@@ -187,4 +187,20 @@ private CommonPathCapabilities() {
    */
   public static final String BULK_DELETE = "fs.capability.bulk.delete";
 
+  /**
+   * Capability string to probe for block locations returned in {@code LocatedFileStatus}
+   * instances from calls such as {@code getBlockLocations()} and {@code listStatus()}l
+   * to be 'virtual' rather than actual values resolved against a Distributed Filesystem including
+   * HDFS: {@value}.
+   * <p>
+   * Key implications from this path capability being true:
+   * <ol>
+   *   <li>Work can be scheduled anywhere</li>
+   *   <li>Creation of the location list is a low cost-client side operation</li>
+   * </ol>
+   * Implication #2 means there is no performance penalty from use of FileSystem operations which
+   * return lists or iterators of {@code LocatedFileStatus}.
+   */
+  public static final String VIRTUAL_BLOCK_LOCATIONS = "fs.capability.virtual.block.locations";
+
 }

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/fs/Options.java

@@ -573,6 +573,12 @@ private OpenFileOptions() {
     public static final String FS_OPTION_OPENFILE_BUFFER_SIZE =
         FS_OPTION_OPENFILE + "buffer.size";
 
+    /**
+     * OpenFile footer cache flag: {@value}.
+     */
+    public static final String FS_OPTION_OPENFILE_FOOTER_CACHE =
+        FS_OPTION_OPENFILE + "footer.cache";
+
     /**
      * OpenFile option for read policies: {@value}.
      */
@@ -586,6 +592,7 @@ private OpenFileOptions() {
     public static final Set<String> FS_OPTION_OPENFILE_STANDARD_OPTIONS =
         Collections.unmodifiableSet(Stream.of(
                 FS_OPTION_OPENFILE_BUFFER_SIZE,
+                FS_OPTION_OPENFILE_FOOTER_CACHE,
                 FS_OPTION_OPENFILE_READ_POLICY,
                 FS_OPTION_OPENFILE_LENGTH,
                 FS_OPTION_OPENFILE_SPLIT_START,
@@ -599,11 +606,61 @@ private OpenFileOptions() {
         "adaptive";
 
     /**
-     * Read policy {@value} -whateve the implementation does by default.
+     * We are an avro file: {@value}.
+     */
+    public static final String FS_OPTION_OPENFILE_READ_POLICY_AVRO = "avro";
+
+    /**
+     * This is a columnar file format.
+     * Do whatever is needed to optimize for it: {@value}.
+     */
+    public static final String FS_OPTION_OPENFILE_READ_POLICY_COLUMNAR =
+        "columnar";
+
+    /**
+     * This is a CSV file of plain or UTF-8 text
+     * to be read sequentially.
+     * Do whatever is needed to optimize for it: {@value}.
+     */
+    public static final String FS_OPTION_OPENFILE_READ_POLICY_CSV =
+        "csv";
+
+    /**
+     * Read policy {@value} -whatever the implementation does by default.
      */
     public static final String FS_OPTION_OPENFILE_READ_POLICY_DEFAULT =
         "default";
 
+    /**
+     * This is a table file for Apache HBase.
+     * Do whatever is needed to optimize for it: {@value}.
+     */
+    public static final String FS_OPTION_OPENFILE_READ_POLICY_HBASE =
+        "hbase";
+
+    /**
+     * This is a JSON file of UTF-8 text, including a
+     * JSON line file where each line is a JSON entity.
+     * Do whatever is needed to optimize for it: {@value}.
+     */
+    public static final String FS_OPTION_OPENFILE_READ_POLICY_JSON =
+        "json";
+
+    /**
+     * This is an ORC file.
+     * Do whatever is needed to optimize for it: {@value}.
+     */
+    public static final String FS_OPTION_OPENFILE_READ_POLICY_ORC =
+        "orc";
+
+    /**
+     * This is a parquet file with a v1/v3 footer: {@value}.
+     * Do whatever is needed to optimize for it, such as footer
+     * prefetch and cache,
+     */
+    public static final String FS_OPTION_OPENFILE_READ_POLICY_PARQUET =
+        "parquet";
+
     /**
      * Read policy for random IO: {@value}.
      */
@@ -634,7 +691,13 @@ private OpenFileOptions() {
     public static final Set<String> FS_OPTION_OPENFILE_READ_POLICIES =
         Collections.unmodifiableSet(Stream.of(
                 FS_OPTION_OPENFILE_READ_POLICY_ADAPTIVE,
+                FS_OPTION_OPENFILE_READ_POLICY_AVRO,
+                FS_OPTION_OPENFILE_READ_POLICY_COLUMNAR,
+                FS_OPTION_OPENFILE_READ_POLICY_CSV,
                 FS_OPTION_OPENFILE_READ_POLICY_DEFAULT,
+                FS_OPTION_OPENFILE_READ_POLICY_JSON,
+                FS_OPTION_OPENFILE_READ_POLICY_ORC,
+                FS_OPTION_OPENFILE_READ_POLICY_PARQUET,
                 FS_OPTION_OPENFILE_READ_POLICY_RANDOM,
                 FS_OPTION_OPENFILE_READ_POLICY_SEQUENTIAL,
                 FS_OPTION_OPENFILE_READ_POLICY_VECTOR,

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

@@ -1320,6 +1320,8 @@ public boolean hasPathCapability(final Path path, final String capability)
     case CommonPathCapabilities.FS_PATHHANDLES:
     case CommonPathCapabilities.FS_PERMISSIONS:
     case CommonPathCapabilities.FS_TRUNCATE:
+      // block locations are generated locally
+    case CommonPathCapabilities.VIRTUAL_BLOCK_LOCATIONS:
       return true;
     case CommonPathCapabilities.FS_SYMLINKS:
       return FileSystem.areSymlinksEnabled();

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

@@ -18,17 +18,30 @@
 
 package org.apache.hadoop.io.wrappedio;
 
+import java.io.IOException;
+import java.io.InputStream;
 import java.io.UncheckedIOException;
+import java.nio.ByteBuffer;
 import java.util.Collection;
 import java.util.List;
 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.BulkDelete;
+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.PathCapabilities;
+import org.apache.hadoop.fs.StreamCapabilities;
+import org.apache.hadoop.util.functional.FutureIO;
 
+import static org.apache.hadoop.fs.Options.OpenFileOptions.FS_OPTION_OPENFILE_LENGTH;
+import static org.apache.hadoop.fs.Options.OpenFileOptions.FS_OPTION_OPENFILE_READ_POLICY;
 import static org.apache.hadoop.util.functional.FunctionalIO.uncheckIOExceptions;
 
 /**
@@ -82,7 +95,8 @@ public static int bulkDelete_pageSize(FileSystem fs, Path path) {
    * @param fs filesystem
    * @param base path to delete under.
    * @param paths list of paths which must be absolute and under the base path.
-   * @return a list of all the paths which couldn't be deleted for a reason other than "not found" and any associated error message.
+   * @return a list of all the paths which couldn't be deleted for a reason other
+   *          than "not found" and any associated error message.
    * @throws UnsupportedOperationException bulk delete under that path is not supported.
    * @throws UncheckedIOException if an IOE was raised.
    * @throws IllegalArgumentException if a path argument is invalid.
@@ -97,4 +111,137 @@ public static List<Map.Entry<Path, String>> bulkDelete_delete(FileSystem fs,
       }
     });
   }
+
+  /**
+   * Does a path have a given capability?
+   * Calls {@link PathCapabilities#hasPathCapability(Path, String)},
+   * mapping IOExceptions to false.
+   * @param fs filesystem
+   * @param path path to query the capability of.
+   * @param capability non-null, non-empty string to query the path for support.
+   * @return true if the capability is supported under that part of the FS.
+   * resolving paths or relaying the call.
+   * @throws IllegalArgumentException invalid arguments
+   */
+  public static boolean pathCapabilities_hasPathCapability(Object fs,
+      Path path,
+      String capability) {
+    try {
+      return ((PathCapabilities) 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?
+   * Calls {@link StreamCapabilities#hasCapability(String)},
+   * @param object object to probe
+   * @param capability capability string
+   * @return true iff the object implements StreamCapabilities and the capability is
+   * declared available.
+   */
+  public static boolean streamCapabilities_hasCapability(Object object, String capability) {
+    if (!(object instanceof StreamCapabilities)) {
+      return false;
+    }
+    return ((StreamCapabilities) object).hasCapability(capability);
+  }
+
+  /**
+   * OpenFile assistant, easy reflection-based access to
+   * {@link FileSystem#openFile(Path)} and blocks
+   * awaiting the operation completion.
+   * @param fs filesystem
+   * @param path path
+   * @param policy read policy
+   * @param status optional file status
+   * @param length optional file length
+   * @param options nullable map of other options
+   * @return stream of the opened file
+   * @throws UncheckedIOException if an IOE was raised.
+   */
+  @InterfaceStability.Stable
+  public static FSDataInputStream fileSystem_openFile(
+      final FileSystem fs,
+      final Path path,
+      final String policy,
+      @Nullable final FileStatus status,
+      @Nullable final Long length,
+      @Nullable final Map<String, String> options) {
+    final FutureDataInputStreamBuilder builder = uncheckIOExceptions(() ->
+        fs.openFile(path));
+    if (policy != null) {
+      builder.opt(FS_OPTION_OPENFILE_READ_POLICY, policy);
+    }
+    if (status != null) {
+      builder.withFileStatus(status);
+    }
+    if (length != null) {
+      builder.opt(FS_OPTION_OPENFILE_LENGTH, Long.toString(length));
+    }
+    if (options != null) {
+      // add all the options map entries
+      options.forEach(builder::opt);
+    }
+    // wait for the opening.
+    return uncheckIOExceptions(() ->
+        FutureIO.awaitFuture(builder.build()));
+  }
+
+  /**
+   * Return path of the enclosing root for a given path.
+   * The enclosing root path is a common ancestor that should be used for temp and staging dirs
+   * as well as within encryption zones and other restricted directories.
+   * @param fs filesystem
+   * @param path file path to find the enclosing root path for
+   * @return a path to the enclosing root
+   * @throws IOException early checks like failure to resolve path cause IO failures
+   */
+  public static Path fileSystem_getEnclosingRoot(FileSystem fs, Path path) throws IOException {
+    return fs.getEnclosingRoot(path);
+  }
+
+  /**
+   * 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.
+   * Note: that is the default behaviour of {@link FSDataInputStream#readFully(long, ByteBuffer)}.
+   */
+  public static void byteBufferPositionedReadable_readFully(
+      InputStream in,
+      long position,
+      ByteBuffer buf) {
+    if (!(in instanceof ByteBufferPositionedReadable)) {
+      throw new UnsupportedOperationException("Not a ByteBufferPositionedReadable: " + in);
+    }
+    uncheckIOExceptions(() -> {
+      ((ByteBufferPositionedReadable) in).readFully(position, buf);
+      return null;
+    });
+  }
+
+  /**
+   * 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 byteBufferPositionedReadable_readFullyAvailable(
+      InputStream in) {
+    if (!(in instanceof ByteBufferPositionedReadable)) {
+      return false;
+    }
+    if (in instanceof FSDataInputStream) {
+      // ask the wrapped stream.
+      return byteBufferPositionedReadable_readFullyAvailable(
+          ((FSDataInputStream) in).getWrappedStream());
+    }
+    // now rely on the input stream implementing path capabilities, which
+    // all the Hadoop FS implementations do.
+    return streamCapabilities_hasCapability(in, StreamCapabilities.PREADBYTEBUFFER);
+  }
 }