Skip to content

Parquet: Use native getRowIndexOffset support instead of calculating it #11520

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

Merged
flyrain merged 8 commits into from
Nov 21, 2024
Merged

Parquet: Use native getRowIndexOffset support instead of calculating it #11520

Show file tree
Hide file tree
Changes from 6 commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
f435ca1
Parquet: Remove the row position since parquet row group has it natively
flyrain Oct 25, 2022
ffa7e61
Change ColumnarBatchReader for Spark 3.4 and 3.5
wypoon Apr 6, 2024
832ee7a
Retain old methods but deprecate them, to avoid breaking the API.
wypoon Apr 9, 2024
73d1380
Fix inadvertent renaming of a parameter.
wypoon Apr 9, 2024
00a4b3c
Fix inadvertent dropping of a final method modifier.
wypoon Apr 9, 2024
ff29af6
Update deprecation comments.
wypoon Nov 12, 2024
d189412
Throw an exception if PageReadStore does not have the row index offset.
wypoon Nov 14, 2024
661b264
Address nit.
wypoon Nov 21, 2024
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
8 changes: 7 additions & 1 deletion arrow/src/main/java/org/apache/iceberg/arrow/vectorized/BaseBatchReader.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,9 +42,15 @@ protected BaseBatchReader(List<VectorizedReader<?>> readers) {
@Override
public void setRowGroupInfo(
PageReadStore pageStore, Map<ColumnPath, ColumnChunkMetaData> metaData, long rowPosition) {
setRowGroupInfo(pageStore, metaData);
}

@Override
public void setRowGroupInfo(
PageReadStore pageStore, Map<ColumnPath, ColumnChunkMetaData> metaData) {
for (VectorizedArrowReader reader : readers) {
if (reader != null) {
reader.setRowGroupInfo(pageStore, metaData, rowPosition);
reader.setRowGroupInfo(pageStore, metaData);
}
}
}
Expand Down
25 changes: 24 additions & 1 deletion arrow/src/main/java/org/apache/iceberg/arrow/vectorized/VectorizedArrowReader.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -432,6 +432,11 @@ private void allocateVectorBasedOnTypeName(PrimitiveType primitive, Field arrowF
@Override
public void setRowGroupInfo(
PageReadStore source, Map<ColumnPath, ColumnChunkMetaData> metadata, long rowPosition) {
setRowGroupInfo(source, metadata);
}

@Override
public void setRowGroupInfo(PageReadStore source, Map<ColumnPath, ColumnChunkMetaData> metadata) {
ColumnChunkMetaData chunkMetaData = metadata.get(ColumnPath.get(columnDescriptor.getPath()));
this.dictionary =
vectorizedColumnIterator.setRowGroupInfo(
Expand Down Expand Up @@ -475,6 +480,10 @@ public VectorHolder read(VectorHolder reuse, int numValsToRead) {
public void setRowGroupInfo(
PageReadStore source, Map<ColumnPath, ColumnChunkMetaData> metadata, long rowPosition) {}

@Override
public void setRowGroupInfo(
PageReadStore source, Map<ColumnPath, ColumnChunkMetaData> metadata) {}

@Override
public String toString() {
return "NullReader";
Expand Down Expand Up @@ -541,7 +550,13 @@ private static NullabilityHolder newNullabilityHolder(int size) {
@Override
public void setRowGroupInfo(
PageReadStore source, Map<ColumnPath, ColumnChunkMetaData> metadata, long rowPosition) {
this.rowStart = rowPosition;
setRowGroupInfo(source, metadata);
}

@Override
public void setRowGroupInfo(
PageReadStore source, Map<ColumnPath, ColumnChunkMetaData> metadata) {
this.rowStart = source.getRowIndexOffset().orElse(0L);
}

@Override
Expand Down Expand Up @@ -586,6 +601,10 @@ public VectorHolder read(VectorHolder reuse, int numValsToRead) {
public void setRowGroupInfo(
PageReadStore source, Map<ColumnPath, ColumnChunkMetaData> metadata, long rowPosition) {}

@Override
public void setRowGroupInfo(
PageReadStore source, Map<ColumnPath, ColumnChunkMetaData> metadata) {}

@Override
public String toString() {
return String.format("ConstantReader: %s", value);
Expand Down Expand Up @@ -613,6 +632,10 @@ public VectorHolder read(VectorHolder reuse, int numValsToRead) {
public void setRowGroupInfo(
PageReadStore source, Map<ColumnPath, ColumnChunkMetaData> metadata, long rowPosition) {}

@Override
public void setRowGroupInfo(
PageReadStore source, Map<ColumnPath, ColumnChunkMetaData> metadata) {}

@Override
public String toString() {
return "DeletedVectorReader";
Expand Down
5 changes: 1 addition & 4 deletions parquet/src/main/java/org/apache/iceberg/parquet/ParquetReader.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -99,7 +99,6 @@ private static class FileIterator<T> implements CloseableIterator<T> {
private final ParquetValueReader<T> model;
private final long totalValues;
private final boolean reuseContainers;
private final long[] rowGroupsStartRowPos;

private int nextRowGroup = 0;
private long nextRowGroupStart = 0;
Expand All @@ -112,7 +111,6 @@ private static class FileIterator<T> implements CloseableIterator<T> {
this.model = conf.model();
this.totalValues = conf.totalValues();
this.reuseContainers = conf.reuseContainers();
this.rowGroupsStartRowPos = conf.startRowPositions();
}

@Override
Expand Down Expand Up @@ -149,11 +147,10 @@ private void advance() {
throw new RuntimeIOException(e);
}

long rowPosition = rowGroupsStartRowPos[nextRowGroup];
nextRowGroupStart += pages.getRowCount();
nextRowGroup += 1;

model.setPageSource(pages, rowPosition);
model.setPageSource(pages);
}

@Override
Expand Down
9 changes: 9 additions & 0 deletions parquet/src/main/java/org/apache/iceberg/parquet/ParquetValueReader.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,5 +28,14 @@ public interface ParquetValueReader<T> {

List<TripleIterator<?>> columns();

/**
* @deprecated since 1.8.0, will be removed in 1.9.0; use setPageSource(PageReadStore) instead.
Copy link
Member

@szehon-ho szehon-ho Nov 20, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nit: typically we use a link?

Copy link
Contributor Author

@wypoon wypoon Nov 21, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done.

*/
@Deprecated
void setPageSource(PageReadStore pageStore, long rowPosition);

default void setPageSource(PageReadStore pageStore) {
throw new UnsupportedOperationException(
this.getClass().getName() + " doesn't implement setPageSource(PageReadStore)");
}
}
48 changes: 42 additions & 6 deletions parquet/src/main/java/org/apache/iceberg/parquet/ParquetValueReaders.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -113,6 +113,9 @@ public List<TripleIterator<?>> columns() {

@Override
public void setPageSource(PageReadStore pageStore, long rowPosition) {}

@Override
public void setPageSource(PageReadStore pageStore) {}
}

static class ConstantReader<C> implements ParquetValueReader<C> {
Expand Down Expand Up @@ -176,6 +179,9 @@ public List<TripleIterator<?>> columns() {

@Override
public void setPageSource(PageReadStore pageStore, long rowPosition) {}

@Override
public void setPageSource(PageReadStore pageStore) {}
}

static class PositionReader implements ParquetValueReader<Long> {
Expand All @@ -200,7 +206,12 @@ public List<TripleIterator<?>> columns() {

@Override
public void setPageSource(PageReadStore pageStore, long rowPosition) {
this.rowGroupStart = rowPosition;
setPageSource(pageStore);
}

@Override
public void setPageSource(PageReadStore pageStore) {
this.rowGroupStart = pageStore.getRowIndexOffset().orElse(0L);
this.rowOffset = -1;
}
}
Expand All @@ -221,6 +232,11 @@ protected PrimitiveReader(ColumnDescriptor desc) {

@Override
public void setPageSource(PageReadStore pageStore, long rowPosition) {
setPageSource(pageStore);
}

@Override
public void setPageSource(PageReadStore pageStore) {
column.setPageSource(pageStore.getPageReader(desc));
}

Expand Down Expand Up @@ -405,7 +421,12 @@ private static class OptionReader<T> implements ParquetValueReader<T> {

@Override
public void setPageSource(PageReadStore pageStore, long rowPosition) {
reader.setPageSource(pageStore, rowPosition);
setPageSource(pageStore);
}

@Override
public void setPageSource(PageReadStore pageStore) {
reader.setPageSource(pageStore);
}

@Override
Expand Down Expand Up @@ -450,7 +471,12 @@ protected RepeatedReader(

@Override
public void setPageSource(PageReadStore pageStore, long rowPosition) {
reader.setPageSource(pageStore, rowPosition);
setPageSource(pageStore);
}

@Override
public void setPageSource(PageReadStore pageStore) {
reader.setPageSource(pageStore);
}

@Override
Expand Down Expand Up @@ -569,8 +595,13 @@ protected RepeatedKeyValueReader(

@Override
public void setPageSource(PageReadStore pageStore, long rowPosition) {
keyReader.setPageSource(pageStore, rowPosition);
valueReader.setPageSource(pageStore, rowPosition);
setPageSource(pageStore);
}

@Override
public void setPageSource(PageReadStore pageStore) {
keyReader.setPageSource(pageStore);
valueReader.setPageSource(pageStore);
}

@Override
Expand Down Expand Up @@ -720,8 +751,13 @@ protected StructReader(List<Type> types, List<ParquetValueReader<?>> readers) {

@Override
public final void setPageSource(PageReadStore pageStore, long rowPosition) {
setPageSource(pageStore);
}

@Override
public final void setPageSource(PageReadStore pageStore) {
for (ParquetValueReader<?> reader : readers) {
reader.setPageSource(pageStore, rowPosition);
reader.setPageSource(pageStore);
}
}

Expand Down
44 changes: 0 additions & 44 deletions parquet/src/main/java/org/apache/iceberg/parquet/ReadConf.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,23 +19,19 @@
package org.apache.iceberg.parquet;

import java.io.IOException;
import java.io.UncheckedIOException;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.function.Function;
import java.util.stream.Collectors;
import org.apache.iceberg.MetadataColumns;
import org.apache.iceberg.Schema;
import org.apache.iceberg.exceptions.RuntimeIOException;
import org.apache.iceberg.expressions.Expression;
import org.apache.iceberg.io.InputFile;
import org.apache.iceberg.mapping.NameMapping;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.parquet.ParquetReadOptions;
import org.apache.parquet.crypto.FileDecryptionProperties;
import org.apache.parquet.hadoop.ParquetFileReader;
import org.apache.parquet.hadoop.metadata.BlockMetaData;
import org.apache.parquet.hadoop.metadata.ColumnChunkMetaData;
Expand All @@ -59,7 +55,6 @@ class ReadConf<T> {
private final long totalValues;
private final boolean reuseContainers;
private final Integer batchSize;
private final long[] startRowPositions;

// List of column chunk metadata for each row group
private final List<Map<ColumnPath, ColumnChunkMetaData>> columnChunkMetaDataForRowGroups;
Expand Down Expand Up @@ -95,10 +90,6 @@ class ReadConf<T> {

this.rowGroups = reader.getRowGroups();
this.shouldSkip = new boolean[rowGroups.size()];
this.startRowPositions = new long[rowGroups.size()];

// Fetch all row groups starting positions to compute the row offsets of the filtered row groups
Map<Long, Long> offsetToStartPos = generateOffsetToStartPos(expectedSchema);

ParquetMetricsRowGroupFilter statsFilter = null;
ParquetDictionaryRowGroupFilter dictFilter = null;
Expand All @@ -112,8 +103,6 @@ class ReadConf<T> {
long computedTotalValues = 0L;
for (int i = 0; i < shouldSkip.length; i += 1) {
BlockMetaData rowGroup = rowGroups.get(i);
startRowPositions[i] =
offsetToStartPos == null ? 0 : offsetToStartPos.get(rowGroup.getStartingPos());
boolean shouldRead =
filter == null
|| (statsFilter.shouldRead(typeWithIds, rowGroup)
Expand Down Expand Up @@ -155,7 +144,6 @@ private ReadConf(ReadConf<T> toCopy) {
this.batchSize = toCopy.batchSize;
this.vectorizedModel = toCopy.vectorizedModel;
this.columnChunkMetaDataForRowGroups = toCopy.columnChunkMetaDataForRowGroups;
this.startRowPositions = toCopy.startRowPositions;
}

ParquetFileReader reader() {
Expand All @@ -181,38 +169,6 @@ boolean[] shouldSkip() {
return shouldSkip;
}

private Map<Long, Long> generateOffsetToStartPos(Schema schema) {
if (schema.findField(MetadataColumns.ROW_POSITION.fieldId()) == null) {
return null;
}

FileDecryptionProperties decryptionProperties =
(options == null) ? null : options.getDecryptionProperties();

ParquetReadOptions readOptions =
ParquetReadOptions.builder().withDecryption(decryptionProperties).build();

try (ParquetFileReader fileReader = newReader(file, readOptions)) {
Map<Long, Long> offsetToStartPos = Maps.newHashMap();

long curRowCount = 0;
for (int i = 0; i < fileReader.getRowGroups().size(); i += 1) {
BlockMetaData meta = fileReader.getRowGroups().get(i);
offsetToStartPos.put(meta.getStartingPos(), curRowCount);
curRowCount += meta.getRowCount();
}

return offsetToStartPos;

} catch (IOException e) {
throw new UncheckedIOException("Failed to create/close reader for file: " + file, e);
}
}

long[] startRowPositions() {
return startRowPositions;
}

long totalValues() {
return totalValues;
}
Expand Down
5 changes: 1 addition & 4 deletions parquet/src/main/java/org/apache/iceberg/parquet/VectorizedParquetReader.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -113,7 +113,6 @@ private static class FileIterator<T> implements CloseableIterator<T> {
private long nextRowGroupStart = 0;
private long valuesRead = 0;
private T last = null;
private final long[] rowGroupsStartRowPos;

FileIterator(ReadConf conf) {
this.reader = conf.reader();
Expand All @@ -124,7 +123,6 @@ private static class FileIterator<T> implements CloseableIterator<T> {
this.batchSize = conf.batchSize();
this.model.setBatchSize(this.batchSize);
this.columnChunkMetadata = conf.columnChunkMetadataForRowGroups();
this.rowGroupsStartRowPos = conf.startRowPositions();
}

@Override
Expand Down Expand Up @@ -165,8 +163,7 @@ private void advance() {
throw new RuntimeIOException(e);
}

long rowPosition = rowGroupsStartRowPos[nextRowGroup];
model.setRowGroupInfo(pages, columnChunkMetadata.get(nextRowGroup), rowPosition);
model.setRowGroupInfo(pages, columnChunkMetadata.get(nextRowGroup));
nextRowGroupStart += pages.getRowCount();
nextRowGroup += 1;
}
Expand Down
15 changes: 15 additions & 0 deletions parquet/src/main/java/org/apache/iceberg/parquet/VectorizedReader.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,10 +43,25 @@ public interface VectorizedReader<T> {
* @param pages row group information for all the columns
* @param metadata map of {@link ColumnPath} -&gt; {@link ColumnChunkMetaData} for the row group
* @param rowPosition the row group's row offset in the parquet file
* @deprecated since 1.8.0, will be removed in 1.9.0; use setRowGroupInfo(PageReadStore,
* Map&lt;ColumnPath, ColumnChunkMetaData&gt;) instead.
*/
@Deprecated
void setRowGroupInfo(
PageReadStore pages, Map<ColumnPath, ColumnChunkMetaData> metadata, long rowPosition);

/**
* Sets the row group information to be used with this reader
*
* @param pages row group information for all the columns
* @param metadata map of {@link ColumnPath} -&gt; {@link ColumnChunkMetaData} for the row group
*/
default void setRowGroupInfo(PageReadStore pages, Map<ColumnPath, ColumnChunkMetaData> metadata) {
throw new UnsupportedOperationException(
this.getClass().getName()
+ " doesn't implement setRowGroupInfo(PageReadStore, Map<ColumnPath, ColumnChunkMetaData>)");
}

/** Release any resources allocated. */
void close();
}
10 changes: 8 additions & 2 deletions ...3.3/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/ColumnarBatchReader.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,8 +55,14 @@ public ColumnarBatchReader(List<VectorizedReader<?>> readers) {
@Override
public void setRowGroupInfo(
PageReadStore pageStore, Map<ColumnPath, ColumnChunkMetaData> metaData, long rowPosition) {
super.setRowGroupInfo(pageStore, metaData, rowPosition);
this.rowStartPosInBatch = rowPosition;
setRowGroupInfo(pageStore, metaData);
}

@Override
public void setRowGroupInfo(
PageReadStore pageStore, Map<ColumnPath, ColumnChunkMetaData> metaData) {
super.setRowGroupInfo(pageStore, metaData);
this.rowStartPosInBatch = pageStore.getRowIndexOffset().orElse(0L);
}

public void setDeleteFilter(DeleteFilter<InternalRow> deleteFilter) {
Expand Down
Loading