[ https://issues.apache.org/jira/browse/FLINK-20289?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17238118#comment-17238118 ]
Nicholas Jiang commented on FLINK-20289: ---------------------------------------- [~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)