Update Flink Parquet reader and writer to use schema visitor #1237

chenjunjiedada · 2020-07-23T09:19:12Z

This changes the current FlinkParquetReader/Writer to use a schema visitor.

As discussed in #1215, the current FlinkParquetReader/Writer are not built with schema visitor so that it cannot recognize the field in Flink data model Row or RowData. This PR change adds ReaderBuilder and WriterBuilder that extends the schema visitor so that it could build value readers and writers though schema visiting.

This assumes the Flink internal data model is RowData rather than Row, it also adds aRandomRowDataGenerator, which generates random RowData, and assertRowData assertion for the unit test.

rdblue · 2020-07-26T18:47:58Z

flink/src/main/java/org/apache/iceberg/flink/data/FlinkParquetReaders.java

+    }
+  }
+
+  private static class FallbackReadBuilder extends ReadBuilder {


This can probably be done later, but I think we could rewrite the fallback read builder to use a visitor that is passed in. That way we could use the same one across sources:

static class FallbackReadBuilder extends TypeWithSchemaVisitor<ParquetValueReader<?>> { private final TypeWithSchemaVisitor<ParquetValueReader<?>> builder; FallbackReadBuilder(TypeWithSchemaVisitor<ParquetValueReader<?>> builder) { this.builder = builder; } @Override public ParquetValueReader<?> message(Types.StructType expected, MessageType message, List<ParquetValueReader<?>> fieldReaders) { // the top level matches by ID, but the remaining IDs are missing builder.struct(expected, message, fieldReaders); } @Override public ParquetValueReader<?> struct(Types.StructType ignored, GroupType struct, List<ParquetValueReader<?>> fieldReaders) { // the expected struct is ignored because nested fields are never found ... } }

rdblue · 2020-07-26T18:53:03Z

flink/src/main/java/org/apache/iceberg/flink/data/FlinkParquetReaders.java

+            }
+          case INT_64:
+          case TIMESTAMP_MICROS:
+            return new TimestampMicroReader(desc);


The timestamp reader should not be used for longs.

rdblue · 2020-07-26T18:58:03Z

flink/src/main/java/org/apache/iceberg/flink/data/FlinkParquetReaders.java

+          case INT_8:
+          case INT_16:
+          case INT_32:
+          case TIME_MICROS:


According to the table on #1215, a time should be in milliseconds. Because Parquet stores the time in micros for TIME_MICROS, this will need a converter.

@JingsongLi, how should we handle lossy conversions to Flink types?

Yes, unfortunately Flink can only store time in milliseconds. (Although the time type can be defined to nanoseconds, the internal implementation is only milliseconds).
Compared with behavior that throwing exception when encountering a non-zero micro value (If there is a piece of dirty data, it will make the program very dangerous), I prefer just ignore micro part. Though it's not friendly, the user is most convenient to use.

We need to document this behavior.

Done. FYI, I discard the microseconds in milliseconds in #1266.

rdblue · 2020-07-26T19:07:09Z

flink/src/main/java/org/apache/iceberg/flink/data/FlinkParquetReaders.java

+
+    @Override
+    public TimestampData read(TimestampData ignored) {
+      return TimestampData.fromEpochMillis(readLong() / 1000);


This should not discard microseconds. It should instead call TimestampData.fromEpochMillis(millis, nanosOfMilli).

rdblue · 2020-07-26T19:08:14Z

flink/src/main/java/org/apache/iceberg/flink/data/FlinkParquetWriters.java

+          case INT_8:
+          case INT_16:
+          case INT_32:
+          case TIME_MICROS:


Like the read path, time needs to be handled separately to convert from millis to micros.

Will update in the writer side PR.

rdblue · 2020-07-26T19:08:35Z

flink/src/main/java/org/apache/iceberg/flink/data/FlinkParquetWriters.java

+          case INT_32:
+          case TIME_MICROS:
+            return ints(sType, desc);
+          case INT_64:


The timestamp writer should not be used for longs.

rdblue · 2020-07-26T19:11:44Z

flink/src/main/java/org/apache/iceberg/flink/data/FlinkParquetReaders.java


-  public static ParquetValueReader<Row> buildReader(Schema expectedSchema, MessageType fileSchema) {
-    return INSTANCE.createReader(expectedSchema, fileSchema);
+  private static class BytesReader extends ParquetValueReaders.PrimitiveReader<byte[]> {


This is identical to the reader in SparkParquetReaders. Can you move the reader to ParquetValueReaders and use the same implementation?

rdblue · 2020-07-26T19:15:44Z

flink/src/main/java/org/apache/iceberg/flink/data/FlinkParquetWriters.java

+
+      BigDecimal bigDecimal = decimal.toBigDecimal();
+
+      byte fillByte = (byte) (bigDecimal.signum() < 0 ? 0xFF : 0x00);


I think this should be rewritten to use an abstract implementation that defines write(in repetitionLevel, BigDecimal decimal). This is the same as the one in SparkParquetWriters, so we should avoid duplicating this.

The #1265 PR seams doing the refactor job. I will rebase the writer PR when it has done.

rdblue · 2020-07-26T19:16:02Z

flink/src/main/java/org/apache/iceberg/flink/data/FlinkParquetWriters.java

+    }
+  }
+
+  private static class TimeStampDataWriter extends ParquetValueWriters.PrimitiveWriter<TimestampData> {


Typo: TimeStamp should be Timestamp

rdblue · 2020-07-26T19:16:23Z

flink/src/main/java/org/apache/iceberg/flink/data/FlinkParquetWriters.java

+
+    @Override
+    public void write(int repetitionLevel, TimestampData value) {
+      column.writeLong(repetitionLevel, value.getMillisecond() * 1000);


This cannot ignore the nanoseconds of the timestamp value.

rdblue · 2020-07-26T19:24:07Z

flink/src/test/java/org/apache/iceberg/flink/data/TestFlinkParquetReaderWriter.java

    }
  }

+  private void assertRowData(Type type, RowData expected, RowData actual) {


Can you move these to a TestHelpers class with methods like assertEqualsInternal for the internal representation?

Also, the tests are missing conversions that discard microseconds because this is generating internal representations (like RowData) and comparing to RowData. These tests should write generics and validate a generic record against a row. See the TestHelpers in Spark.

rdblue · 2020-07-26T19:24:34Z

flink/src/test/java/org/apache/iceberg/flink/data/RandomData.java

  }

-  private static class RandomRowGenerator extends RandomGenericData.RandomDataGenerator<Row> {
+  private static class RandomRowDataGenerator extends TypeUtil.CustomOrderSchemaVisitor<Object> {


I think this should use the generator for generics and validate against generic data. That will catch more cases.

Done. Now it writes generics in parquet files and readout through Flink parquet reader, and compare the Record with RowData.

rdblue · 2020-07-26T19:28:21Z

Thanks for working on this, @chenjunjiedada!

You might consider separating this into read and write PRs to make it easier to review.

chenjunjiedada · 2020-07-27T12:43:53Z

@rdblue , Thanks a lot for your comments. Let me go through the comments and separate the patch.

chenjunjiedada · 2020-07-29T12:55:08Z

@rdblue , I created #1266 for the reader, you may want to take a look. The comments for the reader side are addressed in that PR as well.

chenjunjiedada · 2020-07-30T11:06:07Z

@rdblue, I open #1272 for writer side. The comments for the writer are addressed there. I close this accordingly.

)

chenjunjiedada force-pushed the update-flink-parquet-writer-with-schema-visitor branch 2 times, most recently from 40448b7 to ebcfba9 Compare July 24, 2020 02:09

Update Flink Parquet reader and writer to use schema visitor

ebcfba9

rdblue reviewed Jul 26, 2020

View reviewed changes

rdblue mentioned this pull request Jul 26, 2020

Abstract the generic task writers for sharing the common codes between spark and flink #1213

Merged

fix timestamp precision issue

7671d83

chenjunjiedada mentioned this pull request Jul 29, 2020

Flink: update parquet reader with schema visitor #1266

Merged

chenjunjiedada mentioned this pull request Jul 30, 2020

Flink: use schema visitor for parquet writer #1272

Merged

chenjunjiedada closed this Jul 30, 2020

szehon-ho pushed a commit to szehon-ho/iceberg that referenced this pull request Sep 16, 2024

rdar://124678730: Iceberg S3FileIO Support for Unified Auth (apache#1237

4a13278

)


		BigDecimal bigDecimal = decimal.toBigDecimal();

		byte fillByte = (byte) (bigDecimal.signum() < 0 ? 0xFF : 0x00);

Update Flink Parquet reader and writer to use schema visitor #1237

Update Flink Parquet reader and writer to use schema visitor #1237

Uh oh!

Conversation

chenjunjiedada commented Jul 23, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

rdblue Jul 26, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

chenjunjiedada Jul 28, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

rdblue commented Jul 26, 2020

Uh oh!

chenjunjiedada commented Jul 27, 2020

Uh oh!

chenjunjiedada commented Jul 29, 2020

Uh oh!

chenjunjiedada commented Jul 30, 2020

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

chenjunjiedada commented Jul 23, 2020 •

edited

Loading

rdblue Jul 26, 2020 •

edited

Loading

chenjunjiedada Jul 28, 2020 •

edited

Loading