wgtmac commented on code in PR #1293:
URL: https://github.com/apache/parquet-mr/pull/1293#discussion_r1519832629
##########
parquet-column/src/main/java/org/apache/parquet/column/impl/ColumnReaderBase.java:
##########
@@ -799,7 +831,11 @@ boolean isPageFullyConsumed() {
*/
@Override
public void consume() {
- checkRead();
+ if (pageReader.isEager()) {
+ consumeAllPages();
Review Comment:
ditto, it seems that we do not need any change here.
##########
parquet-column/src/test/java/org/apache/parquet/column/page/mem/MemPageReader.java:
##########
@@ -34,18 +34,30 @@ public class MemPageReader implements PageReader {
private final Iterator<DataPage> pages;
private final DictionaryPage dictionaryPage;
- public MemPageReader(long totalValueCount, Iterator<DataPage> pages,
DictionaryPage dictionaryPage) {
+ private final boolean isEager;
+
+ public MemPageReader(
+ long totalValueCount, Iterator<DataPage> pages, DictionaryPage
dictionaryPage, boolean isEager) {
super();
this.pages = Objects.requireNonNull(pages, "pages cannot be null");
this.totalValueCount = totalValueCount;
this.dictionaryPage = dictionaryPage;
+ this.isEager = isEager;
}
@Override
public long getTotalValueCount() {
+ if (!isEager && pages.hasNext()) {
+ throw new IllegalStateException("Can't return totalValueCount until lazy
iterator has been exhausted");
+ }
return totalValueCount;
}
+ @Override
+ public boolean isEager() {
+ return isEager || !pages.hasNext();
Review Comment:
Why should we consider `pages.hasNext()` here?
##########
parquet-column/src/main/java/org/apache/parquet/column/impl/ColumnReaderBase.java:
##########
@@ -465,14 +464,13 @@ void writeValue() {
} else {
this.dictionary = null;
}
- this.totalValueCount = pageReader.getTotalValueCount();
- if (totalValueCount <= 0) {
- throw new ParquetDecodingException("totalValueCount '" + totalValueCount
+ "' <= 0");
+ if (pageReader.isEager() && pageReader.getTotalValueCount() <= 0) {
+ throw new ParquetDecodingException("totalValueCount '" +
pageReader.getTotalValueCount() + "' <= 0");
}
}
boolean isFullyConsumed() {
- return readValues >= totalValueCount;
+ return pageReader.isEager() && readValues >=
pageReader.getTotalValueCount();
Review Comment:
Is it possible not to change this? `totalValueCount` is accessible from the
column chunk metadata, so the page reader should be easy to obtain this value
whether it is eager or not.
##########
parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java:
##########
@@ -1274,7 +1276,7 @@ private void readChunkPages(Chunk chunk, BlockMetaData
block, ColumnChunkPageRea
} else { // encrypted column
rowGroup.addColumn(
chunk.descriptor.col,
- chunk.readAllPages(
+ chunk.readAllPages( // @Todo this must be made lazy too?
Review Comment:
> ahh I hadn't seen that PR, looks really similar. It's still open but
hasn't been updated in a year -- is there any plan to merge it?
IMO, https://github.com/apache/parquet-mr/pull/1139 is more promising and
probably you want to take a look.
##########
parquet-column/src/main/java/org/apache/parquet/column/impl/ColumnReaderBase.java:
##########
@@ -647,7 +647,10 @@ public int getCurrentDefinitionLevel() {
return definitionLevel;
}
- private void checkRead() {
+ private int skipValues = 0;
+
+ /** Reads all pages. */
+ private void consumeAllPages() {
Review Comment:
IIUC, the original `checkRead()` does not read all the pages. It just stops
at a non-skippable value.
##########
parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java:
##########
@@ -2034,58 +2251,15 @@ public void addChunk(ChunkDescriptor descriptor) {
}
/**
- * @param f file to read the chunks from
+ * @param fileStream file to read the chunks from
* @param builder used to build chunk list to read the pages for the
different columns
* @throws IOException if there is an error while reading from the stream
*/
- public void readAll(SeekableInputStream f, ChunkListBuilder builder)
throws IOException {
- f.seek(offset);
-
- int fullAllocations = Math.toIntExact(length /
options.getMaxAllocationSize());
- int lastAllocationSize = Math.toIntExact(length %
options.getMaxAllocationSize());
-
- int numAllocations = fullAllocations + (lastAllocationSize > 0 ? 1 : 0);
- List<ByteBuffer> buffers = new ArrayList<>(numAllocations);
-
- for (int i = 0; i < fullAllocations; i += 1) {
-
buffers.add(options.getAllocator().allocate(options.getMaxAllocationSize()));
- }
-
- if (lastAllocationSize > 0) {
- buffers.add(options.getAllocator().allocate(lastAllocationSize));
- }
- builder.addBuffersToRelease(buffers);
-
- long readStart = System.nanoTime();
- for (ByteBuffer buffer : buffers) {
- f.readFully(buffer);
- buffer.flip();
- }
- setReadMetrics(readStart);
-
- // report in a counter the data we just scanned
- BenchmarkCounter.incrementBytesRead(length);
- ByteBufferInputStream stream = ByteBufferInputStream.wrap(buffers);
- for (final ChunkDescriptor descriptor : chunks) {
- builder.add(descriptor, stream.sliceBuffers(descriptor.size), f);
- }
- }
-
- private void setReadMetrics(long startNs) {
- ParquetMetricsCallback metricsCallback = options.getMetricsCallback();
- if (metricsCallback != null) {
- long totalFileReadTimeNs = Math.max(System.nanoTime() - startNs, 0);
- double sizeInMb = ((double) length) / (1024 * 1024);
- double timeInSec = ((double) totalFileReadTimeNs) / 1000_0000_0000L;
- double throughput = sizeInMb / timeInSec;
- LOG.debug(
- "Parquet: File Read stats: Length: {} MB, Time: {} secs,
throughput: {} MB/sec ",
- sizeInMb,
- timeInSec,
- throughput);
- metricsCallback.setDuration(ParquetFileReaderMetrics.ReadTime.name(),
totalFileReadTimeNs);
- metricsCallback.setValueLong(ParquetFileReaderMetrics.ReadSize.name(),
length);
-
metricsCallback.setValueDouble(ParquetFileReaderMetrics.ReadThroughput.name(),
throughput);
+ public void readAll(ParquetFileStream fileStream, ChunkListBuilder
builder) throws IOException {
+ if (options.columnChunkBufferSize() <= 0) {
+ fileStream.createEagerChunkStream(builder, offset, length);
+ } else {
+ fileStream.createLazyChunkStream(builder,
options.columnChunkBufferSize());
Review Comment:
Any chance to incorporate https://github.com/apache/parquet-mr/pull/1139?
##########
parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java:
##########
@@ -1724,160 +1975,125 @@ public ColumnChunkPageReader readAllPages(
int rowGroupOrdinal,
int columnOrdinal)
throws IOException {
- List<DataPage> pagesInChunk = new ArrayList<>();
- DictionaryPage dictionaryPage = null;
PrimitiveType type = getFileMetaData()
.getSchema()
.getType(descriptor.col.getPath())
.asPrimitiveType();
- long valuesCountReadSoFar = 0L;
- int dataPageCountReadSoFar = 0;
- byte[] dataPageHeaderAAD = null;
+
+ byte[] moduleAAD = null;
if (null != headerBlockDecryptor) {
- dataPageHeaderAAD = AesCipher.createModuleAAD(
- aadPrefix,
- ModuleType.DataPageHeader,
- rowGroupOrdinal,
- columnOrdinal,
- getPageOrdinal(dataPageCountReadSoFar));
+ moduleAAD = AesCipher.createModuleAAD(
+ aadPrefix, ModuleType.DataPageHeader, rowGroupOrdinal,
columnOrdinal, getPageOrdinal(0));
}
- while (hasMorePages(valuesCountReadSoFar, dataPageCountReadSoFar)) {
- byte[] pageHeaderAAD = dataPageHeaderAAD;
- if (null != headerBlockDecryptor) {
- // Important: this verifies file integrity (makes sure dictionary
page had not been removed)
- if (null == dictionaryPage &&
descriptor.metadata.hasDictionaryPage()) {
- pageHeaderAAD = AesCipher.createModuleAAD(
- aadPrefix, ModuleType.DictionaryPageHeader, rowGroupOrdinal,
columnOrdinal, -1);
- } else {
- int pageOrdinal = getPageOrdinal(dataPageCountReadSoFar);
- AesCipher.quickUpdatePageAAD(dataPageHeaderAAD, pageOrdinal);
+ final Chunk chunk = this;
+
+ final PageIterator iterator = new PageIterator(moduleAAD) {
+ private long valuesCountReadSoFar = 0L;
+ private int dataPageCountReadSoFar = 0;
+
+ private DictionaryPage dictionaryPage;
+ private PageWithHeader nextPage = null;
+ private boolean bufferedToFirstDataPage = false;
+
+ @Override
+ DictionaryPage getDictionaryPage() {
+ if (!bufferedToFirstDataPage) {
+ bufferNextDataPage();
}
+ return dictionaryPage;
}
- PageHeader pageHeader = readPageHeader(headerBlockDecryptor,
pageHeaderAAD);
- int uncompressedPageSize = pageHeader.getUncompressed_page_size();
- int compressedPageSize = pageHeader.getCompressed_page_size();
- final BytesInput pageBytes;
- switch (pageHeader.type) {
- case DICTIONARY_PAGE:
- // there is only one dictionary page per column chunk
- if (dictionaryPage != null) {
- throw new ParquetDecodingException(
- "more than one dictionary page in column " + descriptor.col);
- }
- pageBytes = this.readAsBytesInput(compressedPageSize);
- if (options.usePageChecksumVerification() &&
pageHeader.isSetCrc()) {
- verifyCrc(
- pageHeader.getCrc(),
- pageBytes,
- "could not verify dictionary page integrity, CRC checksum
verification failed");
- }
- DictionaryPageHeader dicHeader =
pageHeader.getDictionary_page_header();
- dictionaryPage = new DictionaryPage(
- pageBytes,
- uncompressedPageSize,
- dicHeader.getNum_values(),
- converter.getEncoding(dicHeader.getEncoding()));
- // Copy crc to new page, used for testing
- if (pageHeader.isSetCrc()) {
- dictionaryPage.setCrc(pageHeader.getCrc());
- }
- break;
- case DATA_PAGE:
- DataPageHeader dataHeaderV1 = pageHeader.getData_page_header();
- pageBytes = this.readAsBytesInput(compressedPageSize);
- if (options.usePageChecksumVerification() &&
pageHeader.isSetCrc()) {
- verifyCrc(
- pageHeader.getCrc(),
- pageBytes,
- "could not verify page integrity, CRC checksum verification
failed");
- }
- DataPageV1 dataPageV1 = new DataPageV1(
- pageBytes,
- dataHeaderV1.getNum_values(),
- uncompressedPageSize,
- converter.fromParquetStatistics(
- getFileMetaData().getCreatedBy(),
dataHeaderV1.getStatistics(), type),
-
converter.getEncoding(dataHeaderV1.getRepetition_level_encoding()),
-
converter.getEncoding(dataHeaderV1.getDefinition_level_encoding()),
- converter.getEncoding(dataHeaderV1.getEncoding()));
- // Copy crc to new page, used for testing
- if (pageHeader.isSetCrc()) {
- dataPageV1.setCrc(pageHeader.getCrc());
- }
- pagesInChunk.add(dataPageV1);
- valuesCountReadSoFar += dataHeaderV1.getNum_values();
- ++dataPageCountReadSoFar;
- break;
- case DATA_PAGE_V2:
- DataPageHeaderV2 dataHeaderV2 =
pageHeader.getData_page_header_v2();
- int dataSize = compressedPageSize
- - dataHeaderV2.getRepetition_levels_byte_length()
- - dataHeaderV2.getDefinition_levels_byte_length();
- final BytesInput repetitionLevels =
-
this.readAsBytesInput(dataHeaderV2.getRepetition_levels_byte_length());
- final BytesInput definitionLevels =
-
this.readAsBytesInput(dataHeaderV2.getDefinition_levels_byte_length());
- final BytesInput values = this.readAsBytesInput(dataSize);
- if (options.usePageChecksumVerification() &&
pageHeader.isSetCrc()) {
- pageBytes = BytesInput.concat(repetitionLevels,
definitionLevels, values);
- verifyCrc(
- pageHeader.getCrc(),
- pageBytes,
- "could not verify page integrity, CRC checksum verification
failed");
+
+ @Override
+ public boolean hasNext() {
+ if (!bufferedToFirstDataPage) {
+ bufferNextDataPage();
+ }
+ return nextPage != null;
+ }
+
+ private boolean hasMorePages(long valuesCountReadSoFar, int
dataPageCountReadSoFar) {
+ return offsetIndex == null
+ ? valuesCountReadSoFar < descriptor.metadata.getValueCount()
+ : dataPageCountReadSoFar < offsetIndex.getPageCount();
+ }
+
+ private void bufferNextDataPage() {
+ bufferedToFirstDataPage = true;
+ while (true) {
+ if (!hasMorePages(valuesCountReadSoFar, dataPageCountReadSoFar)) {
+ return;
}
- DataPageV2 dataPageV2 = new DataPageV2(
- dataHeaderV2.getNum_rows(),
- dataHeaderV2.getNum_nulls(),
- dataHeaderV2.getNum_values(),
- repetitionLevels,
- definitionLevels,
- converter.getEncoding(dataHeaderV2.getEncoding()),
- values,
- uncompressedPageSize,
- converter.fromParquetStatistics(
- getFileMetaData().getCreatedBy(),
dataHeaderV2.getStatistics(), type),
- dataHeaderV2.isIs_compressed());
- // Copy crc to new page, used for testing
- if (pageHeader.isSetCrc()) {
- dataPageV2.setCrc(pageHeader.getCrc());
+ try {
+ byte[] pageHeaderAAD = dataPageHeaderAAD;
+ if (null != headerBlockDecryptor) {
+ // Important: this verifies file integrity (makes sure
dictionary page had not been
+ // removed)
+ if (null == dictionaryPage &&
descriptor.metadata.hasDictionaryPage()) {
+ pageHeaderAAD = AesCipher.createModuleAAD(
+ aadPrefix,
+ ModuleType.DictionaryPageHeader,
+ rowGroupOrdinal,
+ columnOrdinal,
+ -1);
+ } else {
+ int pageOrdinal = getPageOrdinal(dataPageCountReadSoFar);
+ AesCipher.quickUpdatePageAAD(dataPageHeaderAAD, pageOrdinal);
+ }
+ }
+
+ PageWithHeader pageWithHeader =
+ chunkStream.readNextPage(chunk, type, headerBlockDecryptor,
pageHeaderAAD);
+ if (pageWithHeader == null) {
+ return;
+ }
+
+ if (pageWithHeader.header.type == PageType.DICTIONARY_PAGE) {
Review Comment:
It seems that the header member variable in the new `PageWithHeader` class
is only used here for checking page type. Could we get it from the page class?
Perhaps add a Page.type() method?
##########
parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ColumnChunkPageReadStore.java:
##########
@@ -81,25 +80,36 @@ static final class ColumnChunkPageReader implements
PageReader {
private final byte[] dictionaryPageAAD;
private final ByteBufferReleaser releaser;
+ private final boolean isEager;
+ private long valueCount;
+
ColumnChunkPageReader(
BytesInputDecompressor decompressor,
- List<DataPage> compressedPages,
+ Iterator<DataPage> compressedPages,
DictionaryPage compressedDictionaryPage,
OffsetIndex offsetIndex,
long rowCount,
BlockCipher.Decryptor blockDecryptor,
byte[] fileAAD,
int rowGroupOrdinal,
int columnOrdinal,
- ParquetReadOptions options) {
+ ParquetReadOptions options,
+ boolean isEager) {
this.decompressor = decompressor;
- this.compressedPages = new ArrayDeque<DataPage>(compressedPages);
+ this.isEager = isEager;
this.compressedDictionaryPage = compressedDictionaryPage;
- long count = 0;
- for (DataPage p : compressedPages) {
- count += p.getValueCount();
+ this.valueCount = 0;
+ if (isEager) {
+ final List<DataPage> materializedPages = new ArrayList<>();
+ for (Iterator<DataPage> it = compressedPages; it.hasNext(); ) {
+ final DataPage next = it.next();
+ this.valueCount += next.getValueCount();
Review Comment:
Can we get this value from
https://github.com/apache/parquet-format/blob/master/src/main/thrift/parquet.thrift#L786
##########
parquet-column/src/main/java/org/apache/parquet/column/impl/ColumnReaderBase.java:
##########
@@ -675,6 +678,35 @@ private void checkRead() {
definitionLevel = dl;
}
+ /**
+ * Reads the next page.
+ */
+ private void consumePage() {
+ int rl, dl;
+ if (isPageFullyConsumed()) {
+ skipValues = 0;
+ if (isFullyConsumed()) {
+ LOG.debug("end reached");
+ repetitionLevel = 0; // the next repetition level
+ return;
+ }
+ readPage();
+ }
+ rl = repetitionLevelColumn.nextInt();
+ dl = definitionLevelColumn.nextInt();
+ ++readValues;
+
+ if (skipRL(rl)) {
+ if (dl == maxDefinitionLevel) {
+ ++skipValues;
+ }
+ } else {
+ repetitionLevel = rl;
+ definitionLevel = dl;
+ binding.skip(skipValues);
Review Comment:
Based on my previous comment, `skipValues` should always be zero here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
