diff --git a/parquet-cli/src/main/java/org/apache/parquet/cli/commands/ShowFooterCommand.java b/parquet-cli/src/main/java/org/apache/parquet/cli/commands/ShowFooterCommand.java index 49a67cb723..8d07100362 100644 --- a/parquet-cli/src/main/java/org/apache/parquet/cli/commands/ShowFooterCommand.java +++ b/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,6 +65,16 @@ 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)) { @@ -71,6 +82,7 @@ private String readFooter(InputFile inputFile) throws JsonProcessingException, I 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; diff --git a/parquet-cli/src/test/java/org/apache/parquet/cli/commands/ShowFooterCommandTest.java b/parquet-cli/src/test/java/org/apache/parquet/cli/commands/ShowFooterCommandTest.java index 850910de40..97c01f124e 100644 --- a/parquet-cli/src/test/java/org/apache/parquet/cli/commands/ShowFooterCommandTest.java +++ b/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"); + } } diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java index 9af4b4ac60..a02e4ed782 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java +++ b/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); } diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetInputSplit.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetInputSplit.java index 3c65ef7c01..cf53117a8c 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetInputSplit.java +++ b/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; diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetRecordReader.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetRecordReader.java index c0e52fc5c6..76790c0b8c 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetRecordReader.java +++ b/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 diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/metadata/ParquetMetadata.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/metadata/ParquetMetadata.java index 4f6b1cc5e8..3b6b91db8d 100755 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/metadata/ParquetMetadata.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/metadata/ParquetMetadata.java @@ -18,6 +18,7 @@ */ 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; @@ -25,6 +26,7 @@ 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 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 + "}"; diff --git a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestFooterReuse.java b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestFooterReuse.java new file mode 100644 index 0000000000..3a9873e3ea --- /dev/null +++ b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestFooterReuse.java @@ -0,0 +1,341 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.parquet.hadoop; + +import static org.apache.parquet.hadoop.ParquetInputFormat.READ_SUPPORT_CLASS; +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.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.DataInputStream; +import java.io.DataOutputStream; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.atomic.AtomicLong; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.mapreduce.TaskAttemptContext; +import org.apache.hadoop.mapreduce.TaskAttemptID; +import org.apache.parquet.example.data.Group; +import org.apache.parquet.filter2.recordlevel.PhoneBookWriter; +import org.apache.parquet.hadoop.example.ExampleParquetWriter; +import org.apache.parquet.hadoop.example.GroupReadSupport; +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.io.InputFile; +import org.apache.parquet.io.SeekableInputStream; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +/** + * Tests of a footer, and the file it was read from, being passed on to a reader through + * {@link ParquetInputSplit#setFooter(ParquetMetadata)} rather than being read again. + */ +@SuppressWarnings("deprecation") +public class TestFooterReuse { + + private static final List DATA = TestParquetReader.makeUsers(100); + + /** Size of the footer length and the trailing magic: only a footer read goes here. */ + private static final int TAIL_SIZE = ParquetFileWriter.MAGIC.length + Integer.BYTES; + + @TempDir + static java.nio.file.Path tempDir; + + private static Path file; + private static long fileLength; + + @BeforeAll + public static void writeFile() throws IOException { + file = new Path(tempDir.resolve("phonebook.parquet").toString()); + PhoneBookWriter.write(ExampleParquetWriter.builder(file), DATA); + fileLength = file.getFileSystem(new Configuration()).getFileStatus(file).getLen(); + } + + @Test + public void testFooterKnowsItsInputFile() throws Exception { + InputFile inputFile = HadoopInputFile.fromPath(file, new Configuration()); + try (ParquetFileReader reader = ParquetFileReader.open(inputFile)) { + assertThat(reader.getFooter().getInputFile()).isSameAs(inputFile); + } + } + + @Test + public void testInputFileNotSerializedToJson() throws Exception { + InputFile inputFile = HadoopInputFile.fromPath(file, new Configuration()); + ParquetMetadata footer; + try (ParquetFileReader reader = ParquetFileReader.open(inputFile)) { + footer = reader.getFooter(); + } + + assertThat(footer.getInputFile()).isNotNull(); + assertThat(ParquetMetadata.toJSON(footer)).doesNotContain("inputFile"); + } + + /** + * A split carrying a footer is read without going back to the file for it: the record reader returns the same + * records as one which reads the footer itself, without ever touching the tail of the file where the footer + * length and the magic live. + */ + @Test + public void testSplitFooterIsReused() throws Exception { + CountingInputFile footerReadFile = new CountingInputFile(HadoopInputFile.fromPath(file, new Configuration())); + ParquetMetadata footer; + try (ParquetFileReader reader = ParquetFileReader.open(footerReadFile)) { + footer = reader.getFooter(); + } + assertThat(footerReadFile.tailBytesRead()) + .as("bytes read from the tail of the file while reading the footer") + .isPositive(); + + CountingInputFile countingFile = new CountingInputFile(HadoopInputFile.fromPath(file, new Configuration())); + List baseline = readSplit(newSplit(file), null, null); + List reusing = readSplit(newSplit(file), countingFile, footer); + + assertThat(reusing).hasSize(DATA.size()).isEqualTo(baseline); + assertThat(countingFile.bytesRead()) + .as("bytes read through the file supplied with the footer") + .isPositive(); + assertThat(countingFile.tailBytesRead()) + .as("bytes read from the last %d bytes of the file, which only a footer read touches", TAIL_SIZE) + .isZero(); + } + + /** + * The file recorded in the footer is the one opened, so a split whose own path no longer resolves is still + * readable when it carries a footer. + */ + @Test + public void testFooterInputFileIsUsedOverSplitPath() throws Exception { + InputFile inputFile = HadoopInputFile.fromPath(file, new Configuration()); + ParquetMetadata footer; + try (ParquetFileReader reader = ParquetFileReader.open(inputFile)) { + footer = reader.getFooter(); + } + + Path missing = new Path(file.getParent(), "no-such-file.parquet"); + List records = readSplit(newSplit(missing), inputFile, footer); + + assertThat(records).hasSize(DATA.size()); + } + + /** + * A split carrying a footer serializes to json without it: the footer is state of the JVM which built the split, + * not part of its description. + */ + @Test + public void testFooterNotSerializedToJson() throws Exception { + ParquetInputSplit split = newSplit(file); + split.setFooter(readFooter()); + + ObjectMapper mapper = new ObjectMapper(); + mapper.setVisibility(PropertyAccessor.ALL, Visibility.NONE); + mapper.setVisibility(PropertyAccessor.FIELD, Visibility.ANY); + String json = mapper.writeValueAsString(split); + + assertThat(json).contains("rowGroupOffsets").doesNotContain("footer"); + } + + /** + * A split carrying a footer survives a Writable round trip with everything else intact, but the footer + * itself does not cross it: a split deserialized in a task has to read the footer as it always did. + */ + @Test + public void testFooterNotSerializedThroughWritable() throws Exception { + ParquetInputSplit split = newSplit(file); + split.setFooter(readFooter()); + + ByteArrayOutputStream bytes = new ByteArrayOutputStream(); + try (DataOutputStream out = new DataOutputStream(bytes)) { + split.write(out); + } + ParquetInputSplit read = new ParquetInputSplit(); + try (DataInputStream in = new DataInputStream(new ByteArrayInputStream(bytes.toByteArray()))) { + read.readFields(in); + } + + assertThat(read.getFooter()) + .as("footer of a split read back from its serialized form") + .isNull(); + assertThat(read.getPath()).isEqualTo(split.getPath()); + assertThat(read.getStart()).isEqualTo(split.getStart()); + assertThat(read.getEnd()).isEqualTo(split.getEnd()); + assertThat(read.getLength()).isEqualTo(split.getLength()); + } + + private static ParquetMetadata readFooter() throws IOException { + try (ParquetFileReader reader = ParquetFileReader.open(HadoopInputFile.fromPath(file, new Configuration()))) { + return reader.getFooter(); + } + } + + private static ParquetInputSplit newSplit(Path path) { + return new ParquetInputSplit(path, 0, fileLength, fileLength, new String[0], null); + } + + /** + * Read a split through a {@link ParquetRecordReader}, with the footer and the file it was read from attached to + * the split when a footer is supplied. + */ + private static List readSplit(ParquetInputSplit split, InputFile inputFile, ParquetMetadata footer) + throws IOException, InterruptedException { + if (footer != null) { + footer.setInputFile(inputFile); + split.setFooter(footer); + } + + Configuration conf = new Configuration(); + conf.set(READ_SUPPORT_CLASS, GroupReadSupport.class.getName()); + TaskAttemptContext taskContext = + ContextUtil.newTaskAttemptContext(conf, TaskAttemptID.forName("attempt_0_1_m_1_1")); + + List records = new ArrayList<>(); + ParquetRecordReader reader = new ParquetRecordReader<>(new GroupReadSupport()); + try { + reader.initialize(split, taskContext); + while (reader.nextKeyValue()) { + records.add(reader.getCurrentValue().toString()); + } + } finally { + reader.close(); + } + return records; + } + + /** + * An {@link InputFile} which counts the bytes read through the streams it hands out, separately counting those + * read from the tail of the file: the footer length and the trailing magic. + */ + private static final class CountingInputFile implements InputFile { + + private final InputFile wrapped; + private final long tailStart; + private final AtomicLong bytesRead = new AtomicLong(); + private final AtomicLong tailBytesRead = new AtomicLong(); + + public long bytesRead() { + return bytesRead.get(); + } + + public long tailBytesRead() { + return tailBytesRead.get(); + } + + private CountingInputFile(InputFile wrapped) throws IOException { + this.wrapped = wrapped; + this.tailStart = wrapped.getLength() - TAIL_SIZE; + } + + @Override + public long getLength() throws IOException { + return wrapped.getLength(); + } + + @Override + public SeekableInputStream newStream() throws IOException { + return new CountingInputStream(wrapped.newStream()); + } + + private final class CountingInputStream extends SeekableInputStream { + + private final SeekableInputStream in; + + private CountingInputStream(SeekableInputStream in) { + this.in = in; + } + + /** + * Count a read of {@code read} bytes which started at {@code start}. + */ + private int counted(long start, int read) { + if (read > 0) { + bytesRead.addAndGet(read); + tailBytesRead.addAndGet(Math.max(0, start + read - Math.max(start, tailStart))); + } + return read; + } + + @Override + public long getPos() throws IOException { + return in.getPos(); + } + + @Override + public void seek(long newPos) throws IOException { + in.seek(newPos); + } + + @Override + public void readFully(byte[] bytes) throws IOException { + long start = in.getPos(); + in.readFully(bytes); + counted(start, bytes.length); + } + + @Override + public void readFully(byte[] bytes, int start, int len) throws IOException { + long pos = in.getPos(); + in.readFully(bytes, start, len); + counted(pos, len); + } + + @Override + public int read(ByteBuffer buf) throws IOException { + long start = in.getPos(); + return counted(start, in.read(buf)); + } + + @Override + public void readFully(ByteBuffer buf) throws IOException { + long start = in.getPos(); + int remaining = buf.remaining(); + in.readFully(buf); + counted(start, remaining); + } + + @Override + public int read() throws IOException { + long start = in.getPos(); + int read = in.read(); + if (read >= 0) { + counted(start, 1); + } + return read; + } + + @Override + public int read(byte[] bytes, int off, int len) throws IOException { + long start = in.getPos(); + return counted(start, in.read(bytes, off, len)); + } + + @Override + public void close() throws IOException { + in.close(); + } + } + } +}