GH-3697: Reuse hadoop file status and footer in ParquetRecordReader

parquet-cli/src/main/java/org/apache/parquet/cli/commands/ShowFooterCommand.java

@@ -24,6 +24,7 @@
 import com.beust.jcommander.Parameter;
 import com.beust.jcommander.Parameters;
 import com.fasterxml.jackson.annotation.JsonAutoDetect.Visibility;
+import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.PropertyAccessor;
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -64,13 +65,24 @@ public int run() throws IOException {
     return 0;
   }
 
+  /**
+   * The footer records the file it was read from, which is of no interest here and not something Jackson can walk.
+   * Its own {@code @JsonIgnore} is not visible to this mapper, as the annotations in parquet-hadoop are relocated
+   * into {@code shaded.parquet}, so the property is dropped with a mix-in instead.
+   */
+  abstract static class MixIn {
+    @JsonIgnore
+    abstract InputFile getInputFile();
+  }
+
   private String readFooter(InputFile inputFile) throws JsonProcessingException, IOException {
     String json;
     try (ParquetFileReader reader = ParquetFileReader.open(inputFile)) {
       ParquetMetadata footer = reader.getFooter();
       ObjectMapper mapper = RawUtils.createObjectMapper();
       mapper.setVisibility(PropertyAccessor.ALL, Visibility.NONE);
       mapper.setVisibility(PropertyAccessor.FIELD, Visibility.ANY);
+      mapper.addMixIn(ParquetMetadata.class, MixIn.class);
       json = mapper.writerWithDefaultPrettyPrinter().writeValueAsString(footer);
     }
     return json;

parquet-cli/src/test/java/org/apache/parquet/cli/commands/ShowFooterCommandTest.java

@@ -21,9 +21,18 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 
+import com.fasterxml.jackson.annotation.JsonAutoDetect.Visibility;
+import com.fasterxml.jackson.annotation.PropertyAccessor;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import java.io.File;
 import java.io.IOException;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.parquet.cli.util.RawUtils;
+import org.apache.parquet.hadoop.ParquetFileReader;
+import org.apache.parquet.hadoop.metadata.ParquetMetadata;
+import org.apache.parquet.hadoop.util.HadoopInputFile;
+import org.apache.parquet.io.InputFile;
 import org.junit.jupiter.api.Test;
 
 public class ShowFooterCommandTest extends ParquetFileTest {
@@ -39,4 +48,24 @@ public void testShowDirectoryCommand() throws IOException {
     command.raw = true;
     assertThat(command.run()).isZero();
   }
+
+  /**
+   * The file a footer was read from is not part of the printed footer, even though this mapper serializes fields
+   * rather than getters and cannot see the relocated annotations of parquet-hadoop.
+   */
+  @Test
+  public void testInputFileNotPrinted() throws IOException {
+    InputFile inputFile = HadoopInputFile.fromPath(new Path(parquetFile().getAbsolutePath()), new Configuration());
+    ParquetMetadata footer;
+    try (ParquetFileReader reader = ParquetFileReader.open(inputFile)) {
+      footer = reader.getFooter();
+    }
+    assertThat(footer.getInputFile()).isNotNull();
+
+    ObjectMapper mapper = RawUtils.createObjectMapper();
+    mapper.setVisibility(PropertyAccessor.ALL, Visibility.NONE);
+    mapper.setVisibility(PropertyAccessor.FIELD, Visibility.ANY);
+    mapper.addMixIn(ParquetMetadata.class, ShowFooterCommand.MixIn.class);
+    assertThat(mapper.writeValueAsString(footer)).doesNotContain("inputFile");
+  }
 }

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

@@ -646,8 +646,10 @@ private static final ParquetMetadata readFooter(
 
       // Regular file, or encrypted file with plaintext footer
       if (!encryptedFooterMode) {
-        return converter.readParquetMetadata(
+        ParquetMetadata parquetMetadata = converter.readParquetMetadata(
             footerBytesStream, options.getMetadataFilter(), fileDecryptor, false, fileMetadataLength);
+        parquetMetadata.setInputFile(file);
+        return parquetMetadata;
       }
 
       // Encrypted file with encrypted footer
@@ -658,8 +660,10 @@ private static final ParquetMetadata readFooter(
       fileDecryptor.setFileCryptoMetaData(
           fileCryptoMetaData.getEncryption_algorithm(), true, fileCryptoMetaData.getKey_metadata());
       // footer length is required only for signed plaintext footers
-      return converter.readParquetMetadata(
+      ParquetMetadata parquetMetadata = converter.readParquetMetadata(
           footerBytesStream, options.getMetadataFilter(), fileDecryptor, true, 0);
+      parquetMetadata.setInputFile(file);
+      return parquetMetadata;
     } finally {
       options.getAllocator().release(footerBytesBuffer);
     }

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

@@ -18,6 +18,7 @@
  */
 package org.apache.parquet.hadoop;
 
+import com.fasterxml.jackson.annotation.JsonIgnore;
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
 import java.io.DataInput;
@@ -36,6 +37,7 @@
 import org.apache.hadoop.mapreduce.lib.input.FileSplit;
 import org.apache.parquet.hadoop.metadata.BlockMetaData;
 import org.apache.parquet.hadoop.metadata.ColumnChunkMetaData;
+import org.apache.parquet.hadoop.metadata.ParquetMetadata;
 import org.apache.parquet.schema.MessageType;
 import org.apache.parquet.schema.MessageTypeParser;
 
@@ -55,6 +57,13 @@ public class ParquetInputSplit extends FileSplit implements Writable {
   private long end;
   private long[] rowGroupOffsets;
 
+  /**
+   * Footer of the file, if the split was built by a caller which had already read it.
+   * Not written by {@link #write(DataOutput)}, so it is only visible within the JVM which set it.
+   */
+  @JsonIgnore
+  private volatile ParquetMetadata footer;
+
   /**
    * Writables must have a parameterless constructor
    */
@@ -222,6 +231,24 @@ public long[] getRowGroupOffsets() {
     return rowGroupOffsets;
   }
 
+  /**
+   * @return the footer of the file, if it was passed in by whoever built the split, else null.
+   */
+  public ParquetMetadata getFooter() {
+    return footer;
+  }
+
+  /**
+   * Pass in a footer already read from the file, so that a reader created from this split does not have to read it
+   * again. As the footer is not serialized with the split, this only has an effect on readers created in the same
+   * JVM.
+   *
+   * @param footer footer of the file this split refers to
+   */
+  public void setFooter(ParquetMetadata footer) {
+    this.footer = footer;
+  }
+
   @Override
   public String toString() {
     String hosts;

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

@@ -43,9 +43,11 @@
 import org.apache.parquet.hadoop.metadata.BlockMetaData;
 import org.apache.parquet.hadoop.metadata.ColumnChunkMetaData;
 import org.apache.parquet.hadoop.metadata.FileMetaData;
+import org.apache.parquet.hadoop.metadata.ParquetMetadata;
 import org.apache.parquet.hadoop.util.ContextUtil;
 import org.apache.parquet.hadoop.util.HadoopInputFile;
 import org.apache.parquet.hadoop.util.counters.BenchmarkCounter;
+import org.apache.parquet.io.InputFile;
 import org.apache.parquet.io.ParquetDecodingException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -154,9 +156,16 @@ private void initializeInternalReader(ParquetInputSplit split, Configuration con
       optionsBuilder.withRange(split.getStart(), split.getEnd());
     }
 
-    // open a reader with the metadata filter
-    ParquetFileReader reader =
-        ParquetFileReader.open(HadoopInputFile.fromPath(path, configuration), optionsBuilder.build());
+    // open a reader with the metadata filter, reusing the footer of the split and the file it was read from
+    // when the split was built by a caller which had already read them
+    ParquetMetadata footer = split.getFooter();
+    InputFile inputFile = footer != null && footer.getInputFile() != null
+        ? footer.getInputFile()
+        : HadoopInputFile.fromPath(path, configuration);
+    ParquetReadOptions options = optionsBuilder.build();
+    ParquetFileReader reader = footer != null
+        ? ParquetFileReader.open(inputFile, footer, options, inputFile.newStream())
+        : ParquetFileReader.open(inputFile, options);
 
     if (rowGroupOffsets != null) {
       // verify a row group was found for each offset

parquet-hadoop/src/main/java/org/apache/parquet/hadoop/metadata/ParquetMetadata.java

@@ -18,13 +18,15 @@
  */
 package org.apache.parquet.hadoop.metadata;
 
+import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.ObjectWriter;
 import com.fasterxml.jackson.databind.SerializationFeature;
 import java.io.IOException;
 import java.io.StringReader;
 import java.io.StringWriter;
 import java.util.List;
+import org.apache.parquet.io.InputFile;
 
 /**
  * Metadata block stored in the footer of the file
@@ -91,6 +93,12 @@ public static ParquetMetadata fromJSON(String json) {
   private final FileMetaData fileMetaData;
   private final List<BlockMetaData> blocks;
 
+  /**
+   * The file this metadata was read from, if known; not part of the serialized footer.
+   */
+  @JsonIgnore
+  private volatile InputFile inputFile;
+
   /**
    * @param fileMetaData file level metadata
    * @param blocks       block level metadata
@@ -114,6 +122,26 @@ public FileMetaData getFileMetaData() {
     return fileMetaData;
   }
 
+  /**
+   * The file this metadata was read from. Set when the footer is read through
+   * {@link org.apache.parquet.hadoop.ParquetFileReader}; a caller which already holds the metadata can pass this
+   * file on to a new reader rather than opening the path again.
+   *
+   * @return the file the footer was read from, or null if it is not known
+   */
+  public InputFile getInputFile() {
+    return inputFile;
+  }
+
+  /**
+   * Record the file this metadata was read from, so that it can be reused when opening a reader over the same file.
+   *
+   * @param inputFile the file the footer was read from
+   */
+  public void setInputFile(InputFile inputFile) {
+    this.inputFile = inputFile;
+  }
+
   @Override
   public String toString() {
     return "ParquetMetaData{" + fileMetaData + ", blocks: " + blocks + "}";