Flink: Add Orc value reader, writer implementations #1158

openinx · 2020-07-02T04:09:15Z

The unit tests are passed now, but seems it did not report back to this issue (https://travis-ci.org/github/apache/iceberg/builds/704218624). Ping @rdblue for reviewing.
Thanks.

rdblue · 2020-07-03T16:28:00Z

@shardulm94 and @rdsr, could you help review this?

rdsr · 2020-07-03T16:32:34Z

I'll have a look this week, thks!

openinx · 2020-07-08T04:07:23Z

@rdsr Please help to take a look when you have time , Thanks.

rdsr · 2020-07-08T06:58:20Z

data/src/main/java/org/apache/iceberg/data/orc/BaseOrcReader.java

+import org.apache.iceberg.types.Types;
+import org.apache.orc.TypeDescription;
+
+public abstract class BaseOrcReader<T> implements OrcRowReader<T> {


It seems that we are extending the GenericOrcReader to also be used in Flink. Won't that create problems? E.g GenericOrcReader is being used to construct a readerFunc for IcebergGenerics . For instance it uses LocalTime for Iceberg's Time datatype. Won't Flink have its own in-memory representation for primitive types and maybe also for map and list types?

I think it will be better to have a completely separate FlinkOrcReader which does not rely on GenericOrcReader similar to SparkOrcReader. In this way changes to GenericOrcReader won't break FlinkOrcReader and there is no tight coupling between the two.

@rdblue , @openinx thoughts?

In current flink stable version, Flink is using the Row type with an array of Java objects, it's the most common way for flink now. In feature, it will use RowData interface , whose implementation could be binary-oriented or java object oriented, I think in that time we could separate the FlinkOrcReader. (issue: https://issues.apache.org/jira/browse/FLINK-16995).

What about Flink's primitive types do they align with well with Iceberg Generics?

I think I agree with @rdsr - main concern would be the flexibility for changes to GenericOrcReader, but I guess the trade-off is vs code re-usability. If there's confidence that the generics readers are fairly stable then it should not be a huge issue, but the concern seems valid on coupling these readers. I wonder if instead of using inheritance a delegator approach would be possible avoid a tight coupling.

What about Flink's primitive types do they align with well with Iceberg Generics?

My understanding is that Flink does (or can) use the same representations, except for structs. It would be good to have a response for @openinx or @JingsongLi, though. From looking at the Flink code, not all of the default conversions are these types. VarBinary uses byte[] instead of ByteBuffer and LocalZonedTimestampType uses Instant (but the Javadoc says its behavior is like OffsetDateTime that we use). That said, it looks like Flink might support multiple conversions.

Depending on what Flink uses internally, @rdsr might be right about building a set of readers specific to those types. But if we can make this more generic easily, then I like the idea of doing that. Ideally, I think new object models would be created by providing a few methods to create and read into an object, kind of like our methods to plug in struct types.

Yea, I also think the GenericOrcReader is a pretty small wrapper and the bulk of the functionality is provided by the readers/functions for specific types defined in OrcGenericReaders. In that regard extending the GenericOrcReader doesn't buy us much. We can easily share code by picking and choosing the right readers/functions from GenericOrcReaders and providing flink specific type readers where flink types diverge from Iceberg Generics. The good thing about doing this IMO is that we get rid of extending classes which makes code changes brittle and introduces tight coupling

I agreed that extending the BaseOrcReader and BaseOrcWriter introduces tight coupling, I tried to de-couple flink writer from generic orc writers and let them share the common writers. But seems it's hard to share the codes because we used a static buildConverter method to build the converter for each data type and few Converter depends on the static buildConverter, makes hard to abstract to the common converters. Just curious why did we implement the orc writer in converter way instead of visiting the types by OrcSchemaWithTypeVisitor and generate relative OrcRowWriter (in this way we could share most of the writers.), the current converter seems strange compared to other parquet writers and avro writers.

@edgarRd @rdsr @rdblue I did a refactor for the GenericOrcWriter and moved the common writers to GenericOrcWriter, the pull request is here: https://github.com/apache/iceberg/pull/1197/files. Mind to take a look ?

@chenjunjiedada opened an issue for the concern about data types and @JingsongLi clarified the types that Flink uses there. @rdsr was right and it isn't correct to copy generics with a different row type.

Sounds like #1197 is a good start. We should probably reverse how we have refactored the Avro and Parquet generics as well.

Yes, you are right. After the #1197 get merged, I will recreate this patch for reviewing. Thanks.

rdsr · 2020-07-09T15:21:53Z

data/src/main/java/org/apache/iceberg/data/orc/BaseOrcWriter.java

+  }
+
+  /**
+   * The interface for the conversion from Spark's SpecializedGetters to


nit: Remove Spark from comments

rdsr · 2020-07-09T15:22:16Z

data/src/main/java/org/apache/iceberg/data/orc/BaseOrcWriter.java

+    Class<T> getJavaClass();
+
+    /**
+     * Take a value from the Spark data value and add it to the ORC output.


here as well

rdsr

Overall looks ok to me, but I'm concerned regarding coupling of the GenericOrc[Reader|Writer] and FlinkOrc[Reader|Writer].

rdsr · 2020-07-09T19:14:48Z

data/src/main/java/org/apache/iceberg/data/orc/BaseOrcWriter.java

+    }
+  }
+
+  protected abstract Converter<T> createStructConverter(TypeDescription schema);


It seems apart from Struct, flink will use the same in-memory objects for map, list and primitive types?

edgarRd · 2020-07-09T16:56:30Z

data/src/main/java/org/apache/iceberg/data/orc/BaseOrcWriter.java

+public abstract class BaseOrcWriter<T> implements OrcValueWriter<T> {
+  private final Converter[] converters;
+  private static final OffsetDateTime EPOCH = Instant.ofEpochSecond(0).atOffset(ZoneOffset.UTC);
+  private static final LocalDate EPOCH_DAY = EPOCH.toLocalDate();


I think you can use DateTimeUtil.EPOCH and DateTimeUtil.EPOCH_DAY instead.

edgarRd · 2020-07-09T20:07:54Z

data/src/main/java/org/apache/iceberg/data/orc/BaseOrcReader.java

+import org.apache.iceberg.types.Types;
+import org.apache.orc.TypeDescription;
+
+public abstract class BaseOrcReader<T> implements OrcRowReader<T> {


I think I agree with @rdsr - main concern would be the flexibility for changes to GenericOrcReader, but I guess the trade-off is vs code re-usability. If there's confidence that the generics readers are fairly stable then it should not be a huge issue, but the concern seems valid on coupling these readers. I wonder if instead of using inheritance a delegator approach would be possible avoid a tight coupling.

rdblue · 2020-07-11T00:29:24Z

data/src/main/java/org/apache/iceberg/data/orc/GenericOrcReader.java

  @Override
  public Record read(VectorizedRowBatch batch, int row) {
-    return (Record) reader.read(new StructColumnVector(batch.size, batch.cols), row);
+    return (Record) getReader().read(new StructColumnVector(batch.size, batch.cols), row);


Getter methods should not start with get. It doesn't add anything (every method "gets" its return value) and is not idiomatic in other JVM languages.

rdblue · 2020-07-11T00:31:05Z

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

+  @Override
+  @SuppressWarnings("unchecked")
+  public void write(Row value, VectorizedRowBatch output) {
+    int row = output.size++;


We avoid using return values from ++ expressions. That helps readability because statement order is clear.

rdblue · 2020-07-11T00:35:00Z

Looks like there are some minor things to clean up, but overall the code changes are close to being ready.

Before merging this, I'd like to understand how Flink will work with the data that is produced so we can make a good decision about whether we should continue with ORC like we have for Parquet and Avro (sharing generics code) or whether we should think about building separate readers and writers for its object model. Thanks for bringing this up, @rdsr!

openinx · 2020-07-27T07:48:30Z

According to the issue #1215, we've upgraded the flink to 1.11 version and planed to support the RowData avro, parquet, orc readers and writers, so I will create a new pull request with RowData implementation. Close this one now.

…ies (apache#1158)

Flink: Add Orc value reader, writer implementations

0520b10

openinx mentioned this pull request Jul 2, 2020

Implement the parquet value reader & writer for apache flink #1125

Merged

Rename the createRecordReader as createStructReader

97c31e4

rdblue requested a review from rdsr July 3, 2020 16:27

rdblue added this to the Flink Sink milestone Jul 7, 2020

rdsr reviewed Jul 8, 2020

View reviewed changes

rdsr reviewed Jul 9, 2020

View reviewed changes

edgarRd reviewed Jul 9, 2020

View reviewed changes

rdblue reviewed Jul 11, 2020

View reviewed changes

openinx closed this Jul 27, 2020

openinx deleted the flink-orc branch July 27, 2020 07:48

openinx mentioned this pull request Jul 29, 2020

Refactor the GenericOrcWriter by using OrcSchemaWithTypeVisitor#visit #1197

Merged

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

Internal: Remove Check for non-null Sequence Number for Manifest Entr…

6087954

…ies (apache#1158)

rodmeneses pushed a commit to rodmeneses/iceberg that referenced this pull request Jun 23, 2025

Internal: Remove Check for non-null Sequence Number for Manifest Entr…

775379a

…ies (apache#1158)

Flink: Add Orc value reader, writer implementations #1158

Flink: Add Orc value reader, writer implementations #1158

Uh oh!

Conversation

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

Uh oh!

Uh oh!

rdblue commented Jul 3, 2020

Uh oh!

rdsr commented Jul 3, 2020

Uh oh!

openinx commented Jul 8, 2020

Uh oh!

rdsr Jul 8, 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!

rdsr Jul 11, 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!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

rdsr left a comment

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 Jul 11, 2020

Uh oh!

openinx commented Jul 27, 2020

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

openinx commented Jul 2, 2020 •

edited

Loading

rdsr Jul 8, 2020 •

edited

Loading

rdsr Jul 11, 2020 •

edited

Loading