diff --git a/parquet-cli/src/main/java/org/apache/parquet/cli/commands/ShowPagesCommand.java b/parquet-cli/src/main/java/org/apache/parquet/cli/commands/ShowPagesCommand.java index 5d8bf4d94a..9a915c1e74 100644 --- a/parquet-cli/src/main/java/org/apache/parquet/cli/commands/ShowPagesCommand.java +++ b/parquet-cli/src/main/java/org/apache/parquet/cli/commands/ShowPagesCommand.java @@ -97,7 +97,9 @@ public int run() throws IOException { } String source = targets.get(0); - try (ParquetFileReader reader = createParquetFileReader(source)) { + try (ParquetFileReader reader = createParquetFileReader(source); + SeekableInputStream input = HadoopInputFile.fromPath(new Path(reader.getFile()), getConf()) + .newStream()) { MessageType schema = reader.getFileMetaData().getSchema(); Map columns = new LinkedHashMap<>(); if (this.columns == null || this.columns.isEmpty()) { @@ -114,7 +116,7 @@ public int run() throws IOException { reader.getRowGroups().get(0).getColumns().get(0).getCodec(); // accumulate formatted lines to print by column Map> formatted = new LinkedHashMap<>(); - PageFormatter formatter = new PageFormatter(reader); + PageFormatter formatter = new PageFormatter(reader, input); PageReadStore pageStore; int rowGroupNum = 0; while ((pageStore = reader.readNextRowGroup()) != null) { @@ -163,14 +165,19 @@ public List getExamples() { private class PageFormatter implements DataPage.Visitor { private final ParquetFileReader reader; + private final SeekableInputStream input; private int rowGroupNum; private int pageNum; private PrimitiveType type; private String shortCodec; private ColumnDescriptor currentColumn; + // compressed sizes of the page headers read so far in the current column chunk + private final List compressedPageSizes = new ArrayList<>(); + private long columnChunkEndPos; - PageFormatter(ParquetFileReader reader) { + PageFormatter(ParquetFileReader reader, SeekableInputStream input) { this.reader = reader; + this.input = input; } String getHeader() { @@ -185,6 +192,8 @@ void setContext(int rowGroupNum, PrimitiveType type, CompressionCodecName codec, this.type = type; this.shortCodec = shortCodec(codec); this.currentColumn = column; + this.compressedPageSizes.clear(); + this.columnChunkEndPos = -1; } String format(Page page) { @@ -199,29 +208,26 @@ String format(Page page) { } private long getPageCompressedSize() { - long compressedSize = 0; - try (SeekableInputStream input = HadoopInputFile.fromPath(new Path(reader.getFile()), getConf()) - .newStream()) { - BlockMetaData rowGroup = reader.getRowGroups().get(rowGroupNum); - int columnIndex = - reader.getFileMetaData().getSchema().getColumns().indexOf(currentColumn); - ColumnChunkMetaData columnChunk = rowGroup.getColumns().get(columnIndex); + try { + if (columnChunkEndPos < 0) { + BlockMetaData rowGroup = reader.getRowGroups().get(rowGroupNum); + int columnIndex = + reader.getFileMetaData().getSchema().getColumns().indexOf(currentColumn); + ColumnChunkMetaData columnChunk = rowGroup.getColumns().get(columnIndex); - long startOffset = columnChunk.getStartingPos(); - input.seek(startOffset); - long endPos = startOffset + columnChunk.getTotalSize(); - int currentPageIndex = 0; + long startOffset = columnChunk.getStartingPos(); + input.seek(startOffset); + columnChunkEndPos = startOffset + columnChunk.getTotalSize(); + } - while (input.getPos() < endPos) { + // Pages are formatted in file order, so continue from the last header read instead of + // reading all the headers of the column chunk again for every page. + while (compressedPageSizes.size() <= pageNum && input.getPos() < columnChunkEndPos) { PageHeader pageHeader = Util.readPageHeader(input); - if (currentPageIndex == pageNum) { - compressedSize = pageHeader.getCompressed_page_size(); - break; - } + compressedPageSizes.add((long) pageHeader.getCompressed_page_size()); input.skip(pageHeader.getCompressed_page_size()); - currentPageIndex++; } - return compressedSize; + return pageNum < compressedPageSizes.size() ? compressedPageSizes.get(pageNum) : 0; } catch (IOException e) { throw new RuntimeException(e); } diff --git a/parquet-cli/src/test/java/org/apache/parquet/cli/commands/ShowPagesCommandTest.java b/parquet-cli/src/test/java/org/apache/parquet/cli/commands/ShowPagesCommandTest.java index 61fbebcde7..52e428e3fe 100644 --- a/parquet-cli/src/test/java/org/apache/parquet/cli/commands/ShowPagesCommandTest.java +++ b/parquet-cli/src/test/java/org/apache/parquet/cli/commands/ShowPagesCommandTest.java @@ -23,15 +23,20 @@ import static org.apache.parquet.column.Encoding.RLE; import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.BINARY; import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.INT32; +import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.INT64; import static org.assertj.core.api.Assertions.assertThat; import java.io.File; import java.io.IOException; +import java.net.URI; +import java.util.ArrayList; import java.util.Arrays; import java.util.List; import java.util.Map; import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; +import org.apache.hadoop.fs.RawLocalFileSystem; import org.apache.parquet.bytes.BytesInput; import org.apache.parquet.column.ParquetProperties.WriterVersion; import org.apache.parquet.column.page.DictionaryPage; @@ -47,6 +52,7 @@ import org.apache.parquet.hadoop.ParquetWriter; import org.apache.parquet.hadoop.example.ExampleParquetWriter; import org.apache.parquet.hadoop.example.GroupReadSupport; +import org.apache.parquet.hadoop.metadata.BlockMetaData; import org.apache.parquet.hadoop.metadata.ColumnChunkMetaData; import org.apache.parquet.hadoop.metadata.CompressionCodecName; import org.apache.parquet.hadoop.util.HadoopInputFile; @@ -215,4 +221,84 @@ public void testPageSizesMatchHeaders(CompressionCodecName codec, WriterVersion } }); } + + @Test + public void testReadsEachPageHeaderOnce() throws IOException { + Configuration conf = new Configuration(); + conf.setClass("fs.counting.impl", CountingFileSystem.class, FileSystem.class); + Path path = new Path("counting", null, randomParquetFile().getAbsolutePath()); + MessageType schema = Types.buildMessage() + .required(INT64) + .named("id") + .required(BINARY) + .named("name") + .named("schema"); + SimpleGroupFactory groups = new SimpleGroupFactory(schema); + try (ParquetWriter writer = ExampleParquetWriter.builder(path) + .withConf(conf) + .withType(schema) + .withPageRowCountLimit(10) + .withRowGroupSize(4096) + .build()) { + for (int i = 0; i < 1000; i++) { + writer.write(groups.newGroup().append("id", (long) i).append("name", "name-" + (i % 37))); + } + } + + List expected = new ArrayList<>(); + HadoopInputFile file = HadoopInputFile.fromPath(path, conf); + try (ParquetFileReader reader = ParquetFileReader.open(file); + SeekableInputStream input = file.newStream()) { + assertThat(reader.getRowGroups()).hasSizeGreaterThan(1); + for (int column = 0; column < schema.getFieldCount(); column++) { + for (BlockMetaData rowGroup : reader.getRowGroups()) { + ColumnChunkMetaData chunk = rowGroup.getColumns().get(column); + input.seek(chunk.getStartingPos()); + while (input.getPos() < chunk.getStartingPos() + chunk.getTotalSize()) { + PageHeader header = Util.readPageHeader(input); + int count = header.getType() == PageType.DICTIONARY_PAGE + ? header.getDictionary_page_header().getNum_values() + : header.getData_page_header().getNum_values(); + long size = header.getCompressed_page_size(); + expected.add(String.format( + "%-7d %-10s %-10s", count, humanReadable((float) size / count), humanReadable(size))); + input.seek(input.getPos() + size); + } + } + } + } + assertThat(expected).hasSizeGreaterThan(100); + + CountingFileSystem fs = (CountingFileSystem) path.getFileSystem(conf); + long fileLength = fs.getFileStatus(path).getLen(); + long bytesReadBefore = fs.getBytesRead(); + withLogger((console, events) -> { + ShowPagesCommand command = new ShowPagesCommand(console); + command.targets = List.of(path.toString()); + command.setConf(conf); + assertThat(command.run()).isZero(); + List pageLines = events.stream() + .map(LoggingEvent::getMessage) + .filter(line -> line.matches("\\s*\\d+-.*")) + .toList(); + assertThat(pageLines).hasSameSizeAs(expected); + for (int i = 0; i < pageLines.size(); i++) { + assertThat(pageLines.get(i)).contains(expected.get(i)); + } + }); + // the pages and their headers are each read once, not the headers again for every page + assertThat(fs.getBytesRead() - bytesReadBefore).isLessThan(3 * fileLength); + } + + /** A local file system under its own scheme, so that its read statistics only count this test. */ + public static class CountingFileSystem extends RawLocalFileSystem { + @Override + public URI getUri() { + return URI.create("counting:///"); + } + + long getBytesRead() { + return statistics.getBytesRead(); + } + } }