Parquet: Add readers and writers for the internal object model #11904

ajantha-bhat · 2025-01-03T16:27:41Z

Refactor BaseParquetWriter and BaseParquetReaders to reuse for internal writers and readers.
Added InternalWriter and InternalReader class for parquet that consumes and produces the Iceberg in-memory data model.
Fixed some bugs in genric readers like UUID, timestamp millis, fixed length validatation etc.

parquet/src/main/java/org/apache/iceberg/data/parquet/InternalReader.java

parquet/src/test/java/org/apache/iceberg/parquet/TestInternalWriter.java

.palantir/revapi.yml

rdblue · 2025-01-07T20:42:30Z

parquet/src/main/java/org/apache/iceberg/parquet/ParquetValueReaders.java

+
+    @Override
+    public UUID read(UUID reuse) {
+      return UUIDUtil.convert(column.nextBinary().toByteBuffer());


This looks fine to me.

parquet/src/main/java/org/apache/iceberg/data/parquet/InternalWriter.java

rdblue · 2025-01-07T21:41:35Z

parquet/src/main/java/org/apache/iceberg/data/parquet/InternalReader.java

+    return new ParquetValueReaders.UnboxedReader<>(desc);
+  }
+
+  private static class ParquetStructReader extends StructReader<StructLike, StructLike> {


Here also, there's not much value in using Parquet in the class name. Since this will produce GenericRecord instances, how about RecordReader?

When checking that name (RecordReader) for consistency, I noticed that there's already a RecordReader in GenericParquetReaders. You can reuse that class.

Cannot reuse the class from GenericParquetReaders as it is based on Record interface, we need a class based on StructLike interface.

I will rename to StructLikeReader, just like the StructLikeWriter from InternalWriter class.

rdblue · 2025-01-07T22:04:52Z

parquet/src/main/java/org/apache/iceberg/data/parquet/InternalReader.java

+  @Override
+  protected ParquetValueReaders.PrimitiveReader<?> int96Reader(ColumnDescriptor desc) {
+    // normal handling as int96
+    return new ParquetValueReaders.UnboxedReader<>(desc);


This isn't correct. The unboxed reader will return a Binary for int96 columns. Instead, this needs to use the same logic as the Spark reader (which also uses the internal representation):

private static class TimestampInt96Reader extends UnboxedReader<Long> { TimestampInt96Reader(ColumnDescriptor desc) { super(desc); } @Override public Long read(Long ignored) { return readLong(); } @Override public long readLong() { final ByteBuffer byteBuffer = column.nextBinary().toByteBuffer().order(ByteOrder.LITTLE_ENDIAN); return ParquetUtil.extractTimestampInt96(byteBuffer); } }

You can move that class into the parquet package to share it.

rdblue · 2025-01-07T22:44:17Z

parquet/src/main/java/org/apache/iceberg/data/parquet/BaseParquetReaders.java

@@ -359,10 +250,10 @@ public ParquetValueReader<?> primitive(

      ColumnDescriptor desc = type.getColumnDescription(currentPath());

-      if (primitive.getOriginalType() != null) {
+      if (primitive.getLogicalTypeAnnotation() != null) {


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 ConvertedType (which is what OriginalType is called in Parquet format and the logical type docs).

parquet/src/main/java/org/apache/iceberg/data/parquet/BaseParquetReaders.java

rdblue · 2025-01-07T22:54:07Z

parquet/src/main/java/org/apache/iceberg/data/parquet/BaseParquetReaders.java

@@ -76,6 +64,16 @@ protected ParquetValueReader<T> createReader(
  protected abstract ParquetValueReader<T> createStructReader(
      List<Type> types, List<ParquetValueReader<?>> fieldReaders, Types.StructType structType);

+  protected abstract LogicalTypeAnnotation.LogicalTypeAnnotationVisitor<ParquetValueReader<?>>


I don't think it makes sense to have the subclasses provide this visitor.

rdblue · 2025-01-07T22:54:52Z

parquet/src/main/java/org/apache/iceberg/data/parquet/GenericParquetReaders.java

+  private static final OffsetDateTime EPOCH = Instant.ofEpochSecond(0).atOffset(ZoneOffset.UTC);
+  private static final LocalDate EPOCH_DAY = EPOCH.toLocalDate();
+
+  private static class DateReader extends ParquetValueReaders.PrimitiveReader<LocalDate> {


I agree with moving the date/time reader classes here.

rdblue · 2025-01-07T22:57:17Z

parquet/src/main/java/org/apache/iceberg/data/parquet/InternalReader.java

+    @Override
+    public Optional<ParquetValueReader<?>> visit(
+        LogicalTypeAnnotation.TimestampLogicalTypeAnnotation timestampLogicalType) {
+      return Optional.of(new ParquetValueReaders.UnboxedReader<>(desc));


This isn't correct. The unit of the incoming timestamp value still needs to be handled, even if the in-memory representation of the value is the same (a long).

Looks like the Spark implementations for this should work well, just like the int96 cases.

parquet/src/main/java/org/apache/iceberg/data/parquet/InternalReader.java

parquet/src/main/java/org/apache/iceberg/data/parquet/GenericParquetWriter.java

rdblue · 2025-01-07T23:08:04Z

parquet/src/test/java/org/apache/iceberg/parquet/TestInternalWriter.java

+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+public class TestInternalWriter {


As with the Avro tests, I think this should extend DataTest. It is probably easier to do the Avro work first and then reuse it here.

rdblue · 2025-01-07T23:10:38Z

.palantir/revapi.yml

+        \ org.apache.iceberg.data.parquet.BaseParquetReaders<T>::logicalTypeReaderVisitor(org.apache.parquet.column.ColumnDescriptor,\
+        \ org.apache.iceberg.types.Type.PrimitiveType, org.apache.parquet.schema.PrimitiveType)"
+      justification: "{Refactor Parquet reader and writer}"
+    - code: "java.method.abstractMethodAdded"


This PR should not introduce revapi failures. Instead, the new methods should have default implementations that match the previous behavior (returning the generic representations).

New methods are abstract and abstract method cannot have default implementation. So, I think we have to handle revapi failures.

Oh, I think what you mean is don't add it as abstract. Add it as methods with default implementation. I got it. I will update it today.

rdblue · 2025-01-22T20:07:36Z

parquet/src/main/java/org/apache/iceberg/parquet/ParquetValueReaders.java

@@ -850,4 +919,42 @@ private TripleIterator<?> firstNonNullColumn(List<TripleIterator<?>> columns) {
      return NullReader.NULL_COLUMN;
    }
  }
+
+  private static class RecordReader<T extends StructLike> extends StructReader<T, T> {


This returns Record so I don't think it needed to be modified. It doesn't return any other subclass of StructLike.

parquet/src/main/java/org/apache/iceberg/data/parquet/BaseParquetReaders.java

rdblue · 2025-01-22T20:14:09Z

parquet/src/main/java/org/apache/iceberg/data/parquet/BaseParquetWriter.java

+
+  protected ParquetValueWriter<?> uuidWriter(ColumnDescriptor desc) {
+    // Use primitive-type writer (as FIXED_LEN_BYTE_ARRAY); no special writer needed.
+    return null;


I think I commented on this in the last round of reviews. This isn't correct. Incoming values are of type UUID so this needs a writer that can convert UUID into a byte array. This should return ParquetValueWriters.uuids(desc).

There's also no need to add a method for this because it is the same between the generic and internal object models.

Sorry, the existing testcases was the reason for confusion as I mentioned in #11904 (comment)

I will update the existing testcases of Arrow too in this PR.

parquet/src/main/java/org/apache/iceberg/data/parquet/BaseParquetWriter.java

rdblue · 2025-01-22T20:19:15Z

parquet/src/main/java/org/apache/iceberg/parquet/ParquetValueReaders.java

+    return new ParquetValueReaders.TimestampMillisReader(desc);
+  }
+
+  public static <T extends StructLike> StructReader<T, T> recordReader(


Should be ParquetValueReader<Record>

parquet/src/main/java/org/apache/iceberg/data/parquet/InternalWriter.java

parquet/src/main/java/org/apache/iceberg/data/parquet/GenericParquetReaders.java

parquet/src/main/java/org/apache/iceberg/data/parquet/GenericParquetWriter.java

ajantha-bhat · 2025-01-23T12:30:31Z

@rdblue: Thanks for giving additional context for unresolved comments. I think I understood all the comments this time. PR is ready. It also fixed base code issues and testcases.

parquet/src/main/java/org/apache/iceberg/data/parquet/InternalReader.java

rdblue · 2025-01-23T20:18:40Z

parquet/src/main/java/org/apache/iceberg/parquet/ParquetValueReaders.java

+
+    @Override
+    public long readLong() {
+      return 1000L * column.nextInteger();


This is valid for time but not for timestamp. I may have mixed up the timestamp reader and time reader in an earlier comment. This needs to be nextLong.

I think the confusion was from this comment: #11904 (comment)

I was talking about the time type, but the code I pasted had the wrong class name, TimestampMillisReader should have been TimeMillisReader. Timestamps (millis) should use nextLong and time (millis) should use nextInteger.

Fixed in ajantha-bhat#74

Lack of test coverage in the base code for these milliseconds time, timestamp and int96 timestamps is the reason for back and forth. Tests would have caught this. Will try to add in a follow up.

parquet/src/main/java/org/apache/iceberg/data/parquet/InternalReader.java

ajantha-bhat · 2025-01-24T16:41:38Z

rebasing the PR as Flink hit a flaky test #11833 (comment)

rdblue · 2025-01-24T21:53:19Z

Thanks, @ajantha-bhat! Good to get this in.

ajantha-bhat marked this pull request as draft January 3, 2025 16:27

github-actions bot added the parquet label Jan 3, 2025

ajantha-bhat requested a review from rdblue January 3, 2025 16:30

ajantha-bhat commented Jan 3, 2025

View reviewed changes

parquet/src/main/java/org/apache/iceberg/data/parquet/InternalReader.java Outdated Show resolved Hide resolved

ajantha-bhat commented Jan 3, 2025

View reviewed changes

parquet/src/test/java/org/apache/iceberg/parquet/TestInternalWriter.java Outdated Show resolved Hide resolved

ajantha-bhat closed this Jan 4, 2025

ajantha-bhat reopened this Jan 4, 2025

ajantha-bhat force-pushed the parquet_internal_writer branch from 772f5c2 to 233a00b Compare January 6, 2025 11:33