From 4aa365d50890f45bc63417c5e921337c0289c660 Mon Sep 17 00:00:00 2001 From: Zihan Dai Date: Wed, 11 Nov 2026 13:53:35 +1100 Subject: [PATCH] GH-3952: Read each page header only once in the CLI pages command ShowPagesCommand looked up the compressed size of every page by opening a new stream on the file and reading the page headers of the column chunk from its start up to that page. A column chunk with N pages therefore cost N stream opens and N(N+1)/2 header reads, so the command slowed down quadratically with the number of pages per column chunk. Keep one stream open for the command and read the headers of the current column chunk once, in order, as its pages are formatted. The printed sizes are unchanged. --- .../cli/commands/ShowPagesCommand.java | 48 ++++++----- .../cli/commands/ShowPagesCommandTest.java | 86 +++++++++++++++++++ 2 files changed, 113 insertions(+), 21 deletions(-) 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(); + } + } }