Flink: use schema visitor for parquet writer #1272

chenjunjiedada · 2020-07-30T11:04:03Z

This is sub PR for #1237. I will rebase this once ~~#1266~~ get merged.

openinx · 2020-07-30T14:35:39Z

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

 import org.apache.parquet.schema.Type;

-public class FlinkParquetReaders extends BaseParquetReaders<Row> {
+public class FlinkParquetReaders {


This class seems don't have to be public, only the FlinkParquetReader will access those readers. It also don't need to be accessed by other classes I think.

Make sense to me.

openinx · 2020-07-30T14:38:18Z

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

+    @Override
+    public ParquetValueReader<?> struct(Types.StructType ignored, GroupType struct,
+                                        List<ParquetValueReader<?>> fieldReaders) {
+      // the expected struct is ignored because nested fields are never found when the


nit: the comment is not complete ?

openinx · 2020-07-30T14:43:12Z

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

+              return new ParquetValueReaders.UnboxedReader<>(desc);
+            }
+          case TIME_MICROS:
+            return new TimeMillisReader(desc);


Q: is there any problem here ? the original type is TIME_MICROS, while the reader name is TimeMillisReader ?

This is because Flink only supports milliseconds and the parquet store microseconds, so the naming express that it reads out milliseconds.

I agree this is confusing. There are other places where we use a unit in the class name to indicate the unit being read. Instead, let's be more specific and use something like LossyMicrosToMillisTimeReader.

openinx · 2020-07-30T14:48:43Z

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

+      }
+    }
+
+    protected MessageType type() {


Will any subclass of ReadBuilder access the message type ?

Previously, the FallbackReader uses it. Now I think this could be removed since the fallback reader defines its own type . That is because we can't get the type from passing builder.

openinx · 2020-07-30T15:10:54Z

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

    };
  }

+  private static Iterable<Record> generateIcebergGenerics(Schema schema, int numRecords,


Seems it could share the common code with RandomGenericData#generate ? Make the RandomGenericData#generate to return a Iterable ?

You are right, let me refactor this.

This method accepts a Record supplier and then generate records. We should keep it for generating fallback records and dictionary encoded records. But for generateRecords method we can update it to call RandomGenericData#generate directly.

openinx · 2020-07-30T15:26:34Z

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

+    }
+  }
+
+  private static class TimeMicrosWriter extends ParquetValueWriters.PrimitiveWriter<Integer> {


The reader is named TimeMillisReader, and the writer is TimeMicrosWriter, could them be symmetrical ?

The naming logic is what we actually perform. In the reader side, we read in the milliseconds for Flink. In the writer side, we write out microseconds for Parquet.

openinx · 2020-07-30T15:29:38Z

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

+    public void write(int repetitionLevel, DecimalData decimal) {
+      Preconditions.checkArgument(decimal.scale() == scale,
+          "Cannot write value as decimal(%s,%s), wrong scale: %s", precision, scale, decimal);
+      Preconditions.checkArgument(decimal.precision() <= precision,


Seem the upper bound of precision of IntegerDecimalWriter is 9 ? Could we add the precision <= 9 assertion ?

Will use the latest DecimalUtil.

Seems DecimalUtil doesn't handle this. I fixed in the new commit.

openinx · 2020-07-30T15:30:21Z

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

+    public void write(int repetitionLevel, DecimalData decimal) {
+      Preconditions.checkArgument(decimal.scale() == scale,
+          "Cannot write value as decimal(%s,%s), wrong scale: %s", precision, scale, decimal);
+      Preconditions.checkArgument(decimal.precision() <= precision,


Also could we add the precision <= 18 assertion ?

How about adding this when allocating the writer? Seems like that would be a suitable place since here we are checking Flink type.

@openinx, any comment?

I think it would be better to do this in the constructor, like @chenjunjiedada suggests. That way we have a check that precision is not larger than the maximum allowed by the type, and that the correct writer is used for the type.

chenjunjiedada · 2020-07-31T02:31:35Z

@openinx, Thanks a lot for your comments. Will rebase and update PRs.

openinx · 2020-08-12T08:02:09Z

flink/src/main/java/org/apache/iceberg/flink/RowTaskWriterFactory.java

    @Override
    public FileAppender<Row> newAppender(OutputFile outputFile, FileFormat format) {
      MetricsConfig metricsConfig = MetricsConfig.fromProperties(props);
+      LogicalType logicalType = FlinkSchemaUtil.convert(schema);


This should have simiar issue to the comment, which will break the unit test. If we rebase the master once #1320 get merged, then it should have no problem.

BTW, we may also need to add the parquet into the parameterized unit tests, such as TestIcebergStreamWriter & TestTaskWriters.

Agreed, will take a look when these PRs get in.

openinx · 2020-08-12T08:15:29Z

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

    };
  }

+  private static Iterable<RowData> generateRowData(Schema schema, int numRecords,


We could use RandomRowData#generate when rebasing the patch https://github.com/apache/iceberg/pull/1320/files#diff-4b2a9fd76495497db9212d74bf03f671R33.

openinx · 2020-08-12T08:32:09Z

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

+    public void write(int repetitionLevel, DecimalData decimal) {
+      Preconditions.checkArgument(decimal.scale() == scale,
+          "Cannot write value as decimal(%s,%s), wrong scale: %s", precision, scale, decimal);
+      Preconditions.checkArgument(decimal.precision() <= 9,


Seems it should be decimal.precision <= precision ?

Seems like I misunderstood your comments, let me update this.

openinx · 2020-08-12T08:32:47Z

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

+    public void write(int repetitionLevel, DecimalData decimal) {
+      Preconditions.checkArgument(decimal.scale() == scale,
+          "Cannot write value as decimal(%s,%s), wrong scale: %s", precision, scale, decimal);
+      Preconditions.checkArgument(decimal.precision() <= 18,


openinx · 2020-08-12T08:36:21Z

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

+    private class ElementIterator<E> implements Iterator<E> {
+      private final int size;
+      private final ArrayData list;
+      private int index;
+
+      private ElementIterator(ArrayData list) {
+        this.list = list;
+        size = list.size();
+        index = 0;
+      }
+
+      @Override
+      public boolean hasNext() {
+        return index != size;
+      }
+
+      @Override
+      @SuppressWarnings("unchecked")
+      public E next() {
+        if (index >= size) {
+          throw new NoSuchElementException();
+        }
+
+        E element;
+        if (list.isNullAt(index)) {
+          element = null;
+        } else {
+          element = (E) ArrayData.createElementGetter(elementType).getElementOrNull(list, index);
+        }
+
+        index += 1;
+
+        return element;
+      }
+    }


How about moving this ElementIterator to be a static class, then the map's EntryIterator could share it ? Seems we could do it, you could decide wether there is necessary.

I 'm not sure how can it be shared with EntryIterator.

openinx · 2020-08-12T08:42:11Z

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

+import org.apache.parquet.schema.PrimitiveType;
+import org.apache.parquet.schema.Type;
+
+public class ParquetWithFlinkSchemaVisitor<T> {


TODO: we could share both flink and spark ParquetSchemaVisitor in a common class , can be a separate issue.

Agreed, I would prefer to do the refactor in a separated PR.

Yes, a WithPartner visitor like @JingsongLi added would be great.

rdblue · 2020-08-12T19:47:30Z

From other comments, it sounds like I should review ~~#1320~~ first and then this will be rebased. I also reviewed the read side, which can be done in parallel.

chenjunjiedada · 2020-08-20T09:31:53Z

@rdblue @openinx, Just rebased this and also added the follow-up from the reader side. Please take a look at your convenience.

rdblue · 2020-08-20T16:40:01Z

@chenjunjiedada, looks like this is conflicting again. Must have been one of the patches I merged this morning. Sorry about that!

I'll take a look at this one next, thanks for your patience with reviews. I've been running behind on reviews lately.

chenjunjiedada · 2020-08-21T01:16:51Z

@rdblue , Never mind, it is just a small conflict that already fixed. Take your time.

rdblue · 2020-08-25T00:26:36Z

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

+    }
+
+    @Override
+    public ParquetValueWriter<?> primitive(LogicalType sType, PrimitiveType primitive) {


Nit: s in sType indicates Spark. The equivalent here would be fType or a better name.

rdblue · 2020-08-25T00:32:19Z

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

+
+    @Override
+    public void write(int repetitionLevel, Integer value) {
+      long micros = Long.valueOf(value) * 1000;


This conversion from Integer doesn't make much sense. Java exposes 2 valueOf with string arguments and one with a primitive long argument. The last is what is called here. In that case, this is implicitly casting Integer to long, boxing the result, and then multiplying to produce a primitive.

It would be better to use value.longValue() * 1000 instead.

rdblue · 2020-08-25T00:37:23Z

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

+        if (list.isNullAt(index)) {
+          element = null;
+        } else {
+          element = (E) ArrayData.createElementGetter(elementType).getElementOrNull(list, index);


This method is called in a tight loop, so for performance any preparation that can be done in advance should be.

That means this getter should be created in the constructor and stored as an instance field. Then it can be called here.

Also, if there is already a null check above, does this need to call getElementOrNull or should it just call a get variant that assumes the value is non-null?

Alternatively, you could replace the if here:

E element = (E) getter.getElementOrNull(list, index);

That means this getter should be created in the constructor and stored as an instance field. Then it can be called here.

Yeah, that sounds good to me, great point.

does this need to call getElementOrNull or should it just call a get variant that assumes the value is non-null?
The getter in ArrayData don't have a get interface, it have only the interface:

/** * Accessor for getting the elements of an array during runtime. * * @see #createElementGetter(LogicalType) */ interface ElementGetter extends Serializable { @Nullable Object getElementOrNull(ArrayData array, int pos); }

Replacing the if-else to be E element = (E) getter.getElementOrNull(list, index); sounds reasonable to me.

rdblue · 2020-08-25T00:38:03Z

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

+        }
+
+        if (values.isNullAt(index)) {
+          entry.set((K) ArrayData.createElementGetter(keyType).getElementOrNull(keys, index), null);


Same here. The getters for keys and values should be instance fields.

Keys are not allowed to be null, so there should be no need to call getElementOrNull for the key.

rdblue · 2020-08-25T00:38:57Z

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

-    protected Object get(Row row, int index) {
-      return row.getField(index);
+    protected Object get(RowData struct, int index) {
+      return RowData.createFieldGetter(types.get(index), index).getFieldOrNull(struct);


Each getter should be stored as a field in an array.

rdblue · 2020-08-25T00:42:36Z

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

-      final int fieldPos = i;
      assertEquals(types.get(i), logicalType, expected,
-          () -> RowData.createFieldGetter(logicalType, fieldPos).getFieldOrNull(actualRowData));
+          RowData.createFieldGetter(logicalType, i).getFieldOrNull(actualRowData));


Thanks for fixing these.

rdblue · 2020-08-25T00:45:28Z

@chenjunjiedada, I'm going to merge this. The remaining issues are minor or are not correctness problems. Just be sure to follow up and fix the getter problems or else this will be slower than it should be.

rdblue · 2020-08-25T00:46:11Z

Thanks, @chenjunjiedada for building this, and @openinx for reviewing!

chenjunjiedada mentioned this pull request Jul 30, 2020

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

Closed

openinx reviewed Jul 30, 2020

View reviewed changes

This was referenced Aug 7, 2020

Flink: Refactor to replace Row type with RowData type in write path. #1305

Closed

Flink: Replace Row with RowData in flink write path. #1320

Merged

chenjunjiedada force-pushed the use-schema-visitor-for-flink-writer branch from e78c2ec to f0641b4 Compare August 11, 2020 13:35

openinx reviewed Aug 12, 2020

View reviewed changes

chenjunjiedada force-pushed the use-schema-visitor-for-flink-writer branch from 940e435 to 09c1b41 Compare August 12, 2020 11:31

chenjunjiedada added 2 commits August 20, 2020 11:36

Flink: use schema visitor for flink parquet writer

c0ac93a

fix checkstyle

a73a7d7

chenjunjiedada force-pushed the use-schema-visitor-for-flink-writer branch from 09c1b41 to a73a7d7 Compare August 20, 2020 06:18

probot-autolabeler bot added the flink label Aug 20, 2020

minor fixes

fabf81d

openinx mentioned this pull request Aug 20, 2020

Flink: Add the iceberg files committer to collect data files and commit to iceberg table. #1185

Merged

fix RowDataConverter

7aebde0

Merge branch 'master' into use-schema-visitor-for-flink-writer

caa4ca3

rdblue reviewed Aug 25, 2020

View reviewed changes

rdblue merged commit 6a7c0db into apache:master Aug 25, 2020

chenjunjiedada deleted the use-schema-visitor-for-flink-writer branch August 25, 2020 02:46

chenjunjiedada mentioned this pull request Aug 25, 2020

Flink: optimize the parquet writer getter method #1377

Merged

chenjunjiedada mentioned this pull request Nov 4, 2020

Parquet: Add type visitor with partner type #1391

Closed

Flink: use schema visitor for parquet writer #1272

Flink: use schema visitor for parquet writer #1272

Uh oh!

Conversation

chenjunjiedada commented Jul 30, 2020 • edited by rdblue 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!

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!

chenjunjiedada commented Jul 31, 2020

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 commented Aug 12, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

chenjunjiedada commented Jul 30, 2020 •

edited by rdblue

Loading

rdblue commented Aug 12, 2020 •

edited

Loading