You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by go...@apache.org on 2020/11/10 07:25:14 UTC
[flink] branch master updated: [FLINK-18070][table-planner-blink]
Don't materialize time attribute for middle nodes in
StreamCommonSubGraphBasedOptimizer
This is an automated email from the ASF dual-hosted git repository.
godfrey pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new e9c83a1 [FLINK-18070][table-planner-blink] Don't materialize time attribute for middle nodes in StreamCommonSubGraphBasedOptimizer
e9c83a1 is described below
commit e9c83a1646d2be32f64d2f39ef975407d41eb262
Author: liuyufei9527 <51...@users.noreply.github.com>
AuthorDate: Tue Nov 10 15:22:23 2020 +0800
[FLINK-18070][table-planner-blink] Don't materialize time attribute for middle nodes in StreamCommonSubGraphBasedOptimizer
---
.../StreamCommonSubGraphBasedOptimizer.scala | 2 +-
.../stream/sql/RelTimeIndicatorConverterTest.xml | 31 ++++++++++++++++++++
.../stream/sql/RelTimeIndicatorConverterTest.scala | 33 ++++++++++++++++++++++
3 files changed, 65 insertions(+), 1 deletion(-)
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/optimize/StreamCommonSubGraphBasedOptimizer.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/optimize/StreamCommonSubGraphBasedOptimizer.scala
index f67cc48..279cd34 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/optimize/StreamCommonSubGraphBasedOptimizer.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/optimize/StreamCommonSubGraphBasedOptimizer.scala
@@ -177,7 +177,7 @@ class StreamCommonSubGraphBasedOptimizer(planner: StreamPlanner)
def getMiniBatchInterval: MiniBatchInterval = miniBatchInterval
- override def needFinalTimeIndicatorConversion: Boolean = true
+ override def needFinalTimeIndicatorConversion: Boolean = isSinkBlock
})
}
diff --git a/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/RelTimeIndicatorConverterTest.xml b/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/RelTimeIndicatorConverterTest.xml
index 39d6c48..0a921be 100644
--- a/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/RelTimeIndicatorConverterTest.xml
+++ b/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/RelTimeIndicatorConverterTest.xml
@@ -347,4 +347,35 @@ Calc(select=[w$end AS rowtime, long, EXPR$2])
]]>
</Resource>
</TestCase>
+ <TestCase name="testKeepProcessTimeAttrAfterSubGraphOptimize">
+ <Resource name="planBefore">
+ <![CDATA[
+LogicalLegacySink(name=[`default_catalog`.`default_database`.`appendSink1`], fields=[long, sum])
++- LogicalProject(long=[$1], sum=[CAST($2):BIGINT])
+ +- LogicalAggregate(group=[{0, 1}], EXPR$1=[SUM($2)])
+ +- LogicalProject($f0=[$TUMBLE($2, 10000:INTERVAL SECOND)], long=[$0], int=[$1])
+ +- LogicalTableScan(table=[[default_catalog, default_database, MyTable2]])
+
+LogicalLegacySink(name=[`default_catalog`.`default_database`.`appendSink2`], fields=[long, sum])
++- LogicalProject(long=[$1], sum=[CAST($2):BIGINT])
+ +- LogicalAggregate(group=[{0, 1}], EXPR$1=[SUM($2)])
+ +- LogicalProject($f0=[$TUMBLE($2, 10000:INTERVAL SECOND)], long=[$0], int=[$1])
+ +- LogicalTableScan(table=[[default_catalog, default_database, MyTable2]])
+]]>
+ </Resource>
+ <Resource name="planAfter">
+ <![CDATA[
+Calc(select=[long, CAST(EXPR$1) AS sum], reuse_id=[1])
++- GroupWindowAggregate(groupBy=[long], window=[TumblingGroupWindow('w$, proctime, 10000)], select=[long, SUM(int) AS EXPR$1])
+ +- Exchange(distribution=[hash[long]])
+ +- DataStreamScan(table=[[default_catalog, default_database, MyTable2]], fields=[long, int, proctime])
+
+LegacySink(name=[`default_catalog`.`default_database`.`appendSink1`], fields=[long, sum])
++- Reused(reference_id=[1])
+
+LegacySink(name=[`default_catalog`.`default_database`.`appendSink2`], fields=[long, sum])
++- Reused(reference_id=[1])
+]]>
+ </Resource>
+ </TestCase>
</Root>
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/RelTimeIndicatorConverterTest.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/RelTimeIndicatorConverterTest.scala
index c8f740f..c3683ac 100644
--- a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/RelTimeIndicatorConverterTest.scala
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/RelTimeIndicatorConverterTest.scala
@@ -20,10 +20,12 @@ package org.apache.flink.table.planner.plan.stream.sql
import org.apache.flink.api.scala._
import org.apache.flink.table.api._
+import org.apache.flink.table.api.internal.TableEnvironmentInternal
import org.apache.flink.table.data.TimestampData
import org.apache.flink.table.functions.TableFunction
import org.apache.flink.table.planner.plan.stream.sql.RelTimeIndicatorConverterTest.TableFunc
import org.apache.flink.table.planner.utils.TableTestBase
+import org.apache.flink.table.types.logical.BigIntType
import org.junit.Test
@@ -163,6 +165,37 @@ class RelTimeIndicatorConverterTest extends TableTestBase {
util.verifyPlan(result)
}
+ @Test
+ def testKeepProcessTimeAttrAfterSubGraphOptimize(): Unit = {
+ val stmtSet = util.tableEnv.createStatementSet()
+ val sql =
+ """
+ |SELECT
+ | long,
+ | SUM(`int`)
+ |FROM MyTable2
+ | GROUP BY TUMBLE(proctime, INTERVAL '10' SECOND), long
+ """.stripMargin
+
+ val table = util.tableEnv.sqlQuery(sql)
+
+ val appendSink1 = util.createAppendTableSink(
+ Array("long", "sum"),
+ Array(new BigIntType(), new BigIntType()))
+ util.tableEnv.asInstanceOf[TableEnvironmentInternal].registerTableSinkInternal(
+ "appendSink1", appendSink1)
+ stmtSet.addInsert("appendSink1", table)
+
+ val appendSink2 = util.createAppendTableSink(
+ Array("long", "sum"),
+ Array(new BigIntType(), new BigIntType()))
+ util.tableEnv.asInstanceOf[TableEnvironmentInternal].registerTableSinkInternal(
+ "appendSink2", appendSink2)
+ stmtSet.addInsert("appendSink2", table)
+
+ util.verifyPlan(stmtSet)
+ }
+
// TODO add temporal table join case
}