[
https://issues.apache.org/jira/browse/FLINK-19878?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Leonard Xu updated FLINK-19878:
-------------------------------
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.
{code:java}
+- 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])
{code}
As an improvement, we can move the WatermarkAssigner to be after the SourceCan
Node and thus the watermark will produce like general Source.
was:
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.
As an improvement, we can move the WatermarkAssigner to be after the SourceCan
Node and thus the watermark will produce like general Source.
{code:java}
+- 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])
{code}
> Improve watermark for upsertSource
> -----------------------------------
>
> Key: FLINK-19878
> URL: https://issues.apache.org/jira/browse/FLINK-19878
> Project: Flink
> Issue Type: Sub-task
> Components: Table SQL / Planner
> Reporter: Leonard Xu
> Priority: Major
>
> 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.
>
> {code:java}
> +- 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])
> {code}
>
> As an improvement, we can move the WatermarkAssigner to be after the
> SourceCan Node and thus the watermark will produce like general Source.
>
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
