HADOOP-19131. add "hbase" as a format

declare that hbase is an hbase table; s3a maps to random IO.
abfs recommends disabling prefetch for these files...it should do
it automatically when support for read policies is wired up.

Change-Id: Ia072c0f724f7d7cb538ef9d0a047e74c0191af73

Author: steveloughran

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

@@ -631,6 +631,13 @@ private OpenFileOptions() {
     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.

hadoop-common-project/hadoop-common/src/site/markdown/filesystem/fsdatainputstreambuilder.md

@@ -382,22 +382,26 @@ performance -and vice versa.
 | `adaptive`   | Any adaptive policy implemented by the store.                          |
 | `avro`       | This is an avro format which will be read sequentially                 |
 | `csv`        | This is CSV data which will be read sequentially                       |
-| `json`       | This is a UTF-8 JSON/JSON lines format which will be read sequentially |
 | `default`    | The default policy for this store. Generally "adaptive".               |
 | `columnar`   | This is any columnar format other than ORC/parquet.                    |
+| `hbase`      | This is an HBase Table                                                 |
+| `json`       | This is a UTF-8 JSON/JSON lines format which will be read sequentially |
 | `orc`        | This is an ORC file. Optimize for it.                                  |
 | `parquet`    | This is a Parquet file. Optimize for it.                               |
 | `random`     | Optimize for random access.                                            |
 | `sequential` | Optimize for sequential access.                                        |
 | `vector`     | The Vectored IO API is intended to be used.                            |
 | `whole-file` | The whole file will be read.                                           |
 
-Choosing the wrong read policy for an input source may be inefficient.
+Choosing the wrong read policy for an input source may be inefficient but never fatal.
 
 A list of read policies MAY be supplied; the first one recognized/supported by
-the filesystem SHALL be the one used. This allows for custom policies to be
-supported, for example an `hbase-hfile` policy optimized for HBase HFiles.
-
+the filesystem SHALL be the one used. This allows for configurations which are compatible
+across versions. A policy `parquet, columnar, vector, random, adaptive` will use the parquet policy for
+any filesystem aware of it, falling back to `columnar`, `vector`, `random` and finally `adaptive`.
+The S3A connector will recognize the `random` since Hadoop 3.3.5 (i.e. since the `openFile()` API
+was added), and `vector` from Hadoop 3.4.0.
+ 
 The S3A and ABFS input streams both implement
 the [IOStatisticsSource](iostatistics.html) API, and can be queried for their IO
 Performance.
@@ -479,7 +483,6 @@ Strategies can include:
 Applications which know that the entire file is to be read from an opened stream SHOULD declare this
 read policy.
 
-
 #### <a name="read.policy.columnar"></a> Read Policy `columnar`
 
 Declare that the data is some (unspecific) columnar format and that read sequencies
@@ -511,6 +514,13 @@ libraries targeting multiple versions, SHOULD list their fallback
 policies if these are not recognized, e.g. request a policy such as `avro, sequential`.
 
 
+#### <a name="read.policy.fileformat.hbase"></a> File Format Read Policy `hbase`
+
+The file is an HBase table.
+Use whatever policy is appropriate for these files, where `random` is
+what should be used unless there are specific optimizations related to HBase.
+
+
 ### <a name="openfile.length"></a> Option: `fs.option.openfile.length`: `Long`
 
 Declare the length of a file.

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/util/functional/TestFunctionalIO.java

@@ -99,9 +99,13 @@ public void testUncheckAndExtract() throws Throwable {
 
   @Test
   public void testUncheckedFunction() throws Throwable {
+    // java function which should raise a FileNotFoundException
+    // wrapped into an unchecked exeption
     final Function<String, Object> fn =
-        toUncheckedFunction((String a) -> {throw new FileNotFoundException(a);});
-    intercept(UncheckedIOException.class, "404", () ->
+        toUncheckedFunction((String a) -> {
+          throw new FileNotFoundException(a);
+        });
+    intercept(UncheckedIOException.class, "missing", () ->
         fn.apply("missing"));
   }
 }

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

@@ -30,6 +30,7 @@
 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_CSV;
 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_HBASE;
 import static org.apache.hadoop.fs.Options.OpenFileOptions.FS_OPTION_OPENFILE_READ_POLICY_JSON;
 import static org.apache.hadoop.fs.Options.OpenFileOptions.FS_OPTION_OPENFILE_READ_POLICY_ORC;
 import static org.apache.hadoop.fs.Options.OpenFileOptions.FS_OPTION_OPENFILE_READ_POLICY_PARQUET;
@@ -87,7 +88,8 @@ boolean isAdaptive() {
    * Choose an access policy.
    * @param name strategy name from a configuration option, etc.
    * @param defaultPolicy default policy to fall back to.
-   * @return the chosen strategy
+   * @return the chosen strategy or null if there was no match and
+   * the value of {@code defaultPolicy} was "null".
    */
   public static S3AInputPolicy getPolicy(
       String name,
@@ -100,6 +102,7 @@ public static S3AInputPolicy getPolicy(
       return Normal;
 
       // all these options currently map to random IO.
+    case FS_OPTION_OPENFILE_READ_POLICY_HBASE:
     case FS_OPTION_OPENFILE_READ_POLICY_RANDOM:
     case FS_OPTION_OPENFILE_READ_POLICY_VECTOR:
       return Random;
@@ -111,7 +114,7 @@ public static S3AInputPolicy getPolicy(
     case FS_OPTION_OPENFILE_READ_POLICY_PARQUET:
       return Random;
 
-      // hadle the sequential formats.
+      // handle the sequential formats.
     case FS_OPTION_OPENFILE_READ_POLICY_AVRO:
     case FS_OPTION_OPENFILE_READ_POLICY_CSV:
     case FS_OPTION_OPENFILE_READ_POLICY_JSON:

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/impl/TestOpenFileSupport.java

@@ -43,14 +43,26 @@
 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.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_CSV;
 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_HBASE;
+import static org.apache.hadoop.fs.Options.OpenFileOptions.FS_OPTION_OPENFILE_READ_POLICY_JSON;
+import static org.apache.hadoop.fs.Options.OpenFileOptions.FS_OPTION_OPENFILE_READ_POLICY_ORC;
+import static org.apache.hadoop.fs.Options.OpenFileOptions.FS_OPTION_OPENFILE_READ_POLICY_PARQUET;
 import static org.apache.hadoop.fs.Options.OpenFileOptions.FS_OPTION_OPENFILE_READ_POLICY_RANDOM;
 import static org.apache.hadoop.fs.Options.OpenFileOptions.FS_OPTION_OPENFILE_READ_POLICY_SEQUENTIAL;
+import static org.apache.hadoop.fs.Options.OpenFileOptions.FS_OPTION_OPENFILE_READ_POLICY_VECTOR;
+import static org.apache.hadoop.fs.Options.OpenFileOptions.FS_OPTION_OPENFILE_READ_POLICY_WHOLE_FILE;
 import static org.apache.hadoop.fs.Options.OpenFileOptions.FS_OPTION_OPENFILE_SPLIT_END;
 import static org.apache.hadoop.fs.Options.OpenFileOptions.FS_OPTION_OPENFILE_SPLIT_START;
 import static org.apache.hadoop.fs.s3a.Constants.DEFAULT_ASYNC_DRAIN_THRESHOLD;
 import static org.apache.hadoop.fs.s3a.Constants.INPUT_FADVISE;
 import static org.apache.hadoop.fs.s3a.Constants.READAHEAD_RANGE;
+import static org.apache.hadoop.fs.s3a.S3AInputPolicy.Normal;
+import static org.apache.hadoop.fs.s3a.S3AInputPolicy.Random;
+import static org.apache.hadoop.fs.s3a.S3AInputPolicy.Sequential;
 import static org.apache.hadoop.test.LambdaTestUtils.intercept;
 
 /**
@@ -69,7 +81,7 @@ public class TestOpenFileSupport extends HadoopTestBase {
 
   private static final String USERNAME = "hadoop";
 
-  public static final S3AInputPolicy INPUT_POLICY = S3AInputPolicy.Sequential;
+  public static final S3AInputPolicy INPUT_POLICY = Sequential;
 
   public static final String TESTFILE = "s3a://bucket/name";
 
@@ -142,7 +154,7 @@ public void testSeekRandomIOPolicy() throws Throwable {
     // is picked up
     assertOpenFile(INPUT_FADVISE, option)
         .extracting(f -> f.getInputPolicy())
-        .isEqualTo(S3AInputPolicy.Random);
+        .isEqualTo(Random);
     // and as neither status nor length was set: no file status
     assertOpenFile(INPUT_FADVISE, option)
         .extracting(f -> f.getStatus())
@@ -161,7 +173,7 @@ public void testSeekPolicyAdaptive() throws Throwable {
     assertOpenFile(FS_OPTION_OPENFILE_READ_POLICY,
         FS_OPTION_OPENFILE_READ_POLICY_ADAPTIVE)
         .extracting(f -> f.getInputPolicy())
-        .isEqualTo(S3AInputPolicy.Normal);
+        .isEqualTo(Normal);
   }
 
   /**
@@ -184,7 +196,7 @@ public void testSeekPolicyListS3AOption() throws Throwable {
     // fall back to the second seek policy if the first is unknown
     assertOpenFile(INPUT_FADVISE, "hbase, random")
         .extracting(f -> f.getInputPolicy())
-        .isEqualTo(S3AInputPolicy.Random);
+        .isEqualTo(Random);
   }
 
   /**
@@ -199,14 +211,14 @@ public void testSeekPolicyExtractionFromList() throws Throwable {
         FS_OPTION_OPENFILE_READ_POLICY);
     Assertions.assertThat(S3AInputPolicy.getFirstSupportedPolicy(options, null))
         .describedAs("Policy from " + plist)
-        .isEqualTo(S3AInputPolicy.Random);
+        .isEqualTo(Random);
   }
 
   @Test
   public void testAdaptiveSeekPolicyRecognized() throws Throwable {
     Assertions.assertThat(S3AInputPolicy.getPolicy("adaptive", null))
         .describedAs("adaptive")
-        .isEqualTo(S3AInputPolicy.Normal);
+        .isEqualTo(Normal);
   }
 
   @Test
@@ -222,11 +234,20 @@ public void testUnknownSeekPolicyFallback() throws Throwable {
   @Test
   public void testInputPolicyMapping() throws Throwable {
     Object[][] policyMapping = {
-        {"normal", S3AInputPolicy.Normal},
-        {FS_OPTION_OPENFILE_READ_POLICY_ADAPTIVE, S3AInputPolicy.Normal},
-        {FS_OPTION_OPENFILE_READ_POLICY_DEFAULT, S3AInputPolicy.Normal},
-        {FS_OPTION_OPENFILE_READ_POLICY_RANDOM, S3AInputPolicy.Random},
-        {FS_OPTION_OPENFILE_READ_POLICY_SEQUENTIAL, S3AInputPolicy.Sequential},
+        {"normal", Normal},
+        {FS_OPTION_OPENFILE_READ_POLICY_ADAPTIVE, Normal},
+        {FS_OPTION_OPENFILE_READ_POLICY_AVRO, Sequential},
+        {FS_OPTION_OPENFILE_READ_POLICY_COLUMNAR, Random},
+        {FS_OPTION_OPENFILE_READ_POLICY_CSV, Sequential},
+        {FS_OPTION_OPENFILE_READ_POLICY_DEFAULT, Normal},
+        {FS_OPTION_OPENFILE_READ_POLICY_HBASE, Random},
+        {FS_OPTION_OPENFILE_READ_POLICY_JSON, Sequential},
+        {FS_OPTION_OPENFILE_READ_POLICY_ORC, Random},
+        {FS_OPTION_OPENFILE_READ_POLICY_PARQUET, Random},
+        {FS_OPTION_OPENFILE_READ_POLICY_RANDOM, Random},
+        {FS_OPTION_OPENFILE_READ_POLICY_SEQUENTIAL, Sequential},
+        {FS_OPTION_OPENFILE_READ_POLICY_VECTOR, Random},
+        {FS_OPTION_OPENFILE_READ_POLICY_WHOLE_FILE, Sequential},
     };
     for (Object[] mapping : policyMapping) {
       String name = (String) mapping[0];