allow reading footers from provided file listing and streams

Author: Robert Kruszewski

parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java

@@ -236,6 +236,36 @@ public static List<Footer> readAllFootersInParallel(final Configuration configur
     return readAllFootersInParallel(configuration, partFiles, false);
   }
 
+  /**
+   * read all the footers of the files provided with input streams
+   * (not using summary files)
+   * @param configuration the conf to access the File System
+   * @param partFiles the files to read with equivalent input streams
+   * @param skipRowGroups to skip the rowGroup info
+   * @return the footers
+   * @throws IOException
+   */
+  public static List<Footer> readAllFootersInParallel(final Configuration configuration, Map<FileStatus, FSDataInputStream> partFiles, final boolean skipRowGroups) throws IOException {
+    List<Callable<Footer>> footers = new ArrayList<Callable<Footer>>();
+    for (final Map.Entry<FileStatus, FSDataInputStream> currentFile : partFiles.entrySet()) {
+      footers.add(new Callable<Footer>() {
+        @Override
+        public Footer call() throws Exception {
+          try {
+            return new Footer(currentFile.getKey().getPath(), readFooter(currentFile.getValue(), currentFile.getKey(), filter(skipRowGroups)));
+          } catch (IOException e) {
+            throw new IOException("Could not read footer for file " + currentFile, e);
+          }
+        }
+      });
+    }
+    try {
+      return runAllInParallel(configuration.getInt(PARQUET_READ_PARALLELISM, 5), footers);
+    } catch (ExecutionException e) {
+      throw new IOException("Could not read footer: " + e.getMessage(), e.getCause());
+    }
+  }
+
   /**
    * read all the footers of the files provided
    * (not using summary files)
@@ -434,22 +464,33 @@ public static final ParquetMetadata readFooter(Configuration configuration, File
     FileSystem fileSystem = file.getPath().getFileSystem(configuration);
     FSDataInputStream in = fileSystem.open(file.getPath());
     try {
-      return readFooter(file, in, filter);
+      return readFooter(in, file, filter);
     } finally {
       in.close();
     }
   }
 
-  private static final ParquetMetadata readFooter(FileStatus file, FSDataInputStream f, MetadataFilter filter) throws IOException {
-    long l = file.getLen();
+  /**
+   * Reads the meta data block in the footer of the file using provided input stream
+   * @param f input stream for the file
+   * @param file the parquet File
+   * @param filter the filter to apply to row groups
+   * @return the metadata blocks in the footer
+   * @throws IOException if an error occurs while reading the file
+   */
+  public static final ParquetMetadata readFooter(FSDataInputStream f, FileStatus file, MetadataFilter filter) throws IOException {
+    return readFooter(f, file.getLen(), file.getPath().toString(), filter);
+  }
+
+  private static final ParquetMetadata readFooter(FSDataInputStream f, long fileLen, String filePath, MetadataFilter filter) throws IOException {
     if (Log.DEBUG) {
-      LOG.debug("File length " + l);
+      LOG.debug("File length " + fileLen);
     }
     int FOOTER_LENGTH_SIZE = 4;
-    if (l < MAGIC.length + FOOTER_LENGTH_SIZE + MAGIC.length) { // MAGIC + data + footer + footerIndex + MAGIC
-      throw new RuntimeException(file.getPath() + " is not a Parquet file (too small)");
+    if (fileLen < MAGIC.length + FOOTER_LENGTH_SIZE + MAGIC.length) { // MAGIC + data + footer + footerIndex + MAGIC
+      throw new RuntimeException(filePath + " is not a Parquet file (too small)");
     }
-    long footerLengthIndex = l - FOOTER_LENGTH_SIZE - MAGIC.length;
+    long footerLengthIndex = fileLen - FOOTER_LENGTH_SIZE - MAGIC.length;
     if (Log.DEBUG) {
       LOG.debug("reading footer index at " + footerLengthIndex);
     }
@@ -459,7 +500,7 @@ private static final ParquetMetadata readFooter(FileStatus file, FSDataInputStre
     byte[] magic = new byte[MAGIC.length];
     f.readFully(magic);
     if (!Arrays.equals(MAGIC, magic)) {
-      throw new RuntimeException(file.getPath() + " is not a Parquet file. expected magic number at tail " + Arrays.toString(MAGIC) + " but found " + Arrays.toString(magic));
+      throw new RuntimeException(filePath + " is not a Parquet file. expected magic number at tail " + Arrays.toString(MAGIC) + " but found " + Arrays.toString(magic));
     }
     long footerIndex = footerLengthIndex - footerLength;
     if (Log.DEBUG) {