[FLINK-19694] Support [UPDATE_AFTER, DELETE] ChangelogMode for ScanTableSource

• Type: Sub-task
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Fix Version/s: 1.12.0
• Component/s: Table SQL / API, Table SQL / Planner
• Labels: pull-request-available
• Assignee: Jark Wu
• Reporter: Jark Wu
• Created: 18/Oct/20 14:44
• Updated: 29/Oct/20 09:40
• Resolved: 27/Oct/20 14:13

Description

Currently, we only support insert-only or all-kinds (e.g. CDC format) ChangelogMode. In this issue, we want to support [UPDATE_AFTER, DELETE] ChangelogMode which indicates the source will emit only UPDATE_AFTER and DELETE messages during runtime (e.g. an upsert source). The planner will add the a materialization operator when the ChangelogMode of the source is [UPDATE_AFTER, DELETE].

The materialization operator will materialize the upsert stream and generate changelog stream with full change messages. In the physical operator, we will use state to know whether the key is the first time to be seen. The operator will produce INSERT rows, or additionally generate UPDATE_BEFORE rows for the previous image, or produce DELETE rows with all columns filled with values.

Issue Links

• links to: GitHub Pull Request #13721