You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by he...@apache.org on 2019/12/14 03:27:29 UTC
[flink] branch release-1.9 updated: [FLINK-15001]
[table-planner-blink] The digest of sub-plan reuse should contain
retraction traits for stream physical nodes
This is an automated email from the ASF dual-hosted git repository.
hequn pushed a commit to branch release-1.9
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.9 by this push:
new 9ea0493 [FLINK-15001] [table-planner-blink] The digest of sub-plan reuse should contain retraction traits for stream physical nodes
9ea0493 is described below
commit 9ea04933ee8294da757b5ebc73611cfc3f2a4915
Author: godfreyhe <go...@163.com>
AuthorDate: Mon Dec 2 15:05:34 2019 +0800
[FLINK-15001] [table-planner-blink] The digest of sub-plan reuse should contain retraction traits for stream physical nodes
This closes #10561.
---
.../table/planner/plan/reuse/SubplanReuser.scala | 4 +-
.../table/planner/plan/utils/FlinkRelOptUtil.scala | 62 ++++++++++-
.../table/planner/plan/utils/RelDigestUtil.scala | 120 ---------------------
.../planner/plan/utils/RelTreeWriterImpl.scala | 41 ++++---
.../digest/testGetDigestWithDynamicFunction.out | 28 ++---
.../testGetDigestWithDynamicFunctionView.out | 34 +++---
.../plan/stream/sql/DagOptimizationTest.xml | 105 +++++++++---------
.../plan/stream/sql/DagOptimizationTest.scala | 2 +-
.../planner/plan/utils/FlinkRelOptUtilTest.scala | 58 +++++++++-
.../planner/plan/utils/RelDigestUtilTest.scala | 83 --------------
10 files changed, 236 insertions(+), 301 deletions(-)
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/reuse/SubplanReuser.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/reuse/SubplanReuser.scala
index df66554..cf7404a 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/reuse/SubplanReuser.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/reuse/SubplanReuser.scala
@@ -23,7 +23,7 @@ import org.apache.flink.table.api.{TableConfig, TableException}
import org.apache.flink.table.planner.plan.nodes.calcite.Sink
import org.apache.flink.table.planner.plan.nodes.logical.FlinkLogicalTableSourceScan
import org.apache.flink.table.planner.plan.nodes.physical.PhysicalTableSourceScan
-import org.apache.flink.table.planner.plan.utils.{DefaultRelShuttle, RelDigestUtil}
+import org.apache.flink.table.planner.plan.utils.{DefaultRelShuttle, FlinkRelOptUtil}
import com.google.common.collect.{Maps, Sets}
import org.apache.calcite.rel.core.{Exchange, TableFunctionScan}
@@ -106,7 +106,7 @@ object SubplanReuser {
if (digest != null) {
digest
} else {
- val newDigest = RelDigestUtil.getDigest(node)
+ val newDigest = FlinkRelOptUtil.getDigest(node)
mapRelToDigest.put(node, newDigest)
newDigest
}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/utils/FlinkRelOptUtil.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/utils/FlinkRelOptUtil.scala
index acf20ec..3faeb42 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/utils/FlinkRelOptUtil.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/utils/FlinkRelOptUtil.scala
@@ -83,12 +83,72 @@ object FlinkRelOptUtil {
detailLevel,
withIdPrefix,
withRetractTraits,
- withRowType)
+ withRowType,
+ withTreeStyle = true)
rel.explain(planWriter)
sw.toString
}
/**
+ * Gets the digest for a rel tree.
+ *
+ * The digest of RelNode should contain the result of RelNode#explain method, retraction traits
+ * (for StreamPhysicalRel) and RelNode's row type.
+ *
+ * Row type is part of the digest for the rare occasion that similar
+ * expressions have different types, e.g.
+ * "WITH
+ * t1 AS (SELECT CAST(a as BIGINT) AS a, SUM(b) AS b FROM x GROUP BY CAST(a as BIGINT)),
+ * t2 AS (SELECT CAST(a as DOUBLE) AS a, SUM(b) AS b FROM x GROUP BY CAST(a as DOUBLE))
+ * SELECT t1.*, t2.* FROM t1, t2 WHERE t1.b = t2.b"
+ *
+ * the physical plan is:
+ * {{{
+ * HashJoin(where=[=(b, b0)], join=[a, b, a0, b0], joinType=[InnerJoin],
+ * isBroadcast=[true], build=[right])
+ * :- HashAggregate(groupBy=[a], select=[a, Final_SUM(sum$0) AS b])
+ * : +- Exchange(distribution=[hash[a]])
+ * : +- LocalHashAggregate(groupBy=[a], select=[a, Partial_SUM(b) AS sum$0])
+ * : +- Calc(select=[CAST(a) AS a, b])
+ * : +- ScanTable(table=[[builtin, default, x]], fields=[a, b, c])
+ * +- Exchange(distribution=[broadcast])
+ * +- HashAggregate(groupBy=[a], select=[a, Final_SUM(sum$0) AS b])
+ * +- Exchange(distribution=[hash[a]])
+ * +- LocalHashAggregate(groupBy=[a], select=[a, Partial_SUM(b) AS sum$0])
+ * +- Calc(select=[CAST(a) AS a, b])
+ * +- ScanTable(table=[[builtin, default, x]], fields=[a, b, c])
+ * }}}
+ *
+ * The sub-plan of `HashAggregate(groupBy=[a], select=[a, Final_SUM(sum$0) AS b])`
+ * are different because `CAST(a) AS a` has different types, where one is BIGINT type
+ * and another is DOUBLE type.
+ *
+ * If use the result of `RelOptUtil.toString(aggregate, SqlExplainLevel.DIGEST_ATTRIBUTES)`
+ * on `HashAggregate(groupBy=[a], select=[a, Final_SUM(sum$0) AS b])` as digest,
+ * we will get incorrect result. So rewrite `explain_` method of `RelWriterImpl` to
+ * add row-type to digest value.
+ *
+ * @param rel rel node tree
+ * @return The digest of given rel tree.
+ */
+ def getDigest(rel: RelNode): String = {
+ val sw = new StringWriter
+ rel.explain(new RelTreeWriterImpl(
+ new PrintWriter(sw),
+ explainLevel = SqlExplainLevel.DIGEST_ATTRIBUTES,
+ // ignore id, only contains RelNode's attributes
+ withIdPrefix = false,
+ // add retraction traits to digest for StreamPhysicalRel node
+ withRetractTraits = true,
+ // add row type to digest to avoid corner case that similar
+ // expressions have different types
+ withRowType = true,
+ // ignore tree style, only contains RelNode's attributes
+ withTreeStyle = false))
+ sw.toString
+ }
+
+ /**
* Returns the null direction if not specified.
*
* @param direction Direction that a field is ordered in.
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/utils/RelDigestUtil.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/utils/RelDigestUtil.scala
deleted file mode 100644
index e9334e4..0000000
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/utils/RelDigestUtil.scala
+++ /dev/null
@@ -1,120 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.table.planner.plan.utils
-
-import org.apache.calcite.rel.RelNode
-import org.apache.calcite.rel.externalize.RelWriterImpl
-import org.apache.calcite.sql.SqlExplainLevel
-import org.apache.calcite.util.Pair
-
-import java.io.{PrintWriter, StringWriter}
-import java.util
-
-import scala.collection.JavaConversions._
-
-/**
- * Row type is part of the digest for the rare occasion that similar
- * expressions have different types, e.g.
- * "WITH
- * t1 AS (SELECT CAST(a as BIGINT) AS a, SUM(b) AS b FROM x GROUP BY CAST(a as BIGINT)),
- * t2 AS (SELECT CAST(a as DOUBLE) AS a, SUM(b) AS b FROM x GROUP BY CAST(a as DOUBLE))
- * SELECT t1.*, t2.* FROM t1, t2 WHERE t1.b = t2.b"
- *
- * the physical plan is:
- * {{{
- * HashJoin(where=[=(b, b0)], join=[a, b, a0, b0], joinType=[InnerJoin],
- * isBroadcast=[true], build=[right])
- * :- HashAggregate(groupBy=[a], select=[a, Final_SUM(sum$0) AS b])
- * : +- Exchange(distribution=[hash[a]])
- * : +- LocalHashAggregate(groupBy=[a], select=[a, Partial_SUM(b) AS sum$0])
- * : +- Calc(select=[CAST(a) AS a, b])
- * : +- ScanTable(table=[[builtin, default, x]], fields=[a, b, c])
- * +- Exchange(distribution=[broadcast])
- * +- HashAggregate(groupBy=[a], select=[a, Final_SUM(sum$0) AS b])
- * +- Exchange(distribution=[hash[a]])
- * +- LocalHashAggregate(groupBy=[a], select=[a, Partial_SUM(b) AS sum$0])
- * +- Calc(select=[CAST(a) AS a, b])
- * +- ScanTable(table=[[builtin, default, x]], fields=[a, b, c])
- * }}}
- *
- * The sub-plan of `HashAggregate(groupBy=[a], select=[a, Final_SUM(sum$0) AS b])`
- * are different because `CAST(a) AS a` has different types, where one is BIGINT type
- * and another is DOUBLE type.
- *
- * If use the result of `RelOptUtil.toString(aggregate, SqlExplainLevel.DIGEST_ATTRIBUTES)`
- * on `HashAggregate(groupBy=[a], select=[a, Final_SUM(sum$0) AS b])` as digest,
- * we will get incorrect result. So rewrite `explain_` method of `RelWriterImpl` to
- * add row-type to digest value.
- */
-class RelDigestWriterImpl(sw: StringWriter)
- extends RelWriterImpl(new PrintWriter(sw), SqlExplainLevel.DIGEST_ATTRIBUTES, false) {
-
- override def explain_(rel: RelNode, values: util.List[Pair[String, AnyRef]]): Unit = {
- val inputs = rel.getInputs
- val mq = rel.getCluster.getMetadataQuery
- if (!mq.isVisibleInExplain(rel, getDetailLevel)) {
- // render children in place of this, at same level
- inputs.foreach(_.explain(this))
- return
- }
- val s = explainRel(rel, values)
- pw.println(s)
- inputs.foreach(_.explain(this))
- }
-
- /**
- * Returns explain result for given rel.
- */
- protected def explainRel(rel: RelNode, values: util.List[Pair[String, AnyRef]]): String = {
- val s = new StringBuilder
- s.append(rel.getRelTypeName)
- var j = 0
- s.append("(")
- values.foreach {
- case value if value.right.isInstanceOf[RelNode] => // do nothing
- case value =>
- if (j != 0) s.append(", ")
- j += 1
- s.append(value.left).append("=[").append(value.right).append("]")
- }
- if (j > 0) {
- s.append(",")
- }
- s.append("rowType=[").append(rel.getRowType.toString).append("]")
- s.append(")")
- s.toString()
- }
-
-}
-
-object RelDigestUtil {
-
- /**
- * Gets the digest for a rel tree.
- *
- * @param rel rel node tree
- * @return The digest of given rel tree.
- */
- def getDigest(rel: RelNode): String = {
- val sw = new StringWriter
- rel.explain(new RelDigestWriterImpl(sw))
- sw.toString
- }
-
-}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/utils/RelTreeWriterImpl.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/utils/RelTreeWriterImpl.scala
index 0966d7a..36e0277 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/utils/RelTreeWriterImpl.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/utils/RelTreeWriterImpl.scala
@@ -38,7 +38,8 @@ class RelTreeWriterImpl(
explainLevel: SqlExplainLevel = SqlExplainLevel.EXPPLAN_ATTRIBUTES,
withIdPrefix: Boolean = false,
withRetractTraits: Boolean = false,
- withRowType: Boolean = false)
+ withRowType: Boolean = false,
+ withTreeStyle: Boolean = true)
extends RelWriterImpl(pw, explainLevel, withIdPrefix) {
var lastChildren: Seq[Boolean] = Nil
@@ -55,11 +56,13 @@ class RelTreeWriterImpl(
}
val s = new StringBuilder
- if (depth > 0) {
- lastChildren.init.foreach { isLast =>
- s.append(if (isLast) " " else ": ")
+ if (withTreeStyle) {
+ if (depth > 0) {
+ lastChildren.init.foreach { isLast =>
+ s.append(if (isLast) " " else ": ")
+ }
+ s.append(if (lastChildren.last) "+- " else ":- ")
}
- s.append(if (lastChildren.last) "+- " else ":- ")
}
if (withIdPrefix) {
@@ -112,18 +115,30 @@ class RelTreeWriterImpl(
}
pw.println(s)
if (inputs.length > 1) inputs.toSeq.init.foreach { rel =>
- depth = depth + 1
- lastChildren = lastChildren :+ false
+ if (withTreeStyle) {
+ depth = depth + 1
+ lastChildren = lastChildren :+ false
+ }
+
rel.explain(this)
- depth = depth - 1
- lastChildren = lastChildren.init
+
+ if (withTreeStyle) {
+ depth = depth - 1
+ lastChildren = lastChildren.init
+ }
}
if (!inputs.isEmpty) {
- depth = depth + 1
- lastChildren = lastChildren :+ true
+ if (withTreeStyle) {
+ depth = depth + 1
+ lastChildren = lastChildren :+ true
+ }
+
inputs.toSeq.last.explain(this)
- depth = depth - 1
- lastChildren = lastChildren.init
+
+ if (withTreeStyle) {
+ depth = depth - 1
+ lastChildren = lastChildren.init
+ }
}
}
}
diff --git a/flink-table/flink-table-planner-blink/src/test/resources/digest/testGetDigestWithDynamicFunction.out b/flink-table/flink-table-planner-blink/src/test/resources/digest/testGetDigestWithDynamicFunction.out
index ac3b44b..9269167 100644
--- a/flink-table/flink-table-planner-blink/src/test/resources/digest/testGetDigestWithDynamicFunction.out
+++ b/flink-table/flink-table-planner-blink/src/test/resources/digest/testGetDigestWithDynamicFunction.out
@@ -1,14 +1,14 @@
-LogicalIntersect(all=[false],rowType=[RecordType(INTEGER random)])
-LogicalIntersect(all=[false],rowType=[RecordType(INTEGER random)])
-LogicalProject(random=[$0],rowType=[RecordType(INTEGER random)])
-LogicalSort(sort0=[$1], dir0=[ASC-nulls-first], fetch=[1],rowType=[RecordType(INTEGER random, DOUBLE EXPR$1)])
-LogicalProject(random=[$1], EXPR$1=[RAND()],rowType=[RecordType(INTEGER random, DOUBLE EXPR$1)])
-LogicalTableScan(table=[[default_catalog, default_database, MyTable]],rowType=[RecordType(VARCHAR(2147483647) first, INTEGER id, DOUBLE score, VARCHAR(2147483647) last)])
-LogicalProject(random=[$0],rowType=[RecordType(INTEGER random)])
-LogicalSort(sort0=[$1], dir0=[ASC-nulls-first], fetch=[1],rowType=[RecordType(INTEGER random, DOUBLE EXPR$1)])
-LogicalProject(random=[$1], EXPR$1=[RAND()],rowType=[RecordType(INTEGER random, DOUBLE EXPR$1)])
-LogicalTableScan(table=[[default_catalog, default_database, MyTable]],rowType=[RecordType(VARCHAR(2147483647) first, INTEGER id, DOUBLE score, VARCHAR(2147483647) last)])
-LogicalProject(random=[$0],rowType=[RecordType(INTEGER random)])
-LogicalSort(sort0=[$1], dir0=[ASC-nulls-first], fetch=[1],rowType=[RecordType(INTEGER random, DOUBLE EXPR$1)])
-LogicalProject(random=[$1], EXPR$1=[RAND()],rowType=[RecordType(INTEGER random, DOUBLE EXPR$1)])
-LogicalTableScan(table=[[default_catalog, default_database, MyTable]],rowType=[RecordType(VARCHAR(2147483647) first, INTEGER id, DOUBLE score, VARCHAR(2147483647) last)])
+LogicalIntersect(all=[false]), rowType=[RecordType(INTEGER random)]
+LogicalIntersect(all=[false]), rowType=[RecordType(INTEGER random)]
+LogicalProject(random=[$0]), rowType=[RecordType(INTEGER random)]
+LogicalSort(sort0=[$1], dir0=[ASC-nulls-first], fetch=[1]), rowType=[RecordType(INTEGER random, DOUBLE EXPR$1)]
+LogicalProject(random=[$1], EXPR$1=[RAND()]), rowType=[RecordType(INTEGER random, DOUBLE EXPR$1)]
+LogicalTableScan(table=[[default_catalog, default_database, MyTable]]), rowType=[RecordType(VARCHAR(2147483647) first, INTEGER id, DOUBLE score, VARCHAR(2147483647) last)]
+LogicalProject(random=[$0]), rowType=[RecordType(INTEGER random)]
+LogicalSort(sort0=[$1], dir0=[ASC-nulls-first], fetch=[1]), rowType=[RecordType(INTEGER random, DOUBLE EXPR$1)]
+LogicalProject(random=[$1], EXPR$1=[RAND()]), rowType=[RecordType(INTEGER random, DOUBLE EXPR$1)]
+LogicalTableScan(table=[[default_catalog, default_database, MyTable]]), rowType=[RecordType(VARCHAR(2147483647) first, INTEGER id, DOUBLE score, VARCHAR(2147483647) last)]
+LogicalProject(random=[$0]), rowType=[RecordType(INTEGER random)]
+LogicalSort(sort0=[$1], dir0=[ASC-nulls-first], fetch=[1]), rowType=[RecordType(INTEGER random, DOUBLE EXPR$1)]
+LogicalProject(random=[$1], EXPR$1=[RAND()]), rowType=[RecordType(INTEGER random, DOUBLE EXPR$1)]
+LogicalTableScan(table=[[default_catalog, default_database, MyTable]]), rowType=[RecordType(VARCHAR(2147483647) first, INTEGER id, DOUBLE score, VARCHAR(2147483647) last)]
diff --git a/flink-table/flink-table-planner-blink/src/test/resources/digest/testGetDigestWithDynamicFunctionView.out b/flink-table/flink-table-planner-blink/src/test/resources/digest/testGetDigestWithDynamicFunctionView.out
index c965d30..f218193 100644
--- a/flink-table/flink-table-planner-blink/src/test/resources/digest/testGetDigestWithDynamicFunctionView.out
+++ b/flink-table/flink-table-planner-blink/src/test/resources/digest/testGetDigestWithDynamicFunctionView.out
@@ -1,17 +1,17 @@
-LogicalIntersect(all=[false],rowType=[RecordType(INTEGER random)])
-LogicalIntersect(all=[false],rowType=[RecordType(INTEGER random)])
-LogicalProject(random=[$0],rowType=[RecordType(INTEGER random)])
-LogicalProject(random=[$0],rowType=[RecordType(INTEGER random)])
-LogicalSort(sort0=[$1], dir0=[ASC-nulls-first], fetch=[1],rowType=[RecordType(INTEGER random, DOUBLE EXPR$1)])
-LogicalProject(random=[$1], EXPR$1=[RAND()],rowType=[RecordType(INTEGER random, DOUBLE EXPR$1)])
-LogicalTableScan(table=[[default_catalog, default_database, MyTable]],rowType=[RecordType(VARCHAR(2147483647) first, INTEGER id, DOUBLE score, VARCHAR(2147483647) last)])
-LogicalProject(random=[$0],rowType=[RecordType(INTEGER random)])
-LogicalProject(random=[$0],rowType=[RecordType(INTEGER random)])
-LogicalSort(sort0=[$1], dir0=[ASC-nulls-first], fetch=[1],rowType=[RecordType(INTEGER random, DOUBLE EXPR$1)])
-LogicalProject(random=[$1], EXPR$1=[RAND()],rowType=[RecordType(INTEGER random, DOUBLE EXPR$1)])
-LogicalTableScan(table=[[default_catalog, default_database, MyTable]],rowType=[RecordType(VARCHAR(2147483647) first, INTEGER id, DOUBLE score, VARCHAR(2147483647) last)])
-LogicalProject(random=[$0],rowType=[RecordType(INTEGER random)])
-LogicalProject(random=[$0],rowType=[RecordType(INTEGER random)])
-LogicalSort(sort0=[$1], dir0=[ASC-nulls-first], fetch=[1],rowType=[RecordType(INTEGER random, DOUBLE EXPR$1)])
-LogicalProject(random=[$1], EXPR$1=[RAND()],rowType=[RecordType(INTEGER random, DOUBLE EXPR$1)])
-LogicalTableScan(table=[[default_catalog, default_database, MyTable]],rowType=[RecordType(VARCHAR(2147483647) first, INTEGER id, DOUBLE score, VARCHAR(2147483647) last)])
+LogicalIntersect(all=[false]), rowType=[RecordType(INTEGER random)]
+LogicalIntersect(all=[false]), rowType=[RecordType(INTEGER random)]
+LogicalProject(random=[$0]), rowType=[RecordType(INTEGER random)]
+LogicalProject(random=[$0]), rowType=[RecordType(INTEGER random)]
+LogicalSort(sort0=[$1], dir0=[ASC-nulls-first], fetch=[1]), rowType=[RecordType(INTEGER random, DOUBLE EXPR$1)]
+LogicalProject(random=[$1], EXPR$1=[RAND()]), rowType=[RecordType(INTEGER random, DOUBLE EXPR$1)]
+LogicalTableScan(table=[[default_catalog, default_database, MyTable]]), rowType=[RecordType(VARCHAR(2147483647) first, INTEGER id, DOUBLE score, VARCHAR(2147483647) last)]
+LogicalProject(random=[$0]), rowType=[RecordType(INTEGER random)]
+LogicalProject(random=[$0]), rowType=[RecordType(INTEGER random)]
+LogicalSort(sort0=[$1], dir0=[ASC-nulls-first], fetch=[1]), rowType=[RecordType(INTEGER random, DOUBLE EXPR$1)]
+LogicalProject(random=[$1], EXPR$1=[RAND()]), rowType=[RecordType(INTEGER random, DOUBLE EXPR$1)]
+LogicalTableScan(table=[[default_catalog, default_database, MyTable]]), rowType=[RecordType(VARCHAR(2147483647) first, INTEGER id, DOUBLE score, VARCHAR(2147483647) last)]
+LogicalProject(random=[$0]), rowType=[RecordType(INTEGER random)]
+LogicalProject(random=[$0]), rowType=[RecordType(INTEGER random)]
+LogicalSort(sort0=[$1], dir0=[ASC-nulls-first], fetch=[1]), rowType=[RecordType(INTEGER random, DOUBLE EXPR$1)]
+LogicalProject(random=[$1], EXPR$1=[RAND()]), rowType=[RecordType(INTEGER random, DOUBLE EXPR$1)]
+LogicalTableScan(table=[[default_catalog, default_database, MyTable]]), rowType=[RecordType(VARCHAR(2147483647) first, INTEGER id, DOUBLE score, VARCHAR(2147483647) last)]
diff --git a/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/DagOptimizationTest.xml b/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/DagOptimizationTest.xml
index d786faf..1297d6b 100644
--- a/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/DagOptimizationTest.xml
+++ b/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/DagOptimizationTest.xml
@@ -158,23 +158,22 @@ LogicalSink(name=[appendSink2], fields=[a, b1])
</Resource>
<Resource name="planAfter">
<![CDATA[
-Calc(select=[a AS a1, b AS b1], where=[<=(a, 10)], updateAsRetraction=[false], accMode=[Acc], reuse_id=[1])
-+- TableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c)]]], fields=[a, b, c], updateAsRetraction=[false], accMode=[Acc])
-
-Union(all=[true], union=[a1, b1], updateAsRetraction=[false], accMode=[Acc], reuse_id=[2])
-:- Reused(reference_id=[1])
+Union(all=[true], union=[a1, b1], updateAsRetraction=[false], accMode=[Acc], reuse_id=[1])
+:- Calc(select=[a AS a1, b AS b1], where=[<=(a, 10)], updateAsRetraction=[false], accMode=[Acc])
+: +- TableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c)]]], fields=[a, b, c], updateAsRetraction=[false], accMode=[Acc])
+- Calc(select=[a, b1], updateAsRetraction=[false], accMode=[Acc])
+- Join(joinType=[InnerJoin], where=[=(a, a1)], select=[a1, b1, a, b, c, d, e], leftInputSpec=[NoUniqueKey], rightInputSpec=[NoUniqueKey], updateAsRetraction=[false], accMode=[Acc])
:- Exchange(distribution=[hash[a1]], updateAsRetraction=[true], accMode=[Acc])
- : +- Reused(reference_id=[1])
+ : +- Calc(select=[a AS a1, b AS b1], where=[<=(a, 10)], updateAsRetraction=[true], accMode=[Acc])
+ : +- TableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c)]]], fields=[a, b, c], updateAsRetraction=[true], accMode=[Acc])
+- Exchange(distribution=[hash[a]], updateAsRetraction=[true], accMode=[Acc])
+- TableSourceScan(table=[[default_catalog, default_database, MyTable2, source: [TestTableSource(a, b, c, d, e)]]], fields=[a, b, c, d, e], updateAsRetraction=[true], accMode=[Acc])
Sink(name=[appendSink1], fields=[a, b1], updateAsRetraction=[false], accMode=[Acc])
-+- Reused(reference_id=[2])
++- Reused(reference_id=[1])
Sink(name=[appendSink2], fields=[a, b1], updateAsRetraction=[false], accMode=[Acc])
-+- Reused(reference_id=[2])
++- Reused(reference_id=[1])
]]>
</Resource>
</TestCase>
@@ -370,32 +369,30 @@ LogicalSink(name=[appendSink2], fields=[a, total_c])
</Resource>
<Resource name="planAfter">
<![CDATA[
-TableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c)]]], fields=[a, b, c], updateAsRetraction=[true], accMode=[Acc], reuse_id=[1])
-
-Calc(select=[a, total_c], updateAsRetraction=[false], accMode=[Acc], reuse_id=[2])
+Calc(select=[a, total_c], updateAsRetraction=[false], accMode=[Acc], reuse_id=[1])
+- GroupAggregate(groupBy=[a, b], select=[a, b, COUNT(DISTINCT c) AS total_c], updateAsRetraction=[false], accMode=[Acc])
+- Exchange(distribution=[hash[a, b]], updateAsRetraction=[true], accMode=[Acc])
+- Calc(select=[a, b, f0 AS c], updateAsRetraction=[true], accMode=[Acc])
+- Correlate(invocation=[split($cor0.c)], correlate=[table(split($cor0.c))], select=[a,b,c,f0], rowType=[RecordType(INTEGER a, BIGINT b, VARCHAR(2147483647) c, VARCHAR(2147483647) f0)], joinType=[INNER], updateAsRetraction=[true], accMode=[Acc])
+- Calc(select=[a, -(b, MOD(b, 300)) AS b, c], where=[AND(>=(b, UNIX_TIMESTAMP(_UTF-16LE'${startTime}')), <>(c, _UTF-16LE'':VARCHAR(2147483647) CHARACTER SET "UTF-16LE"))], updateAsRetraction=[true], accMode=[Acc])
- +- Reused(reference_id=[1])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c)]]], fields=[a, b, c], updateAsRetraction=[true], accMode=[Acc])
-Calc(select=[a, 0 AS total_c], where=[>=(b, UNIX_TIMESTAMP(_UTF-16LE'${startTime}'))], updateAsRetraction=[false], accMode=[Acc], reuse_id=[3])
-+- Reused(reference_id=[1])
+Calc(select=[a, 0 AS total_c], where=[>=(b, UNIX_TIMESTAMP(_UTF-16LE'${startTime}'))], updateAsRetraction=[false], accMode=[Acc], reuse_id=[2])
++- TableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c)]]], fields=[a, b, c], updateAsRetraction=[false], accMode=[Acc])
Sink(name=[appendSink1], fields=[a, total_c], updateAsRetraction=[false], accMode=[Acc])
+- Union(all=[true], union=[a, total_c], updateAsRetraction=[false], accMode=[Acc])
:- Calc(select=[a, total_c], where=[>(a, 50)], updateAsRetraction=[false], accMode=[Acc])
- : +- Reused(reference_id=[2])
+ : +- Reused(reference_id=[1])
+- Calc(select=[a, CAST(total_c) AS total_c], where=[>(a, 50)], updateAsRetraction=[false], accMode=[Acc])
- +- Reused(reference_id=[3])
+ +- Reused(reference_id=[2])
Sink(name=[appendSink2], fields=[a, total_c], updateAsRetraction=[false], accMode=[Acc])
+- Union(all=[true], union=[a, total_c], updateAsRetraction=[false], accMode=[Acc])
:- Calc(select=[a, total_c], where=[<(a, 50)], updateAsRetraction=[false], accMode=[Acc])
- : +- Reused(reference_id=[2])
+ : +- Reused(reference_id=[1])
+- Calc(select=[a, CAST(total_c) AS total_c], where=[<(a, 50)], updateAsRetraction=[false], accMode=[Acc])
- +- Reused(reference_id=[3])
+ +- Reused(reference_id=[2])
]]>
</Resource>
</TestCase>
@@ -479,29 +476,34 @@ LogicalSink(name=[appendSink3], fields=[a])
</Resource>
<Resource name="planAfter">
<![CDATA[
-Calc(select=[a], reuse_id=[2])
-+- Union(all=[true], union=[a, c])
- :- Calc(select=[a, c])
- : +- TableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
- +- Calc(select=[d, f])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable1, source: [TestTableSource(d, e, f)]]], fields=[d, e, f])
-
-Exchange(distribution=[single], reuse_id=[1])
-+- Union(all=[true], union=[a])
- :- Reused(reference_id=[2])
- +- Calc(select=[a])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable2, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
-
-Sink(name=[appendSink1], fields=[total_sum])
-+- GroupAggregate(select=[SUM(a) AS total_sum])
+Calc(select=[a, c], updateAsRetraction=[true], accMode=[Acc], reuse_id=[2])
++- TableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c)]]], fields=[a, b, c], updateAsRetraction=[true], accMode=[Acc])
+
+Calc(select=[d, f], updateAsRetraction=[true], accMode=[Acc], reuse_id=[3])
++- TableSourceScan(table=[[default_catalog, default_database, MyTable1, source: [TestTableSource(d, e, f)]]], fields=[d, e, f], updateAsRetraction=[true], accMode=[Acc])
+
+Exchange(distribution=[single], updateAsRetraction=[true], accMode=[Acc], reuse_id=[1])
++- Union(all=[true], union=[a], updateAsRetraction=[true], accMode=[Acc])
+ :- Calc(select=[a], updateAsRetraction=[true], accMode=[Acc])
+ : +- Union(all=[true], union=[a, c], updateAsRetraction=[true], accMode=[Acc])
+ : :- Reused(reference_id=[2])
+ : +- Reused(reference_id=[3])
+ +- Calc(select=[a], updateAsRetraction=[true], accMode=[Acc])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable2, source: [TestTableSource(a, b, c)]]], fields=[a, b, c], updateAsRetraction=[true], accMode=[Acc])
+
+Sink(name=[appendSink1], fields=[total_sum], updateAsRetraction=[false], accMode=[Acc])
++- GroupAggregate(select=[SUM(a) AS total_sum], updateAsRetraction=[false], accMode=[Acc])
+- Reused(reference_id=[1])
-Sink(name=[appendSink2], fields=[total_min])
-+- GroupAggregate(select=[MIN(a) AS total_min])
+Sink(name=[appendSink2], fields=[total_min], updateAsRetraction=[false], accMode=[Acc])
++- GroupAggregate(select=[MIN(a) AS total_min], updateAsRetraction=[false], accMode=[Acc])
+- Reused(reference_id=[1])
-Sink(name=[appendSink3], fields=[a])
-+- Reused(reference_id=[2])
+Sink(name=[appendSink3], fields=[a], updateAsRetraction=[false], accMode=[Acc])
++- Calc(select=[a], updateAsRetraction=[false], accMode=[Acc])
+ +- Union(all=[true], union=[a, c], updateAsRetraction=[false], accMode=[Acc])
+ :- Reused(reference_id=[2])
+ +- Reused(reference_id=[3])
]]>
</Resource>
</TestCase>
@@ -544,29 +546,33 @@ LogicalSink(name=[upsertSink], fields=[total_min])
</Resource>
<Resource name="planAfter">
<![CDATA[
-Union(all=[true], union=[a, c], updateAsRetraction=[false], accMode=[Acc], reuse_id=[1])
-:- Calc(select=[a, c], updateAsRetraction=[true], accMode=[Acc])
-: +- TableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c)]]], fields=[a, b, c], updateAsRetraction=[true], accMode=[Acc])
-+- Calc(select=[d, f], updateAsRetraction=[true], accMode=[Acc])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable1, source: [TestTableSource(d, e, f)]]], fields=[d, e, f], updateAsRetraction=[true], accMode=[Acc])
+Calc(select=[a, c], updateAsRetraction=[true], accMode=[Acc], reuse_id=[1])
++- TableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c)]]], fields=[a, b, c], updateAsRetraction=[true], accMode=[Acc])
+
+Calc(select=[d, f], updateAsRetraction=[true], accMode=[Acc], reuse_id=[2])
++- TableSourceScan(table=[[default_catalog, default_database, MyTable1, source: [TestTableSource(d, e, f)]]], fields=[d, e, f], updateAsRetraction=[true], accMode=[Acc])
Sink(name=[appendSink], fields=[a, c], updateAsRetraction=[false], accMode=[Acc])
-+- Reused(reference_id=[1])
++- Union(all=[true], union=[a, c], updateAsRetraction=[false], accMode=[Acc])
+ :- Reused(reference_id=[1])
+ +- Reused(reference_id=[2])
-Exchange(distribution=[single], updateAsRetraction=[true], accMode=[Acc], reuse_id=[2])
+Exchange(distribution=[single], updateAsRetraction=[true], accMode=[Acc], reuse_id=[3])
+- Calc(select=[a], updateAsRetraction=[true], accMode=[Acc])
+- Union(all=[true], union=[a, c], updateAsRetraction=[true], accMode=[Acc])
- :- Reused(reference_id=[1])
+ :- Union(all=[true], union=[a, c], updateAsRetraction=[true], accMode=[Acc])
+ : :- Reused(reference_id=[1])
+ : +- Reused(reference_id=[2])
+- Calc(select=[a, c], updateAsRetraction=[true], accMode=[Acc])
+- TableSourceScan(table=[[default_catalog, default_database, MyTable2, source: [TestTableSource(a, b, c)]]], fields=[a, b, c], updateAsRetraction=[true], accMode=[Acc])
Sink(name=[retractSink], fields=[total_sum], updateAsRetraction=[true], accMode=[AccRetract])
+- GroupAggregate(select=[SUM(a) AS total_sum], updateAsRetraction=[true], accMode=[AccRetract])
- +- Reused(reference_id=[2])
+ +- Reused(reference_id=[3])
Sink(name=[upsertSink], fields=[total_min], updateAsRetraction=[false], accMode=[Acc])
+- GroupAggregate(select=[MIN(a) AS total_min], updateAsRetraction=[false], accMode=[Acc])
- +- Reused(reference_id=[2])
+ +- Reused(reference_id=[3])
]]>
</Resource>
</TestCase>
@@ -796,12 +802,13 @@ LogicalSink(name=[appendSink], fields=[a1, b1])
<![CDATA[
Sink(name=[appendSink], fields=[a1, b1], updateAsRetraction=[false], accMode=[Acc])
+- Union(all=[true], union=[a1, b1], updateAsRetraction=[false], accMode=[Acc])
- :- Calc(select=[a AS a1, b AS b1], where=[<=(a, 10)], updateAsRetraction=[false], accMode=[Acc], reuse_id=[1])
+ :- Calc(select=[a AS a1, b AS b1], where=[<=(a, 10)], updateAsRetraction=[false], accMode=[Acc])
: +- TableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c)]]], fields=[a, b, c], updateAsRetraction=[false], accMode=[Acc])
+- Calc(select=[a, b1], updateAsRetraction=[false], accMode=[Acc])
+- Join(joinType=[InnerJoin], where=[=(a, a1)], select=[a1, b1, a, b, c, d, e], leftInputSpec=[NoUniqueKey], rightInputSpec=[NoUniqueKey], updateAsRetraction=[false], accMode=[Acc])
:- Exchange(distribution=[hash[a1]], updateAsRetraction=[true], accMode=[Acc])
- : +- Reused(reference_id=[1])
+ : +- Calc(select=[a AS a1, b AS b1], where=[<=(a, 10)], updateAsRetraction=[true], accMode=[Acc])
+ : +- TableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c)]]], fields=[a, b, c], updateAsRetraction=[true], accMode=[Acc])
+- Exchange(distribution=[hash[a]], updateAsRetraction=[true], accMode=[Acc])
+- DataStreamScan(table=[[default_catalog, default_database, MyTable2]], fields=[a, b, c, d, e], updateAsRetraction=[true], accMode=[Acc])
]]>
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/DagOptimizationTest.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/DagOptimizationTest.scala
index bfce5b2..bbcc3a1 100644
--- a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/DagOptimizationTest.scala
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/DagOptimizationTest.scala
@@ -326,7 +326,7 @@ class DagOptimizationTest extends TableTestBase {
val appendSink3 = util.createAppendTableSink(Array("a"), Array(INT))
util.writeToSink(table3, appendSink3, "appendSink3")
- util.verifyPlan()
+ util.verifyPlanWithTrait()
}
@Test
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/utils/FlinkRelOptUtilTest.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/utils/FlinkRelOptUtilTest.scala
index 3b885ec..fd72c78 100644
--- a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/utils/FlinkRelOptUtilTest.scala
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/utils/FlinkRelOptUtilTest.scala
@@ -17,18 +17,41 @@
*/
package org.apache.flink.table.planner.plan.utils
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo.{DOUBLE_TYPE_INFO, INT_TYPE_INFO, STRING_TYPE_INFO}
+import org.apache.flink.api.java.typeutils.RowTypeInfo
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
+import org.apache.flink.table.api.internal.TableEnvironmentImpl
import org.apache.flink.table.api.scala.{StreamTableEnvironment, _}
+import org.apache.flink.table.api.{EnvironmentSettings, TableEnvironment}
import org.apache.flink.table.planner.plan.`trait`.{MiniBatchInterval, MiniBatchMode}
+import org.apache.flink.table.planner.runtime.utils.BatchTableEnvUtil
+import org.apache.flink.table.planner.runtime.utils.BatchTestBase.row
import org.apache.flink.table.planner.utils.TableTestUtil
import org.apache.calcite.sql.SqlExplainLevel
import org.junit.Assert.assertEquals
-import org.junit.Test
+import org.junit.{Before, Test}
+
+import scala.collection.Seq
class FlinkRelOptUtilTest {
+ var tableEnv: TableEnvironment = _
+
+ @Before
+ def before(): Unit = {
+ val settings = EnvironmentSettings.newInstance().useBlinkPlanner().build()
+ val tEnv = TableEnvironmentImpl.create(settings)
+ BatchTableEnvUtil.registerCollection(
+ tEnv,
+ "MyTable",
+ Seq(row("Mike", 1, 12.3, "Smith")),
+ new RowTypeInfo(STRING_TYPE_INFO, INT_TYPE_INFO, DOUBLE_TYPE_INFO, STRING_TYPE_INFO),
+ "first, id, score, last")
+ tableEnv = tEnv
+ }
+
@Test
def testToString(): Unit = {
val env = StreamExecutionEnvironment.createLocalEnvironment()
@@ -75,6 +98,39 @@ class FlinkRelOptUtilTest {
}
@Test
+ def testGetDigestWithDynamicFunction(): Unit = {
+ val table = tableEnv.sqlQuery(
+ """
+ |(SELECT id AS random FROM MyTable ORDER BY rand() LIMIT 1)
+ |INTERSECT
+ |(SELECT id AS random FROM MyTable ORDER BY rand() LIMIT 1)
+ |INTERSECT
+ |(SELECT id AS random FROM MyTable ORDER BY rand() LIMIT 1)
+ """.stripMargin)
+ val rel = TableTestUtil.toRelNode(table)
+ val expected = TableTestUtil.readFromResource("/digest/testGetDigestWithDynamicFunction.out")
+ assertEquals(expected, FlinkRelOptUtil.getDigest(rel))
+ }
+
+ @Test
+ def testGetDigestWithDynamicFunctionView(): Unit = {
+ val view = tableEnv.sqlQuery("SELECT id AS random FROM MyTable ORDER BY rand() LIMIT 1")
+ tableEnv.registerTable("MyView", view)
+ val table = tableEnv.sqlQuery(
+ """
+ |(SELECT * FROM MyView)
+ |INTERSECT
+ |(SELECT * FROM MyView)
+ |INTERSECT
+ |(SELECT * FROM MyView)
+ """.stripMargin)
+ val rel = TableTestUtil.toRelNode(table).accept(new ExpandTableScanShuttle())
+ val expected = TableTestUtil.readFromResource(
+ "/digest/testGetDigestWithDynamicFunctionView.out")
+ assertEquals(expected, FlinkRelOptUtil.getDigest(rel))
+ }
+
+ @Test
def testMergeRowTimeAndNone(): Unit = {
val none = MiniBatchInterval.NONE
val rowtime = MiniBatchInterval(1000L, MiniBatchMode.RowTime)
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/utils/RelDigestUtilTest.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/utils/RelDigestUtilTest.scala
deleted file mode 100644
index 9e003ed..0000000
--- a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/utils/RelDigestUtilTest.scala
+++ /dev/null
@@ -1,83 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.flink.table.planner.plan.utils
-
-import org.apache.flink.api.common.typeinfo.BasicTypeInfo.{DOUBLE_TYPE_INFO, INT_TYPE_INFO, STRING_TYPE_INFO}
-import org.apache.flink.api.java.typeutils.RowTypeInfo
-import org.apache.flink.table.api.internal.TableEnvironmentImpl
-import org.apache.flink.table.api.{EnvironmentSettings, TableEnvironment}
-import org.apache.flink.table.planner.runtime.utils.BatchTableEnvUtil
-import org.apache.flink.table.planner.runtime.utils.BatchTestBase.row
-import org.apache.flink.table.planner.utils.TableTestUtil
-
-import org.junit.Assert.assertEquals
-import org.junit.{Before, Test}
-
-import scala.collection.Seq
-
-class RelDigestUtilTest {
-
- var tableEnv: TableEnvironment = _
-
- @Before
- def before(): Unit = {
- val settings = EnvironmentSettings.newInstance().useBlinkPlanner().build()
- val tEnv = TableEnvironmentImpl.create(settings)
- BatchTableEnvUtil.registerCollection(
- tEnv,
- "MyTable",
- Seq(row("Mike", 1, 12.3, "Smith")),
- new RowTypeInfo(STRING_TYPE_INFO, INT_TYPE_INFO, DOUBLE_TYPE_INFO, STRING_TYPE_INFO),
- "first, id, score, last")
- tableEnv = tEnv
- }
-
- @Test
- def testGetDigestWithDynamicFunction(): Unit = {
- val table = tableEnv.sqlQuery(
- """
- |(SELECT id AS random FROM MyTable ORDER BY rand() LIMIT 1)
- |INTERSECT
- |(SELECT id AS random FROM MyTable ORDER BY rand() LIMIT 1)
- |INTERSECT
- |(SELECT id AS random FROM MyTable ORDER BY rand() LIMIT 1)
- """.stripMargin)
- val rel = TableTestUtil.toRelNode(table)
- val expected = TableTestUtil.readFromResource("/digest/testGetDigestWithDynamicFunction.out")
- assertEquals(expected, RelDigestUtil.getDigest(rel))
- }
-
- @Test
- def testGetDigestWithDynamicFunctionView(): Unit = {
- val view = tableEnv.sqlQuery("SELECT id AS random FROM MyTable ORDER BY rand() LIMIT 1")
- tableEnv.registerTable("MyView", view)
- val table = tableEnv.sqlQuery(
- """
- |(SELECT * FROM MyView)
- |INTERSECT
- |(SELECT * FROM MyView)
- |INTERSECT
- |(SELECT * FROM MyView)
- """.stripMargin)
- val rel = TableTestUtil.toRelNode(table).accept(new ExpandTableScanShuttle())
- val expected = TableTestUtil.readFromResource(
- "/digest/testGetDigestWithDynamicFunctionView.out")
- assertEquals(expected, RelDigestUtil.getDigest(rel))
- }
-
-}