[FLINK-8578] Implement rowtime DataStream to Table upsert conversion.

• Type: Sub-task
• Status: Closed
• Priority: Major
• Resolution: Duplicate
• Component/s: Table SQL / API
• Assignee: Hequn Cheng
• Reporter: Hequn Cheng
• Created: 07/Feb/18 10:47
• Updated: 07/Dec/20 09:16
• Resolved: 07/Dec/20 09:16

Description

Flink-8577 implements upsert from stream under proctime. This task is going to solve the order problem introduce by proctime. As proposed by Fabian in FLINK-8545, it would be good to be able to declare a time attribute that decides whether an upsert is performed or not.

Table table = tEnv.upsertFromStream(input, 'a, 'b.rowtime.upsertOrder, 'c.key)

This is a good way to solve the order problem using rowtime. And an idea comes to my mind that we can even remove the `.upsertOrder`, because the rowtime attribute can only be defined once in a table schema. Removing `.upsertOrder` also makes it easier to design api for TableSource and sql, i.e, we don't need to add another new feature for the api.

Any suggestions are welcomed!