Flink RowDataIterator should support reuseObject flag

[stevenzwu]

While debugging unit test failure for the FLIP-27 source PoC code, I realized that RowDataIterator always reuse and return the same GenericRowData object. It seems to expect caller to clone or convert the object. TestFlinkInputFormat was using TestHelpers.copyRowData to do it. I guess Flink SQL convert RowData to Row.

This behavior makes it difficult for the SplitReader to use the iterator, because SplitReader fetches a batch of records.
https://github.com/stevenzwu/iceberg/blob/flip27IcebergSource/flink/src/main/java/org/apache/iceberg/flink/source/reader/IcebergSourceSplitReader.java#L80-L87

Should we make the RowDataIterator support reuse flag?

Also what should be the default reuse value (true or false)? When I was looking at this code in FlinkInputFormat, it seems that base InputFormat class expects the iterator to only return reused object, if reuse input arg is not null. So the iterator.next() seems to break that contract.

  @Override
  public RowData nextRecord(RowData reuse) {
    return iterator.next();
  }

@JingsongLi @openinx please share your thoughts.

[openinx]

In most flink cases, the FlinkInputFormat will read a record and emit it to the downstream operator, that means it will serialize the RowData and then sends bytes to next operator, so I think it's right to set reuse=true by default, that saves lots of object allocation from JVM.

In your batch-read case, the ideal way should be : allocating a fixed-size array which should be reused in every batch read, when reading a given record we should pass the relative element from reused array for reusing purpose. But I read the code, seems the newAvroIterable, newParquetIterable, newOrcIterable will return a Iterable whose next() method has no way to pass a reused instance for reusing. we have to do the RowData copying (Copy RowData from iterator to the fixed-size array) but we still could reuse the fixed-size array to avoiding allocating too many young objects.

Does that make sense ?

[stevenzwu]

Let's look at the nextRecord contract from the InputFormat. reuse or not is controlled by the caller of nextRecord(OT reuse).

	/**
	 * @param reuse Object that may be reused.
	 */
	OT nextRecord(OT reuse) throws IOException;

If the input reuse arg is null, there is no reuse. Otherwise, input arg should be used for deserialization. That is how BinaryInputFormat implemented.

	@Override
	public T nextRecord(T record) throws IOException {
                // ...
		record = this.deserialize(record, this.dataInputStream);
                // ...
	}

CloseableIterable<T> interface is just NOT a good fit for InputFormat interface. Now the FlinkInputFormat ignore the reuse hint from caller and simply returns reused RowData object.

  @Override
  public RowData nextRecord(RowData reuse) {
    return iterator.next();
  }

So technically, FlinkInputFormat is not conforming to the contract of InputFormat interface. If we truly want to implement the correct nextRecord contract, we need to change the CloseableIterable/Iterator to support next(T reuse), which is impossible to change the java Iterator interface. If it can't conform to the contract, maybe the default behavior should be the more conservative (and less performant) behavior of no reuse.

We don't use InputFormat internally. So its behavior is not very relevant to us. This is purely a semantic discussion. I will address other issues in separate comments.

[stevenzwu]

Regarding copying RowData from iterator to a fixed-size array by the IcebergSourceSplitReader<T>, let's first discuss some of the implementation details.

• How would IcebergSourceSplitReader<T> know how to create a reuseable empty object array for a generic type T?
• How would IcebergSourceSplitReader<T> know how to clone the generic type T returned by DataIterator.next()? Both Flink RowData and Avro GenericRecord don't implement Cloneable interface.

RowDataIterator and its underneath Flink*Reader have the knowledge on how to achieve the correct object reuse semantic.

Assuming IcebergSourceSplitReader<T> can implement proper cloning to reusable array, we should still discuss the reuse semantic for records emitted by Iceberg source. Not sure if Flink defines the reuse semantic for the records emitted by source or it is up to the source implementation. If it is up to the source and we decided that Iceberg source should only emit reused records, we should clearly document the expectation.

E.g. users could have a chained downstream operator (no serialization) that performs de-dup aggregation in memory. When it first time saw a key, it keep the record/row for the key. If there is another row coming for the same key, it simply update a field in the first received row.

[stevenzwu]

• How would IcebergSourceSplitReader know how to create a reuseable empty object array for a generic type T?
• How would IcebergSourceSplitReader know how to clone the generic type T returned by DataIterator.next()?

We already added a DataIteratorFactory on how to create an iterator from split. I think we can extend the interface on how create and clone objects.

public class RowDataIteratorFactory implements DataIteratorFactory<RowData> {
  @Override
  public DataIterator<RowData> createIterator(
      IcebergSourceSplit split,
      TableInfo tableInfo,
      ScanContext scanContext) {
    return new RowDataIterator(...);
  }

The unclear question for is still what should be reuse behavior for Flink source.

[openinx]

The unclear question for is still what should be reuse behavior for Flink source.

I read the DataSourceTask code, there's a objectReuse ( default value is false) from ExecutionConfig to indicate whether we should enable object reuse. I also checked the RecordWriter, for every emited record, it would serialized into a ByteBuffer . So in theory, regardless of enabling or disabling object reuse, it actually would copy the object into a totally new ByteBuffer, we don't have to worry about the objects messing up issue caused by reuse at flink runtime at least.

[openinx]

I think we have reached an agreement that it's useful to reuse objects in the row data iterators, the current problem is : how to reuse it ?

Now we RichInputFormat#nextRecord(T reuse) have provided the reuse instance for reusing, but our parquet, orc, avro have an internal object reuse mechanism. Unless we change the Iteratable interface to accept the reuse from RichInputFormat, we couldn't bridge them into a single reuse instance.

So another way I can image is copying it from the internal parquet, orc, avro iterator's reuse to the RichInputFormat's reuse . But as you said, for a generic type T, it does not expose a Cloneable method to accomplish the copying. In that way, we may need a copy function similar to Cloneable when constructing the FlinkInputFormat ?

[stevenzwu]

We actually enable objectReuse by default so that chained operators can avoid the serialization and deserialization cost, which is huge for embarrassingly parallel DAGs. That is mainly for operator chaining.

RecordWriter is used for network shuffle (like keyBy).

Both probably shouldn't affect or dictate the object reuse semantic in source emitted records.

BTW, Iceberg's ParquetAvroValueReaders returns a new Avro GenericRecord object for each iterator.next() call.

[stevenzwu]

the current problem is : how to reuse it ?

regarding how to create and clone instance, we can extend the new DataIteratorFactory interface in the flip-27 source poc code. So that is doable

public class RowDataIteratorFactory implements DataIteratorFactory<RowData> {

  @Override
  public DataIterator<RowData> createIterator(
      IcebergSourceSplit split,
      TableInfo tableInfo,
      ScanContext scanContext) {
    return new RowDataIterator(...);
  }

  @Override
  public RowData create() {
  ...
  }

  @Override
  public void clone(RowData from, RowData to) {
  ...
  }

[stevenzwu]

it's useful to reuse objects in the row data iterators

Yeah. I am fine either way. My main question is for the records emitted by Iceberg source (FLIP-27) should reuse objects or not.

[openinx]

I'd love to comment the patch if you plan to update it and propose a new one for the reuse idea.

[stevenzwu]

@openinx I updated the patch. You can find the iterator factory change here. Right now, it is just doing simple clone if iterator returns reused object. for Avro iterator (which already returns new objects), we don't need to clone again.

BTW, some unit tests still fail. so it is not completely working.

[JingsongLi]

I think a file like connector like iceberg can use an interface like BulkFormat:
https://github.com/apache/flink/blob/2da55b8f841ba93e1aa7c56de31e6c5632d725ba/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/reader/BulkFormat.java

Here is a parquet format example:
https://github.com/apache/flink/blob/8da968e8d0b1bb13e79d227236f814db69eb31a7/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetVectorizedInputFormat.java

I think in Iceberg, we can integrate Flink reader to vectorized parquet/orc reader too. This should be the good performance reader.

[JingsongLi]

For the row-based reader, can RowDataIteratorFactory just provide a createIterator(...., boolean objectReuse)? Is this enough?

[stevenzwu]

For the row-based reader, can RowDataIteratorFactory just provide a createIterator(...., boolean objectReuse)? Is this enough?

@JingsongLi I really like the suggestion, which hides the reuse behavior within the iterator. It is a better abstraction that doesn't leak the reuse/clone logic to the caller or factory. It is actually what I was originally looking for as the title suggested.

[stevenzwu]

I took a look at the BulkFormat interface. It does make sense for the Flink reader to use BulkFormat abstraction (than the CloseableIterable). Is there any plan to add Avro and Orc BulkFormat impl in Flink?

unclear about the integration effort required with Iceberg schema etc.