[ https://issues.apache.org/jira/browse/FLINK-20909?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Jark Wu updated FLINK-20909: ---------------------------- Priority: Major (was: Minor) > MiniBatch Interval derivation does not work well when enable miniBatch > optimization in a job which contains deduplicate on row and unbounded > aggregate. > ------------------------------------------------------------------------------------------------------------------------------------------------------- > > Key: FLINK-20909 > URL: https://issues.apache.org/jira/browse/FLINK-20909 > Project: Flink > Issue Type: Improvement > Components: Table SQL / Planner > Reporter: Andy > Priority: Major > > MiniBatch Interval derivation does not work well when enable miniBatch > optimization in a job which contains deduplicate on row and unbounded > aggregate. > {code:java} > @Test > def testLastRowOnRowtime1(): Unit = { > val t = env.fromCollection(rowtimeTestData) > .assignTimestampsAndWatermarks(new RowtimeExtractor) > .toTable(tEnv, 'a, 'b, 'c, 'rowtime.rowtime()) > tEnv.registerTable("T", t) > tEnv.executeSql( > s""" > |CREATE TABLE rowtime_sink ( > | cnt BIGINT > |) WITH ( > | 'connector' = 'values', > | 'sink-insert-only' = 'false', > | 'changelog-mode' = 'I,UA,D' > |) > |""".stripMargin) > val sql = > """ > |INSERT INTO rowtime_sink > |SELECT COUNT(b) FROM ( > | SELECT a, b, c, rowtime > | FROM ( > | SELECT *, > | ROW_NUMBER() OVER (PARTITION BY b ORDER BY rowtime DESC) as rowNum > | FROM T > | ) > | WHERE rowNum = 1 > | ) > """.stripMargin > tEnv.executeSql(sql).await() > val rawResult = TestValuesTableFactory.getRawResults("rowtime_sink") > }{code} > E.g for the above sql, when enable MiniBatch optimization, the optimized plan > is as following. > {code:java} > Sink(table=[default_catalog.default_database.rowtime_sink], fields=[EXPR$0]) > +- GlobalGroupAggregate(select=[COUNT_RETRACT(count$0) AS EXPR$0]) > +- Exchange(distribution=[single]) > +- LocalGroupAggregate(select=[COUNT_RETRACT(b) AS count$0, > COUNT_RETRACT(*) AS count1$1]) > +- Calc(select=[b]) > +- Deduplicate(keep=[LastRow], key=[b], order=[ROWTIME]) > +- Exchange(distribution=[hash[b]]) > +- Calc(select=[b, rowtime]) > +- MiniBatchAssigner(interval=[1000ms], mode=[ProcTime]) > +- DataStreamScan(table=[[default_catalog, > default_database, T]], fields=[a, b, c, rowtime]){code} > A `StreamExecMiniBatchAssigner` will be inserted. The behavior is weird > because `Deduplicate` depends on rowTime, however > `ProcTimeMiniBatchAssignerOperator` will send watermark every specified > interval second depends on process time. For `Deduplicate`, the incoming > watermark does not relate to rowTime of incoming record, it cannot indicate > rowTime of all following input records are all larger than or equals to the > current incoming watermark. > -- This message was sent by Atlassian Jira (v8.3.4#803005)