Implements the Flink source based on the new FLIP-27 interface

[stevenzwu]

Flink 1.11 release included a new source interface (FLIP-27) that works better with Iceberg

It has two logical components

• A split enumerator that runs on the jobmanager (driver)
• Parallel readers that run on the taskmanagers (workers).

Here are some benefits of the new source interface

• It is a unified interface for both streaming and batch mode.
• Its dynamic split assignment nature can avoid the straggler/outlier problem from the simple static round-robin split assignment (old interface).
• With enumerator It can maintains the embarrassingly parallel DAG, which can help failure recovery with region recovery mode.

Currently the new interface is fairly finalized, and Flink community is still fine tuning some of the implementation/base classes. Flink community is working on implementing Kafka and file source with this new FLIP-27 interface. It will be great if we can also port Iceberg source to the new interface.

We can target the Iceberg 0.11.0 milestone with the following scopes

• Support both static/batch and continuous/streaming enumeration modes
• Support only simple assigner with no ordering/locality guarantee when handing out split assignment. But make the interface flexible to plug in different assigners (like ordered by time or locality aware)
• It will be @Experimental status as nobody has run FLIP-27 sources in production yet. Flink 1.12.0 release (ETA end of Nov) will have the first set of sources (Kafka and file) implemented with FLIP-27 source framework. We still need to gain more production experiences.

[openinx]

Looking forward to the draft PR or Poc patch, Thanks for the work @stevenzwu .

[kbendick]

Thanks @stevenzwu. I'm looking forward to this as well. The new source interface allows OSS to once again chain AsyncWait tasks which has been a pain point, so thank you!

[stevenzwu]

@openinx @kbendick you can find the PoC code here. Both TestIcebergSource and TestIcebergSourceReaderDeletes pass now.

We will prepare a design doc so that the community can discuss important design and interface questions. Main goal is to support pluggable split assigner, which is not reflected in the current PoC code. E.g. here are some possible assignment strategies: simple assigner that provides no ordering or locality guarantee, assigner that provides some ordering guarantee (for backfill), assigner that is locality aware etc.

@kbendick can you elaborate more on what is the pain point with chaining AsyncWait tasks and why FLIP-27 source interface helps? Just interested to learn.

[stevenzwu]

My colleague (@sundargates) and I worked on this design doc: https://docs.google.com/document/d/1q6xaBxUPFwYsW9aXWxYUh7die6O7rDeAPFQcTAMQ0GM/edit#heading=h.z2xhib39g0nm. We would love to get feedbacks from the community. cc @openinx @JingsongLi @kbendick

Regarding the implementation, we are still waiting for the 1.11.3 release, as we need some of the API changes back-ported from 1.12 branch. Even with 1.11.3, we still need to copy the BulkFormat related code from the file source 1.12.

[openinx]

Thanks for the great work, @stevenzwu ! I will take a look in these days.

[tweise]

https://github.com/apache/iceberg/pull/2105

[github-actions[bot]]

This issue has been automatically marked as stale because it has been open for 180 days with no activity. It will be closed in next 14 days if no further activity occurs. To permanently prevent this issue from being considered stale, add the label 'not-stale', but commenting on the issue is preferred when possible.

[github-actions[bot]]

This issue has been closed because it has not received any activity in the last 14 days since being marked as 'stale'