PARQUET-2493. pull read policy declaration into reader code

Change-Id: I7de43d8426b56800c540a520f1fb7fef21ae60ba

Author: steveloughran

parquet-hadoop/pom.xml

@@ -160,7 +160,11 @@
       <artifactId>zstd-jni</artifactId>
       <version>${zstd-jni.version}</version>
     </dependency>
-
+    <dependency>
+      <groupId>com.google.code.findbugs</groupId>
+      <artifactId>jsr305</artifactId>
+      <version>${jsr305.version}</version>
+    </dependency>
     <dependency>
       <groupId>com.google.guava</groupId>
       <artifactId>guava</artifactId>

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

@@ -73,7 +73,7 @@ public long getLength() {
 
   @Override
   public SeekableInputStream newStream() throws IOException {
-    return HadoopStreams.wrap(DynamicWrappedIO.openFile(fs, stat));
+    return HadoopStreams.wrap(DynamicWrappedIO.openFile(fs, stat, DynamicWrappedIO.PARQUET_READ_POLICIES));
   }
 
   @Override

parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/wrapped/io/DynamicWrappedIO.java

@@ -374,15 +374,16 @@ public static boolean isAvailable() {
    * <p>
    * If the WrappedIO class is found, uses
    * {@link #fileSystem_openFile(FileSystem, Path, String, FileStatus, Long, Map)} with
-   * {@link #PARQUET_READ_POLICIES} as the list of read policies and passing down
+   * the supplied list of read policies and passing down
    * the file status.
    * <p>
    * If not, falls back to the classic {@code fs.open(Path)} call.
    * @param fs filesystem
    * @param status file status
+   * @param policy  read policy
    * @throws IOException any IO failure.
    */
-  public static FSDataInputStream openFile(FileSystem fs, FileStatus status) throws IOException {
+  public static FSDataInputStream openFile(FileSystem fs, FileStatus status, String policy) throws IOException {
     final DynamicWrappedIO instance = DynamicWrappedIO.instance();
     FSDataInputStream stream;
     if (instance.fileSystem_openFile_available()) {
@@ -392,8 +393,8 @@ public static FSDataInputStream openFile(FileSystem fs, FileStatus status) throw
       // in open and choosing the range for GET requests.
       // For other stores, it ultimately invokes the classic open(Path)
       // call so is no more expensive than before.
-      LOG.debug("Opening file {} through fileSystem_openFile", status);
-      stream = instance.fileSystem_openFile(fs, status.getPath(), PARQUET_READ_POLICIES, status, null, null);
+      LOG.debug("Opening file {} through fileSystem_openFile() with policy {}", status, policy);
+      stream = instance.fileSystem_openFile(fs, status.getPath(), policy, status, null, null);
     } else {
       LOG.debug("Opening file {} through open()", status);
       stream = fs.open(status.getPath());

parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestParquetFileWriter.java

@@ -652,7 +652,7 @@ public void testAlignmentWithPadding() throws Exception {
     long fileLen = stat.getLen();
 
     long footerLen;
-    try (FSDataInputStream data = DynamicWrappedIO.openFile(fs, stat)) {
+    try (FSDataInputStream data = DynamicWrappedIO.openFile(fs, stat, DynamicWrappedIO.PARQUET_READ_POLICIES)) {
       data.seek(fileLen - 8); // 4-byte offset + "PAR1"
       footerLen = BytesUtils.readIntLittleEndian(data);
     }