Skip to content

GH-3952: Read each page header only once in the CLI pages command #3953

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
PDGGK wants to merge 1 commit into
base: master
Choose a base branch
from
+113 −21
Open

GH-3952: Read each page header only once in the CLI pages command #3953

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
48 changes: 27 additions & 21 deletions parquet-cli/src/main/java/org/apache/parquet/cli/commands/ShowPagesCommand.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<ColumnDescriptor, PrimitiveType> columns = new LinkedHashMap<>();
if (this.columns == null || this.columns.isEmpty()) {
Expand All @@ -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<String, List<String>> formatted = new LinkedHashMap<>();
PageFormatter formatter = new PageFormatter(reader);
PageFormatter formatter = new PageFormatter(reader, input);
PageReadStore pageStore;
int rowGroupNum = 0;
while ((pageStore = reader.readNextRowGroup()) != null) {
Expand Down Expand Up @@ -163,14 +165,19 @@ public List<String> getExamples() {

private class PageFormatter implements DataPage.Visitor<String> {
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<Long> compressedPageSizes = new ArrayList<>();
private long columnChunkEndPos;

PageFormatter(ParquetFileReader reader) {
PageFormatter(ParquetFileReader reader, SeekableInputStream input) {
this.reader = reader;
this.input = input;
}

String getHeader() {
Expand All @@ -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) {
Expand All @@ -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);
}
Expand Down
86 changes: 86 additions & 0 deletions parquet-cli/src/test/java/org/apache/parquet/cli/commands/ShowPagesCommandTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -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<Group> 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<String> 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<String> 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();
}
}
}