[FLINK-19878] Fix WatermarkAssigner shouldn't be after ChangelogNormalize

• Type: Sub-task
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Fix Version/s: 1.12.0
• Component/s: Table SQL / Planner
• Labels: pull-request-available
• Assignee: Jark Wu
• Reporter: Leonard Xu
• Created: 29/Oct/20 09:32
• Updated: 23/Nov/20 12:47
• Resolved: 23/Nov/20 12:47

Description

Cutrrently, for a upsertSource like upsert-kafka, the WatermarkAssigner is followed after ChangelogNormalize Node,  it may returns Long.MaxValue as watermark if some parallelism doesn't have data. 

 

   +- Exchange(distribution=[hash[currency]], changelogMode=[I,UA,D])
      +- WatermarkAssigner(rowtime=[rowtime], watermark=[rowtime], changelogMode=[I,UA,D])
         +- ChangelogNormalize(key=[currency], changelogMode=[I,UA,D])
            +- Exchange(distribution=[hash[currency]], changelogMode=[UA,D])
               +- TableSourceScan(table=[[default_catalog, default_database, rates_history]], fields=[currency, rate, rowtime], changelogMode=[UA,D])

 

As an improvement, we can move the WatermarkAssigner to be after the SourceCan Node and thus the watermark will produce like general Source.

 

Issue Links

• links to: GitHub Pull Request #14154