HADOOP-19131. changing names; adding avro for a read policy

...and mapping to sequential in s3a

Change-Id: Iebcac57a4b4ff3fc6edc82f120a7448635bd61bc

Author: steveloughran

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

@@ -598,6 +598,11 @@ private OpenFileOptions() {
     public static final String FS_OPTION_OPENFILE_READ_POLICY_ADAPTIVE =
         "adaptive";
 
+    /**
+     * 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}.
@@ -649,12 +654,14 @@ private OpenFileOptions() {
     public static final String FS_OPTION_OPENFILE_READ_POLICY_WHOLE_FILE =
         "whole-file";
 
+
     /**
      * All the current read policies as a set.
      */
     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_DEFAULT,
                 FS_OPTION_OPENFILE_READ_POLICY_ORC,
                 FS_OPTION_OPENFILE_READ_POLICY_PARQUET,

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

@@ -44,6 +44,7 @@
 import org.apache.hadoop.util.functional.FunctionRaisingIOE;
 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.fs.statistics.IOStatisticsContext.getCurrentIOStatisticsContext;
 import static org.apache.hadoop.fs.statistics.IOStatisticsSupport.retrieveIOStatistics;
@@ -79,7 +80,7 @@ public static boolean hasPathCapability(FileSystem fs, Path path, String capabil
    * @param capability capability string
    * @return true iff the capability is declared available.
    */
-  public static boolean hasStreamCapability(Object o, String capability) {
+  public static boolean streamCapabilities_hasStreamCapability(Object o, String capability) {
     if (o instanceof StreamCapabilities) {
       return false;
     }
@@ -98,11 +99,12 @@ public static boolean hasStreamCapability(Object o, String capability) {
    * @throws IOException for any failure.
    */
   @InterfaceStability.Stable
-  public static FSDataInputStream filesystemOpenFile(
+  public static FSDataInputStream fileSystem_openFile(
       final FileSystem fs,
       final Path path,
       final String policy,
       final FileStatus status,
+      final Long length,
       final Map<String, String> options) throws IOException {
     final FutureDataInputStreamBuilder builder = fs.openFile(path);
     if (policy != null) {
@@ -111,6 +113,9 @@ public static FSDataInputStream filesystemOpenFile(
     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);
@@ -128,7 +133,7 @@ public static FSDataInputStream filesystemOpenFile(
    * @return a path to the enclosing root
    * @throws IOException early checks like failure to resolve path cause IO failures
    */
-  public static Path filesystemGetEnclosingRoot(FileSystem fs, Path path) throws IOException {
+  public static Path fileSystem_getEnclosingRoot(FileSystem fs, Path path) throws IOException {
     return fs.getEnclosingRoot(path);
   }
 
@@ -316,7 +321,7 @@ public static Serializable iostatisticsContextSnapshot() {
    * 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(
+  public static void byteBufferPositionedReadable_readFully(
       Object in,
       long position,
       ByteBuffer buf)
@@ -335,18 +340,18 @@ public static void byteBufferPositionedReadableReadFully(
    * @return true if the stream implements the interface (including a wrapped stream)
    * and that it declares the stream capability.
    */
-  public static boolean byteBufferPositionedReadableReadFullyAvailable(
+  public static boolean byteBufferPositionedReadable_readFullyAvailable(
       Object in) {
     if (!(in instanceof ByteBufferPositionedReadable)) {
       return false;
     }
     if (in instanceof FSDataInputStream) {
       // ask the wrapped stream.
-      return byteBufferPositionedReadableReadFullyAvailable(
+      return byteBufferPositionedReadable_readFullyAvailable(
           ((FSDataInputStream) in).getWrappedStream());
     }
     // now rely on the input stream implementing path capabilities, which
     // all the Hadoop FS implementations do.
-    return hasStreamCapability(in, StreamCapabilities.PREADBYTEBUFFER);
+    return streamCapabilities_hasStreamCapability(in, StreamCapabilities.PREADBYTEBUFFER);
   }
 }

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AInputPolicy.java

@@ -26,6 +26,7 @@
 import org.apache.hadoop.classification.InterfaceStability;
 
 import static org.apache.hadoop.fs.Options.OpenFileOptions.FS_OPTION_OPENFILE_READ_POLICY_ADAPTIVE;
+import static org.apache.hadoop.fs.Options.OpenFileOptions.FS_OPTION_OPENFILE_READ_POLICY_AVRO;
 import static org.apache.hadoop.fs.Options.OpenFileOptions.FS_OPTION_OPENFILE_READ_POLICY_COLUMNAR;
 import static org.apache.hadoop.fs.Options.OpenFileOptions.FS_OPTION_OPENFILE_READ_POLICY_DEFAULT;
 import static org.apache.hadoop.fs.Options.OpenFileOptions.FS_OPTION_OPENFILE_READ_POLICY_ORC;
@@ -109,6 +110,7 @@ public static S3AInputPolicy getPolicy(
     case FS_OPTION_OPENFILE_READ_POLICY_PARQUET:
       return Random;
 
+    case FS_OPTION_OPENFILE_READ_POLICY_AVRO:
     case FS_OPTION_OPENFILE_READ_POLICY_SEQUENTIAL:
     case FS_OPTION_OPENFILE_READ_POLICY_WHOLE_FILE:
       return Sequential;