Skip to content

ORC: Use ConstantReader for identity partition columns #1191

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
rdblue merged 3 commits into from
Jul 11, 2020
Merged

ORC: Use ConstantReader for identity partition columns #1191

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
c55c16e
ORC: Use ConstantReader for identity partition columns
shardulm94 Jul 10, 2020
26cbff7
Do not project constant columns from ORC reader
shardulm94 Jul 10, 2020
1439094
Address PR comments
shardulm94 Jul 10, 2020
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
4 changes: 3 additions & 1 deletion data/src/main/java/org/apache/iceberg/data/TableScanIterable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,7 @@
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.types.TypeUtil;
import org.apache.iceberg.util.PartitionUtil;

class TableScanIterable extends CloseableGroup implements CloseableIterable<Record> {
Expand Down Expand Up @@ -106,8 +107,9 @@ private CloseableIterable<Record> open(FileScanTask task) {
return parquet.build();

case ORC:
Schema projectionWithoutConstants = TypeUtil.selectNot(projection, partition.keySet());
ORC.ReadBuilder orc = ORC.read(input)
.project(projection)
.project(projectionWithoutConstants)
.createReaderFunc(fileSchema -> GenericOrcReader.buildReader(projection, fileSchema, partition))
.split(task.start(), task.length())
.filter(task.residual());
Expand Down
6 changes: 4 additions & 2 deletions mr/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergInputFormat.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -321,8 +321,10 @@ private CloseableIterable<T> newParquetIterable(InputFile inputFile, FileScanTas
}

private CloseableIterable<T> newOrcIterable(InputFile inputFile, FileScanTask task, Schema readSchema) {
Map<Integer, ?> idToConstant = constantsMap(task, IdentityPartitionConverters::convertConstant);
Schema readSchemaWithoutConstants = TypeUtil.selectNot(readSchema, idToConstant.keySet());
ORC.ReadBuilder orcReadBuilder = ORC.read(inputFile)
.project(readSchema)
.project(readSchemaWithoutConstants)
.filter(task.residual())
.caseSensitive(caseSensitive)
.split(task.start(), task.length());
Expand All @@ -335,7 +337,7 @@ private CloseableIterable<T> newOrcIterable(InputFile inputFile, FileScanTask ta
case GENERIC:
orcReadBuilder.createReaderFunc(
fileSchema -> GenericOrcReader.buildReader(
readSchema, fileSchema, constantsMap(task, IdentityPartitionConverters::convertConstant)));
readSchema, fileSchema, idToConstant));
}

return applyResidualFiltering(orcReadBuilder.build(), task.residual(), readSchema);
Expand Down
66 changes: 40 additions & 26 deletions orc/src/main/java/org/apache/iceberg/orc/OrcValueReaders.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,6 @@
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.types.Types;
import org.apache.orc.storage.ql.exec.vector.BytesColumnVector;
import org.apache.orc.storage.ql.exec.vector.ColumnVector;
Expand Down Expand Up @@ -59,6 +58,10 @@ public static OrcValueReader<byte[]> bytes() {
return BytesReader.INSTANCE;
}

public static <C> OrcValueReader<C> constants(C constant) {
return new ConstantReader<>(constant);
}

private static class BooleanReader implements OrcValueReader<Boolean> {
static final BooleanReader INSTANCE = new BooleanReader();

Expand Down Expand Up @@ -136,31 +139,22 @@ public byte[] nonNullRead(ColumnVector vector, int row) {

public abstract static class StructReader<T> implements OrcValueReader<T> {
private final OrcValueReader<?>[] readers;
private final int[] positions;
private final Object[] constants;

protected StructReader(List<OrcValueReader<?>> readers) {
this.readers = readers.toArray(new OrcValueReader[0]);
this.positions = new int[0];
this.constants = new Object[0];
}
private final boolean[] isConstantField;

protected StructReader(List<OrcValueReader<?>> readers, Types.StructType struct, Map<Integer, ?> idToConstant) {
this.readers = readers.toArray(new OrcValueReader[0]);
List<Types.NestedField> fields = struct.fields();
List<Integer> positionList = Lists.newArrayListWithCapacity(fields.size());
List<Object> constantList = Lists.newArrayListWithCapacity(fields.size());
for (int pos = 0; pos < fields.size(); pos += 1) {
this.readers = new OrcValueReader[fields.size()];
this.isConstantField = new boolean[fields.size()];
for (int pos = 0, readerIndex = 0; pos < fields.size(); pos += 1) {
Types.NestedField field = fields.get(pos);
Object constant = idToConstant.get(field.fieldId());
if (constant != null) {
positionList.add(pos);
constantList.add(idToConstant.get(field.fieldId()));
if (idToConstant.containsKey(field.fieldId())) {
this.isConstantField[pos] = true;
this.readers[pos] = constants(idToConstant.get(field.fieldId()));
} else {
this.readers[pos] = readers.get(readerIndex);
readerIndex++;
}
}

this.positions = positionList.stream().mapToInt(Integer::intValue).toArray();
this.constants = constantList.toArray();
}

protected abstract T create();
Expand All @@ -178,15 +172,35 @@ public T nonNullRead(ColumnVector vector, int row) {
}

private T readInternal(T struct, ColumnVector[] columnVectors, int row) {
for (int c = 0; c < readers.length; ++c) {
set(struct, c, reader(c).read(columnVectors[c], row));
for (int c = 0, vectorIndex = 0; c < readers.length; ++c) {
ColumnVector vector;
if (isConstantField[c]) {
vector = null;
} else {
vector = columnVectors[vectorIndex];
vectorIndex++;
}
set(struct, c, reader(c).read(vector, row));
}
return struct;
}
}

for (int i = 0; i < positions.length; i += 1) {
set(struct, positions[i], constants[i]);
}
private static class ConstantReader<C> implements OrcValueReader<C> {
private final C constant;

return struct;
private ConstantReader(C constant) {
this.constant = constant;
}

@Override
public C read(ColumnVector ignored, int ignoredRow) {
Copy link
Contributor

@rdblue rdblue Jul 10, 2020

Choose a reason for hiding this comment

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

Is ColumnVector still materialized? Is it possible to avoid reading that entirely?

Copy link
Contributor Author

@shardulm94 shardulm94 Jul 10, 2020 •
edited
Loading

Choose a reason for hiding this comment

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

Yes, I guess we can do that by not asking ORC to project these columns. Let me give it a try.

Copy link
Contributor

@rdblue rdblue Jul 10, 2020

Choose a reason for hiding this comment

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

That's going to be a bigger time savings and that's what we do for Parquet. We just drop it from the projection we pass down to the format.

return constant;
}

@Override
public C nonNullRead(ColumnVector ignored, int ignoredRow) {
return constant;
}
}
}
2 changes: 1 addition & 1 deletion spark/src/main/java/org/apache/iceberg/spark/data/SparkOrcValueReaders.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -120,7 +120,7 @@ static class StructReader extends OrcValueReaders.StructReader<InternalRow> {

protected StructReader(List<OrcValueReader<?>> readers, Types.StructType struct, Map<Integer, ?> idToConstant) {
super(readers, struct, idToConstant);
this.numFields = readers.size();
this.numFields = struct.fields().size();
}

@Override
Expand Down
3 changes: 2 additions & 1 deletion spark/src/main/java/org/apache/iceberg/spark/source/RowDataReader.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -178,8 +178,9 @@ private CloseableIterable<InternalRow> newOrcIterable(
FileScanTask task,
Schema readSchema,
Map<Integer, ?> idToConstant) {
Schema readSchemaWithoutConstants = TypeUtil.selectNot(readSchema, idToConstant.keySet());
return ORC.read(location)
.project(readSchema)
.project(readSchemaWithoutConstants)
.split(task.start(), task.length())
.createReaderFunc(readOrcSchema -> new SparkOrcReader(readSchema, readOrcSchema, idToConstant))
.filter(task.residual())
Expand Down