[
https://issues.apache.org/jira/browse/FLINK-20289?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17238118#comment-17238118
]
Nicholas Jiang edited comment on FLINK-20289 at 11/26/20, 3:20 AM:
-------------------------------------------------------------------
[~jark], is this improvement that all computed column expressions is pushed
down after ChangelogNormalize?
IMO, I could try to work for the calculation of computed columns after
ChangelogNormalize.
was (Author: nicholasjiang):
[~jark], is this improvement that all computed column expressions is pushed
down after ChangelogNormalize? IMO, I could try to work for the calculation of
computed columns after ChangelogNormalize.
> Computed columns can be calculated after ChangelogNormalize to reduce shuffle
> -----------------------------------------------------------------------------
>
> Key: FLINK-20289
> URL: https://issues.apache.org/jira/browse/FLINK-20289
> Project: Flink
> Issue Type: Improvement
> Components: Table SQL / Planner
> Reporter: Jark Wu
> Priority: Major
>
> In FLINK-19878, we improved that the ChangelogNormalize is applied after
> WatermarkAssigner to make the watermark to be close to the source. This helps
> the watermark to be more fine-grained.
> However, in some cases, this may shuffle more data, because we may apply all
> computed column expressions before ChangelogNormalize. As follows, {{a+1}}
> can be applied after ChangelogNormalize to reduce the shuffles.
> {code:sql}
> CREATE TABLE src (
> id STRING,
> a INT,
> b AS a + 1,
> c STRING,
> ts as to_timestamp(c),
> PRIMARY KEY (id) NOT ENFORCED,
> WATERMARK FOR ts AS ts - INTERVAL '1' SECOND
> ) WITH (
> 'connector' = 'values',
> 'changelog-mode' = 'UA,D'
> );
> SELECT a, b, c FROM src WHERE a > 1
> {code}
> {code}
> Calc(select=[a, b, c], where=[>(a, 1)], changelogMode=[I,UB,UA,D])
> +- ChangelogNormalize(key=[id], changelogMode=[I,UB,UA,D])
> +- Exchange(distribution=[hash[id]], changelogMode=[UA,D])
> +- WatermarkAssigner(rowtime=[ts], watermark=[-(ts, 1000:INTERVAL
> SECOND)], changelogMode=[UA,D])
> +- Calc(select=[id, a, +(a, 1) AS b, c, TO_TIMESTAMP(c) AS ts],
> changelogMode=[UA,D])
> +- TableSourceScan(table=[[default_catalog, default_database,
> src]], fields=[id, a, c], changelogMode=[UA,D])
> {code}
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
