Parquet: Add readers and writers for the internal object model #11904
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -18,19 +18,9 @@ | |
| */ | ||
| package org.apache.iceberg.data.parquet; | ||
|
|
||
| import java.util.List; | ||
| import java.util.Map; | ||
| import java.util.Optional; | ||
| import org.apache.iceberg.MetadataColumns; | ||
| import org.apache.iceberg.Schema; | ||
| import org.apache.iceberg.parquet.ParquetSchemaUtil; | ||
|
|
@@ -50,6 +40,10 @@ | |
| import org.apache.parquet.schema.PrimitiveType; | ||
| import org.apache.parquet.schema.Type; | ||
|
|
||
| /** | ||
| * @deprecated since 1.8.0, will be made package-private in 1.9.0 | ||
| */ | ||
| @Deprecated | ||
| public abstract class BaseParquetReaders<T> { | ||
| protected BaseParquetReaders() {} | ||
|
|
||
|
|
@@ -76,6 +70,46 @@ protected ParquetValueReader<T> createReader( | |
| protected abstract ParquetValueReader<T> createStructReader( | ||
| List<Type> types, List<ParquetValueReader<?>> fieldReaders, Types.StructType structType); | ||
|
|
||
| protected ParquetValueReader<?> fixedReader(ColumnDescriptor desc) { | ||
| return new GenericParquetReaders.FixedReader(desc); | ||
| } | ||
|
|
||
| protected ParquetValueReader<?> dateReader(ColumnDescriptor desc) { | ||
| return new GenericParquetReaders.DateReader(desc); | ||
| } | ||
|
|
||
| protected ParquetValueReader<?> timeReader( | ||
| ColumnDescriptor desc, LogicalTypeAnnotation.TimeUnit unit) { | ||
| switch (unit) { | ||
| case MICROS: | ||
| return new GenericParquetReaders.TimeReader(desc); | ||
| case MILLIS: | ||
| return new GenericParquetReaders.TimeMillisReader(desc); | ||
| default: | ||
| throw new UnsupportedOperationException("Unsupported unit for time: " + unit); | ||
| } | ||
| } | ||
|
|
||
| protected ParquetValueReader<?> timestampReader( | ||
| ColumnDescriptor desc, LogicalTypeAnnotation.TimeUnit unit, boolean isAdjustedToUTC) { | ||
| if (desc.getPrimitiveType().getPrimitiveTypeName() == PrimitiveType.PrimitiveTypeName.INT96) { | ||
| return new GenericParquetReaders.TimestampInt96Reader(desc); | ||
| } | ||
|
|
||
| switch (unit) { | ||
| case MICROS: | ||
| return isAdjustedToUTC | ||
| ? new GenericParquetReaders.TimestamptzReader(desc) | ||
| : new GenericParquetReaders.TimestampReader(desc); | ||
| case MILLIS: | ||
| return isAdjustedToUTC | ||
| ? new GenericParquetReaders.TimestamptzMillisReader(desc) | ||
| : new GenericParquetReaders.TimestampMillisReader(desc); | ||
| default: | ||
| throw new UnsupportedOperationException("Unsupported unit for timestamp: " + unit); | ||
| } | ||
| } | ||
|
|
||
| protected Object convertConstant(org.apache.iceberg.types.Type type, Object value) { | ||
| return value; | ||
| } | ||
|
|
@@ -164,37 +198,23 @@ public Optional<ParquetValueReader<?>> visit(DecimalLogicalTypeAnnotation decima | |
| @Override | ||
| public Optional<ParquetValueReader<?>> visit( | ||
| LogicalTypeAnnotation.DateLogicalTypeAnnotation dateLogicalType) { | ||
ajantha-bhat marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| return Optional.of(dateReader(desc)); | ||
| } | ||
|
|
||
| @Override | ||
| public Optional<ParquetValueReader<?>> visit( | ||
| LogicalTypeAnnotation.TimeLogicalTypeAnnotation timeLogicalType) { | ||
| return Optional.of(timeReader(desc, timeLogicalType.getUnit())); | ||
| } | ||
|
|
||
| @Override | ||
| public Optional<ParquetValueReader<?>> visit( | ||
| LogicalTypeAnnotation.TimestampLogicalTypeAnnotation timestampLogicalType) { | ||
| return Optional.of( | ||
| timestampReader( | ||
| desc, | ||
| timestampLogicalType.getUnit(), | ||
| ((Types.TimestampType) expected).shouldAdjustToUTC())); | ||
| } | ||
|
|
||
| @Override | ||
|
|
@@ -219,6 +239,12 @@ public Optional<ParquetValueReader<?>> visit( | |
| LogicalTypeAnnotation.BsonLogicalTypeAnnotation bsonLogicalType) { | ||
| return Optional.of(new ParquetValueReaders.BytesReader(desc)); | ||
| } | ||
|
|
||
| @Override | ||
| public Optional<ParquetValueReader<?>> visit( | ||
| LogicalTypeAnnotation.UUIDLogicalTypeAnnotation uuidLogicalType) { | ||
| return Optional.of(ParquetValueReaders.uuids(desc)); | ||
| } | ||
| } | ||
|
|
||
| private class ReadBuilder extends TypeWithSchemaVisitor<ParquetValueReader<?>> { | ||
|
|
@@ -359,7 +385,7 @@ public ParquetValueReader<?> primitive( | |
|
|
||
| ColumnDescriptor desc = type.getColumnDescription(currentPath()); | ||
|
|
||
| if (primitive.getLogicalTypeAnnotation() != null) { | ||
|
There was a problem hiding this comment. I agree with this change, but please point these kinds of changes out for reviewers. The old version worked because all of the supported logical type annotations had an equivalent |
||
| return primitive | ||
| .getLogicalTypeAnnotation() | ||
| .accept(new LogicalTypeAnnotationParquetValueReaderVisitor(desc, expected, primitive)) | ||
|
|
@@ -371,7 +397,7 @@ public ParquetValueReader<?> primitive( | |
|
|
||
| switch (primitive.getPrimitiveTypeName()) { | ||
| case FIXED_LEN_BYTE_ARRAY: | ||
| return fixedReader(desc); | ||
| case BINARY: | ||
| if (expected.typeId() == org.apache.iceberg.types.Type.TypeID.STRING) { | ||
| return new ParquetValueReaders.StringReader(desc); | ||
|
|
@@ -397,7 +423,7 @@ public ParquetValueReader<?> primitive( | |
| case INT96: | ||
| // Impala & Spark used to write timestamps as INT96 without a logical type. For backwards | ||
| // compatibility we try to read INT96 as timestamps. | ||
| return timestampReader(desc, LogicalTypeAnnotation.TimeUnit.NANOS, true); | ||
| default: | ||
| throw new UnsupportedOperationException("Unsupported type: " + primitive); | ||
| } | ||
|
|
@@ -407,124 +433,4 @@ MessageType type() { | |
| return type; | ||
| } | ||
| } | ||
| } | ||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Addressing the nit from #11919
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
How about
elementSupplierinstead? I think that's more accurate thanelementsfor two reasons. First, the argument is not an element, it is a function that produces an element. Second, there is only one supplier so the name shouldn't be plural.