[FLINK-22899] ValuesUpsertSinkFunction needs to use global upsert

• Type: Sub-task
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Fix Version/s: 1.14.0, 1.13.3
• Component/s: Table SQL / Runtime
• Labels: pull-request-available
• Assignee: Jingsong Lee
• Reporter: Jingsong Lee
• Created: 07/Jun/21 08:51
• Updated: 16/Sep/21 09:44
• Resolved: 17/Jun/21 02:28

Description

At present, each task does its own upsert. We need to simulate the external connector and use the global map to do the upsert.

Issue Links

• relates to: FLINK-20374 Wrong result when shuffling changelog stream on non-primary-key columns
• links to: GitHub Pull Request #16091