You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ja...@apache.org on 2023/03/28 08:17:53 UTC
[flink] branch master updated (d0f52112982 -> 2a3652d8563)
This is an automated email from the ASF dual-hosted git repository.
jark pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
from d0f52112982 [hotfix][doc][table] Fix the incorrect syntax description for explain statement set
new 46a46195618 [FLINK-31598][table] Replace the usage of deprecated TableEnvironment#registerTable() with createTemporaryView()
new 2a3652d8563 [FLINK-31598][table] Quotes the UUID table name to fix the failed tests
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../connector/hbase1/HBaseConnectorITCase.java | 2 +-
.../apache/flink/table/api/internal/TableImpl.java | 2 +-
.../flink/table/api/TableEnvironmentITCase.scala | 2 +-
.../flink/table/api/TableEnvironmentTest.scala | 4 +-
.../apache/flink/table/api/batch/ExplainTest.scala | 2 +-
.../flink/table/api/stream/ExplainTest.scala | 4 +-
.../plan/batch/sql/DagOptimizationTest.scala | 72 ++++++++---------
.../planner/plan/batch/sql/LegacySinkTest.scala | 2 +-
.../plan/stream/sql/DagOptimizationTest.scala | 84 ++++++++++----------
.../planner/plan/stream/sql/LegacySinkTest.scala | 8 +-
.../plan/stream/sql/MatchRecognizeTest.scala | 4 +-
.../validation/LegacyTableSinkValidationTest.scala | 2 +-
.../planner/plan/utils/FlinkRelOptUtilTest.scala | 6 +-
.../planner/runtime/batch/sql/CalcITCase.scala | 8 +-
.../planner/runtime/batch/table/CalcITCase.scala | 6 +-
.../runtime/harness/OverAggregateHarnessTest.scala | 14 ++--
.../runtime/stream/sql/AggregateITCase.scala | 92 +++++++++++-----------
.../runtime/stream/sql/AggregateRemoveITCase.scala | 2 +-
.../runtime/stream/sql/AsyncLookupJoinITCase.scala | 4 +-
.../planner/runtime/stream/sql/CalcITCase.scala | 20 ++---
.../runtime/stream/sql/CorrelateITCase.scala | 40 +++++-----
.../runtime/stream/sql/DeduplicateITCase.scala | 20 ++---
.../runtime/stream/sql/GroupWindowITCase.scala | 6 +-
.../runtime/stream/sql/IntervalJoinITCase.scala | 88 ++++++++++-----------
.../planner/runtime/stream/sql/JoinITCase.scala | 84 ++++++++++----------
.../runtime/stream/sql/Limit0RemoveITCase.scala | 26 +++---
.../planner/runtime/stream/sql/LimitITCase.scala | 6 +-
.../runtime/stream/sql/LookupJoinITCase.scala | 6 +-
.../runtime/stream/sql/MatchRecognizeITCase.scala | 28 +++----
.../runtime/stream/sql/OverAggregateITCase.scala | 48 +++++------
.../planner/runtime/stream/sql/RankITCase.scala | 52 ++++++------
.../stream/sql/SemiAntiJoinStreamITCase.scala | 52 ++++++------
.../runtime/stream/sql/SetOperatorsITCase.scala | 16 ++--
.../planner/runtime/stream/sql/SortITCase.scala | 14 ++--
.../runtime/stream/sql/SortLimitITCase.scala | 6 +-
.../runtime/stream/sql/SplitAggregateITCase.scala | 6 +-
.../runtime/stream/sql/TemporalSortITCase.scala | 6 +-
.../sql/TemporalTableFunctionJoinITCase.scala | 20 ++---
.../planner/runtime/stream/sql/UnnestITCase.scala | 22 +++---
.../stream/table/LegacyTableSinkITCase.scala | 2 +-
.../stream/table/TableToDataStreamITCase.scala | 4 +-
.../flink/table/planner/utils/TableTestBase.scala | 9 ++-
42 files changed, 451 insertions(+), 450 deletions(-)
[flink] 01/02: [FLINK-31598][table] Replace the usage of deprecated TableEnvironment#registerTable() with createTemporaryView()
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 46a46195618fddb1c511e01543583802972dc34c
Author: Jark Wu <ja...@apache.org>
AuthorDate: Wed Mar 22 23:30:45 2023 +0800
[FLINK-31598][table] Replace the usage of deprecated TableEnvironment#registerTable() with createTemporaryView()
---
.../connector/hbase1/HBaseConnectorITCase.java | 2 +-
.../apache/flink/table/api/internal/TableImpl.java | 2 +-
.../flink/table/api/TableEnvironmentITCase.scala | 2 +-
.../flink/table/api/TableEnvironmentTest.scala | 4 +-
.../apache/flink/table/api/batch/ExplainTest.scala | 2 +-
.../flink/table/api/stream/ExplainTest.scala | 4 +-
.../plan/batch/sql/DagOptimizationTest.scala | 72 ++++++++---------
.../planner/plan/batch/sql/LegacySinkTest.scala | 2 +-
.../plan/stream/sql/DagOptimizationTest.scala | 84 ++++++++++----------
.../planner/plan/stream/sql/LegacySinkTest.scala | 8 +-
.../plan/stream/sql/MatchRecognizeTest.scala | 4 +-
.../validation/LegacyTableSinkValidationTest.scala | 2 +-
.../planner/plan/utils/FlinkRelOptUtilTest.scala | 6 +-
.../planner/runtime/batch/sql/CalcITCase.scala | 8 +-
.../planner/runtime/batch/table/CalcITCase.scala | 6 +-
.../runtime/harness/OverAggregateHarnessTest.scala | 14 ++--
.../runtime/stream/sql/AggregateITCase.scala | 92 +++++++++++-----------
.../runtime/stream/sql/AggregateRemoveITCase.scala | 2 +-
.../runtime/stream/sql/AsyncLookupJoinITCase.scala | 4 +-
.../planner/runtime/stream/sql/CalcITCase.scala | 20 ++---
.../runtime/stream/sql/CorrelateITCase.scala | 40 +++++-----
.../runtime/stream/sql/DeduplicateITCase.scala | 20 ++---
.../runtime/stream/sql/GroupWindowITCase.scala | 6 +-
.../runtime/stream/sql/IntervalJoinITCase.scala | 88 ++++++++++-----------
.../planner/runtime/stream/sql/JoinITCase.scala | 84 ++++++++++----------
.../runtime/stream/sql/Limit0RemoveITCase.scala | 26 +++---
.../planner/runtime/stream/sql/LimitITCase.scala | 6 +-
.../runtime/stream/sql/LookupJoinITCase.scala | 6 +-
.../runtime/stream/sql/MatchRecognizeITCase.scala | 28 +++----
.../runtime/stream/sql/OverAggregateITCase.scala | 48 +++++------
.../planner/runtime/stream/sql/RankITCase.scala | 52 ++++++------
.../stream/sql/SemiAntiJoinStreamITCase.scala | 48 +++++------
.../runtime/stream/sql/SetOperatorsITCase.scala | 16 ++--
.../planner/runtime/stream/sql/SortITCase.scala | 14 ++--
.../runtime/stream/sql/SortLimitITCase.scala | 6 +-
.../runtime/stream/sql/SplitAggregateITCase.scala | 6 +-
.../runtime/stream/sql/TemporalSortITCase.scala | 6 +-
.../sql/TemporalTableFunctionJoinITCase.scala | 20 ++---
.../planner/runtime/stream/sql/UnnestITCase.scala | 22 +++---
.../stream/table/LegacyTableSinkITCase.scala | 2 +-
.../stream/table/TableToDataStreamITCase.scala | 4 +-
.../flink/table/planner/utils/TableTestBase.scala | 6 +-
42 files changed, 447 insertions(+), 447 deletions(-)
diff --git a/flink-connectors/flink-connector-hbase-1.4/src/test/java/org/apache/flink/connector/hbase1/HBaseConnectorITCase.java b/flink-connectors/flink-connector-hbase-1.4/src/test/java/org/apache/flink/connector/hbase1/HBaseConnectorITCase.java
index 35be358ad7e..ad5b5def0db 100644
--- a/flink-connectors/flink-connector-hbase-1.4/src/test/java/org/apache/flink/connector/hbase1/HBaseConnectorITCase.java
+++ b/flink-connectors/flink-connector-hbase-1.4/src/test/java/org/apache/flink/connector/hbase1/HBaseConnectorITCase.java
@@ -380,7 +380,7 @@ public class HBaseConnectorITCase extends HBaseTestBase {
String srcTableName = "src";
DataStream<Row> srcDs = execEnv.fromCollection(testData).returns(testTypeInfo);
Table in = tEnv.fromDataStream(srcDs, $("a"), $("b"), $("c"), $("proc").proctime());
- tEnv.registerTable(srcTableName, in);
+ tEnv.createTemporaryView(srcTableName, in);
// perform a temporal table join query
String dimJoinQuery =
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableImpl.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableImpl.java
index fbb070cb0d3..a5d5a3a44f4 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableImpl.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableImpl.java
@@ -487,7 +487,7 @@ public class TableImpl implements Table {
public String toString() {
if (tableName == null) {
tableName = "UnnamedTable$" + uniqueId.getAndIncrement();
- tableEnvironment.registerTable(tableName, this);
+ tableEnvironment.createTemporaryView(tableName, this);
}
return tableName;
}
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/TableEnvironmentITCase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/TableEnvironmentITCase.scala
index 567b6d87259..3ad40ae17e1 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/TableEnvironmentITCase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/TableEnvironmentITCase.scala
@@ -273,7 +273,7 @@ class TableEnvironmentITCase(tableEnvName: String, isStreaming: Boolean) extends
val t = streamEnv
.fromCollection(getPersonData)
.toTable(streamTableEnv, 'first, 'id, 'score, 'last)
- streamTableEnv.registerTable("MyTable", t)
+ streamTableEnv.createTemporaryView("MyTable", t)
val sink1Path = TestTableSourceSinks.createCsvTemporarySinkTable(
streamTableEnv,
new TableSchema(Array("first"), Array(STRING)),
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/TableEnvironmentTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/TableEnvironmentTest.scala
index 2a435f5568b..202013403c9 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/TableEnvironmentTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/TableEnvironmentTest.scala
@@ -79,7 +79,7 @@ class TableEnvironmentTest {
@Test
def testRegisterDataStream(): Unit = {
val table = env.fromElements[(Int, Long, String, Boolean)]().toTable(tableEnv, 'a, 'b, 'c, 'd)
- tableEnv.registerTable("MyTable", table)
+ tableEnv.createTemporaryView("MyTable", table)
val scanTable = tableEnv.from("MyTable")
val relNode = TableTestUtil.toRelNode(scanTable)
val actual = RelOptUtil.toString(relNode)
@@ -96,7 +96,7 @@ class TableEnvironmentTest {
@Test
def testSimpleQuery(): Unit = {
val table = env.fromElements[(Int, Long, String, Boolean)]().toTable(tableEnv, 'a, 'b, 'c, 'd)
- tableEnv.registerTable("MyTable", table)
+ tableEnv.createTemporaryView("MyTable", table)
val queryTable = tableEnv.sqlQuery("SELECT a, c, d FROM MyTable")
val relNode = TableTestUtil.toRelNode(queryTable)
val actual = RelOptUtil.toString(relNode, SqlExplainLevel.NO_ATTRIBUTES)
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/batch/ExplainTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/batch/ExplainTest.scala
index 7d8cdcab649..e6ed0bce82f 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/batch/ExplainTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/batch/ExplainTest.scala
@@ -101,7 +101,7 @@ class ExplainTest(extended: Boolean) extends TableTestBase {
def testExplainWithMultiSinks(): Unit = {
val stmtSet = util.tableEnv.createStatementSet()
val table = util.tableEnv.sqlQuery("SELECT a, COUNT(*) AS cnt FROM MyTable1 GROUP BY a")
- util.tableEnv.registerTable("TempTable", table)
+ util.tableEnv.createTemporaryView("TempTable", table)
val table1 = util.tableEnv.sqlQuery("SELECT * FROM TempTable WHERE cnt > 10")
val sink1 = util.createCollectTableSink(Array("a", "cnt"), Array(INT, LONG))
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/stream/ExplainTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/stream/ExplainTest.scala
index 9ee389c0666..39d2326e332 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/stream/ExplainTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/stream/ExplainTest.scala
@@ -101,7 +101,7 @@ class ExplainTest(extended: Boolean) extends TableTestBase {
def testExplainWithMultiSinks(): Unit = {
val stmtSet = util.tableEnv.createStatementSet()
val table = util.tableEnv.sqlQuery("SELECT a, COUNT(*) AS cnt FROM MyTable1 GROUP BY a")
- util.tableEnv.registerTable("TempTable", table)
+ util.tableEnv.createTemporaryView("TempTable", table)
val table1 = util.tableEnv.sqlQuery("SELECT * FROM TempTable WHERE cnt > 10")
val upsertSink1 = util.createUpsertTableSink(Array(0), Array("a", "cnt"), Array(INT, LONG))
@@ -145,7 +145,7 @@ class ExplainTest(extended: Boolean) extends TableTestBase {
| AND T3.rowtime > T4.rowtime - INTERVAL '5' MINUTE
| AND T3.rowtime < T4.rowtime + INTERVAL '3' MINUTE
""".stripMargin)
- util.tableEnv.registerTable("TempTable", table)
+ util.tableEnv.createTemporaryView("TempTable", table)
val table1 = util.tableEnv.sqlQuery("""
|SELECT id1, LISTAGG(text, '#')
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/DagOptimizationTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/DagOptimizationTest.scala
index 196cab28344..8f4a4dd55d9 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/DagOptimizationTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/DagOptimizationTest.scala
@@ -50,15 +50,15 @@ class DagOptimizationTest extends TableTestBase {
@Test
def testSingleSink2(): Unit = {
val table1 = util.tableEnv.sqlQuery("SELECT a as a1, b FROM MyTable WHERE a <= 10")
- util.tableEnv.registerTable("table1", table1)
+ util.tableEnv.createTemporaryView("table1", table1)
val table2 = util.tableEnv.sqlQuery("SELECT a, b, c FROM MyTable WHERE a >= 0")
- util.tableEnv.registerTable("table2", table2)
+ util.tableEnv.createTemporaryView("table2", table2)
val table3 = util.tableEnv.sqlQuery("SELECT a AS a2, c FROM table2 WHERE b >= 5")
- util.tableEnv.registerTable("table3", table3)
+ util.tableEnv.createTemporaryView("table3", table3)
val table4 = util.tableEnv.sqlQuery("SELECT a AS a3, c as c1 FROM table2 WHERE b < 5")
- util.tableEnv.registerTable("table4", table4)
+ util.tableEnv.createTemporaryView("table4", table4)
val table5 = util.tableEnv.sqlQuery("SELECT a1, b, c as c2 FROM table1, table3 WHERE a1 = a2")
- util.tableEnv.registerTable("table5", table5)
+ util.tableEnv.createTemporaryView("table5", table5)
val table6 = util.tableEnv.sqlQuery("SELECT a1, b, c1 FROM table4, table5 WHERE a1 = a3")
val appendSink = util.createCollectTableSink(Array("a1", "b", "c1"), Array(INT, LONG, STRING))
@@ -69,9 +69,9 @@ class DagOptimizationTest extends TableTestBase {
def testSingleSink3(): Unit = {
util.addDataStream[(Int, Long, String, Double, Boolean)]("MyTable2", 'a, 'b, 'c, 'd, 'e)
val table1 = util.tableEnv.sqlQuery("SELECT a AS a1, b as b1 FROM MyTable WHERE a <= 10")
- util.tableEnv.registerTable("table1", table1)
+ util.tableEnv.createTemporaryView("table1", table1)
val table2 = util.tableEnv.sqlQuery("SELECT a, b1 FROM table1, MyTable2 WHERE a = a1")
- util.tableEnv.registerTable("table2", table2)
+ util.tableEnv.createTemporaryView("table2", table2)
val table3 = util.tableEnv.sqlQuery("SELECT * FROM table1 UNION ALL SELECT * FROM table2")
val appendSink = util.createCollectTableSink(Array("a1", "b1"), Array(INT, LONG))
@@ -81,17 +81,17 @@ class DagOptimizationTest extends TableTestBase {
@Test
def testSingleSink4(): Unit = {
val table1 = util.tableEnv.sqlQuery("SELECT a as a1, b FROM MyTable WHERE a <= 10")
- util.tableEnv.registerTable("table1", table1)
+ util.tableEnv.createTemporaryView("table1", table1)
val table2 = util.tableEnv.sqlQuery("SELECT a, b, c FROM MyTable WHERE a >= 0")
- util.tableEnv.registerTable("table2", table2)
+ util.tableEnv.createTemporaryView("table2", table2)
val table3 = util.tableEnv.sqlQuery("SELECT a AS a2, c FROM table2 WHERE b >= 5")
- util.tableEnv.registerTable("table3", table3)
+ util.tableEnv.createTemporaryView("table3", table3)
val table4 = util.tableEnv.sqlQuery("SELECT a AS a3, c AS c1 FROM table2 WHERE b < 5")
- util.tableEnv.registerTable("table4", table4)
+ util.tableEnv.createTemporaryView("table4", table4)
val table5 = util.tableEnv.sqlQuery("SELECT a1, b, c AS c2 from table1, table3 WHERE a1 = a2")
- util.tableEnv.registerTable("table5", table5)
+ util.tableEnv.createTemporaryView("table5", table5)
val table6 = util.tableEnv.sqlQuery("SELECT a3, b as b1, c1 FROM table4, table5 WHERE a1 = a3")
- util.tableEnv.registerTable("table6", table6)
+ util.tableEnv.createTemporaryView("table6", table6)
val table7 = util.tableEnv.sqlQuery("SELECT a1, b1, c1 FROM table1, table6 WHERE a1 = a3")
val sink = util.createCollectTableSink(Array("a", "b", "c"), Array(INT, LONG, STRING))
@@ -133,7 +133,7 @@ class DagOptimizationTest extends TableTestBase {
RelNodeBlockPlanBuilder.TABLE_OPTIMIZER_REUSE_OPTIMIZE_BLOCK_WITH_DIGEST_ENABLED,
Boolean.box(true))
val table1 = util.tableEnv.sqlQuery("SELECT SUM(a) AS sum_a, c FROM MyTable GROUP BY c")
- util.tableEnv.registerTable("table1", table1)
+ util.tableEnv.createTemporaryView("table1", table1)
val table2 = util.tableEnv.sqlQuery("SELECT SUM(sum_a) AS total_sum FROM table1")
val table3 = util.tableEnv.sqlQuery("SELECT MIN(sum_a) AS total_min FROM table1")
@@ -157,9 +157,9 @@ class DagOptimizationTest extends TableTestBase {
util.addTableSource[(Int, Long, String, Double, Boolean)]("MyTable2", 'a, 'b, 'c, 'd, 'e)
val table1 = util.tableEnv.sqlQuery("SELECT a as a1, b as b1 FROM MyTable WHERE a <= 10")
- util.tableEnv.registerTable("table1", table1)
+ util.tableEnv.createTemporaryView("table1", table1)
val table2 = util.tableEnv.sqlQuery("SELECT a, b1 from table1, MyTable2 where a = a1")
- util.tableEnv.registerTable("table2", table2)
+ util.tableEnv.createTemporaryView("table2", table2)
val table3 = util.tableEnv.sqlQuery("SELECT * FROM table1 UNION ALL SELECT * FROM table2")
val sink1 = util.createCollectTableSink(Array("a", "b1"), Array(INT, LONG))
@@ -182,9 +182,9 @@ class DagOptimizationTest extends TableTestBase {
util.addTableSource[(Int, Long, String, Double, Boolean)]("MyTable2", 'a, 'b, 'c, 'd, 'e)
val table1 = util.tableEnv.sqlQuery("SELECT a AS a1, b AS b1 FROM MyTable WHERE a <= 10")
- util.tableEnv.registerTable("table1", table1)
+ util.tableEnv.createTemporaryView("table1", table1)
val table2 = util.tableEnv.sqlQuery("SELECT a, b1 FROM table1, MyTable2 WHERE a = a1")
- util.tableEnv.registerTable("table2", table2)
+ util.tableEnv.createTemporaryView("table2", table2)
val table3 = util.tableEnv.sqlQuery("SELECT * FROM table1 UNION ALL SELECT * FROM table2")
val sink1 = util.createCollectTableSink(Array("a", "b1"), Array(INT, LONG))
@@ -206,15 +206,15 @@ class DagOptimizationTest extends TableTestBase {
Boolean.box(true))
val table1 = util.tableEnv.sqlQuery("SELECT a as a1, b FROM MyTable WHERE a <= 10")
- util.tableEnv.registerTable("table1", table1)
+ util.tableEnv.createTemporaryView("table1", table1)
val table2 = util.tableEnv.sqlQuery("SELECT a, b, c FROM MyTable WHERE a >= 0")
- util.tableEnv.registerTable("table2", table2)
+ util.tableEnv.createTemporaryView("table2", table2)
val table3 = util.tableEnv.sqlQuery("SELECT a as a2, c FROM table2 WHERE b >= 5")
- util.tableEnv.registerTable("table3", table3)
+ util.tableEnv.createTemporaryView("table3", table3)
val table4 = util.tableEnv.sqlQuery("SELECT a as a3, c as c1 FROM table2 WHERE b < 5")
- util.tableEnv.registerTable("table4", table4)
+ util.tableEnv.createTemporaryView("table4", table4)
val table5 = util.tableEnv.sqlQuery("SELECT a1, b, c as c2 FROM table1, table3 WHERE a1 = a2")
- util.tableEnv.registerTable("table5", table5)
+ util.tableEnv.createTemporaryView("table5", table5)
val table6 = util.tableEnv.sqlQuery("SELECT a1, b, c1 FROM table4, table5 WHERE a1 = a3")
val sink1 = util.createCollectTableSink(Array("a1", "b", "c2"), Array(INT, LONG, STRING))
@@ -237,7 +237,7 @@ class DagOptimizationTest extends TableTestBase {
// test with non-deterministic udf
util.tableEnv.registerFunction("random_udf", new NonDeterministicUdf())
val table1 = util.tableEnv.sqlQuery("SELECT random_udf(a) AS a, c FROM MyTable")
- util.tableEnv.registerTable("table1", table1)
+ util.tableEnv.createTemporaryView("table1", table1)
val table2 = util.tableEnv.sqlQuery("SELECT SUM(a) AS total_sum FROM table1")
val table3 = util.tableEnv.sqlQuery("SELECT MIN(a) AS total_min FROM table1")
@@ -260,13 +260,13 @@ class DagOptimizationTest extends TableTestBase {
Boolean.box(false))
val table1 = util.tableEnv.sqlQuery("SELECT a, b, c FROM MyTable WHERE c LIKE '%hello%'")
- util.tableEnv.registerTable("TempTable1", table1)
+ util.tableEnv.createTemporaryView("TempTable1", table1)
val sink1 = util.createCollectTableSink(Array("a", "b", "c"), Array(INT, LONG, STRING))
util.tableEnv.asInstanceOf[TableEnvironmentInternal].registerTableSinkInternal("sink1", sink1)
stmtSet.addInsert("sink1", table1)
val table2 = util.tableEnv.sqlQuery("SELECT a, b, c FROM MyTable WHERE c LIKE '%world%'")
- util.tableEnv.registerTable("TempTable2", table2)
+ util.tableEnv.createTemporaryView("TempTable2", table2)
val sqlQuery =
"""
@@ -278,7 +278,7 @@ class DagOptimizationTest extends TableTestBase {
|GROUP BY b
""".stripMargin
val table3 = util.tableEnv.sqlQuery(sqlQuery)
- util.tableEnv.registerTable("TempTable3", table3)
+ util.tableEnv.createTemporaryView("TempTable3", table3)
val table4 = util.tableEnv.sqlQuery("SELECT b, cnt FROM TempTable3 WHERE b < 4")
val sink2 = util.createCollectTableSink(Array("b", "cnt"), Array(LONG, LONG))
@@ -306,20 +306,20 @@ class DagOptimizationTest extends TableTestBase {
|WHERE b >= UNIX_TIMESTAMP('${startTime}')
""".stripMargin
val table1 = util.tableEnv.sqlQuery(sqlQuery1)
- util.tableEnv.registerTable("table1", table1)
+ util.tableEnv.createTemporaryView("table1", table1)
val sqlQuery2 =
"SELECT a, b, c1 AS c FROM table1, LATERAL TABLE(split(c)) AS T(c1) WHERE c <> '' "
val table2 = util.tableEnv.sqlQuery(sqlQuery2)
- util.tableEnv.registerTable("table2", table2)
+ util.tableEnv.createTemporaryView("table2", table2)
val sqlQuery3 = "SELECT a, b, COUNT(DISTINCT c) AS total_c FROM table2 GROUP BY a, b"
val table3 = util.tableEnv.sqlQuery(sqlQuery3)
- util.tableEnv.registerTable("table3", table3)
+ util.tableEnv.createTemporaryView("table3", table3)
val sqlQuery4 = "SELECT a, total_c FROM table3 UNION ALL SELECT a, 0 AS total_c FROM table1"
val table4 = util.tableEnv.sqlQuery(sqlQuery4)
- util.tableEnv.registerTable("table4", table4)
+ util.tableEnv.createTemporaryView("table4", table4)
val sqlQuery5 = "SELECT * FROM table4 WHERE a > 50"
val table5 = util.tableEnv.sqlQuery(sqlQuery5)
@@ -345,7 +345,7 @@ class DagOptimizationTest extends TableTestBase {
val table =
util.tableEnv.sqlQuery("SELECT a, c FROM MyTable UNION ALL SELECT d, f FROM MyTable1")
- util.tableEnv.registerTable("TempTable", table)
+ util.tableEnv.createTemporaryView("TempTable", table)
val table1 = util.tableEnv.sqlQuery("SELECT SUM(a) AS total_sum FROM TempTable")
val sink1 = util.createCollectTableSink(Array("total_sum"), Array(INT))
@@ -380,7 +380,7 @@ class DagOptimizationTest extends TableTestBase {
|SELECT a, c FROM MyTable2
""".stripMargin
val table = util.tableEnv.sqlQuery(sqlQuery1)
- util.tableEnv.registerTable("TempTable", table)
+ util.tableEnv.createTemporaryView("TempTable", table)
val table1 = util.tableEnv.sqlQuery("SELECT SUM(a) AS total_sum FROM TempTable")
val sink1 = util.createCollectTableSink(Array("total_sum"), Array(INT))
@@ -411,7 +411,7 @@ class DagOptimizationTest extends TableTestBase {
val sqlQuery1 = "SELECT a, c FROM MyTable UNION ALL SELECT d, f FROM MyTable1"
val table = util.tableEnv.sqlQuery(sqlQuery1)
- util.tableEnv.registerTable("TempTable", table)
+ util.tableEnv.createTemporaryView("TempTable", table)
val sink1 = util.createCollectTableSink(Array("a", "c"), Array(INT, STRING))
util.tableEnv.asInstanceOf[TableEnvironmentInternal].registerTableSinkInternal("sink1", sink1)
@@ -419,7 +419,7 @@ class DagOptimizationTest extends TableTestBase {
val sqlQuery2 = "SELECT a, c FROM TempTable UNION ALL SELECT a, c FROM MyTable2"
val table1 = util.tableEnv.sqlQuery(sqlQuery2)
- util.tableEnv.registerTable("TempTable1", table1)
+ util.tableEnv.createTemporaryView("TempTable1", table1)
val table2 = util.tableEnv.sqlQuery("SELECT SUM(a) AS total_sum FROM TempTable1")
val sink2 = util.createCollectTableSink(Array("total_sum"), Array(INT))
@@ -451,7 +451,7 @@ class DagOptimizationTest extends TableTestBase {
|SELECT a, c FROM MyTable2
""".stripMargin
val table = util.tableEnv.sqlQuery(sqlQuery)
- util.tableEnv.registerTable("TempTable", table)
+ util.tableEnv.createTemporaryView("TempTable", table)
val table1 = util.tableEnv.sqlQuery("SELECT SUM(a) AS total_sum FROM TempTable")
val sink1 = util.createCollectTableSink(Array("total_sum"), Array(INT))
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/LegacySinkTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/LegacySinkTest.scala
index 3da186709af..a1a8b9bf05b 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/LegacySinkTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/LegacySinkTest.scala
@@ -49,7 +49,7 @@ class LegacySinkTest extends TableTestBase {
RelNodeBlockPlanBuilder.TABLE_OPTIMIZER_REUSE_OPTIMIZE_BLOCK_WITH_DIGEST_ENABLED,
Boolean.box(true))
val table1 = util.tableEnv.sqlQuery("SELECT SUM(a) AS sum_a, c FROM MyTable GROUP BY c")
- util.tableEnv.registerTable("table1", table1)
+ util.tableEnv.createTemporaryView("table1", table1)
val table2 = util.tableEnv.sqlQuery("SELECT SUM(sum_a) AS total_sum FROM table1")
val table3 = util.tableEnv.sqlQuery("SELECT MIN(sum_a) AS total_min FROM table1")
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/DagOptimizationTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/DagOptimizationTest.scala
index fa26873ef62..c98a049c5f0 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/DagOptimizationTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/DagOptimizationTest.scala
@@ -46,15 +46,15 @@ class DagOptimizationTest extends TableTestBase {
@Test
def testSingleSink2(): Unit = {
val table1 = util.tableEnv.sqlQuery("SELECT a as a1, b FROM MyTable WHERE a <= 10")
- util.tableEnv.registerTable("table1", table1)
+ util.tableEnv.createTemporaryView("table1", table1)
val table2 = util.tableEnv.sqlQuery("SELECT a, b, c FROM MyTable WHERE a >= 0")
- util.tableEnv.registerTable("table2", table2)
+ util.tableEnv.createTemporaryView("table2", table2)
val table3 = util.tableEnv.sqlQuery("SELECT a AS a2, c FROM table2 WHERE b >= 5")
- util.tableEnv.registerTable("table3", table3)
+ util.tableEnv.createTemporaryView("table3", table3)
val table4 = util.tableEnv.sqlQuery("SELECT a AS a3, c as c1 FROM table2 WHERE b < 5")
- util.tableEnv.registerTable("table4", table4)
+ util.tableEnv.createTemporaryView("table4", table4)
val table5 = util.tableEnv.sqlQuery("SELECT a1, b, c as c2 FROM table1, table3 WHERE a1 = a2")
- util.tableEnv.registerTable("table5", table5)
+ util.tableEnv.createTemporaryView("table5", table5)
val table6 = util.tableEnv.sqlQuery("SELECT a1, b, c1 FROM table4, table5 WHERE a1 = a3")
val appendSink = util.createAppendTableSink(Array("a1", "b", "c1"), Array(INT, LONG, STRING))
@@ -65,9 +65,9 @@ class DagOptimizationTest extends TableTestBase {
def testSingleSink3(): Unit = {
util.addDataStream[(Int, Long, String, Double, Boolean)]("MyTable2", 'a, 'b, 'c, 'd, 'e)
val table1 = util.tableEnv.sqlQuery("SELECT a AS a1, b as b1 FROM MyTable WHERE a <= 10")
- util.tableEnv.registerTable("table1", table1)
+ util.tableEnv.createTemporaryView("table1", table1)
val table2 = util.tableEnv.sqlQuery("SELECT a, b1 FROM table1, MyTable2 WHERE a = a1")
- util.tableEnv.registerTable("table2", table2)
+ util.tableEnv.createTemporaryView("table2", table2)
val table3 = util.tableEnv.sqlQuery("SELECT * FROM table1 UNION ALL SELECT * FROM table2")
val appendSink = util.createAppendTableSink(Array("a1", "b1"), Array(INT, LONG))
@@ -77,17 +77,17 @@ class DagOptimizationTest extends TableTestBase {
@Test
def testSingleSink4(): Unit = {
val table1 = util.tableEnv.sqlQuery("SELECT a as a1, b FROM MyTable WHERE a <= 10")
- util.tableEnv.registerTable("table1", table1)
+ util.tableEnv.createTemporaryView("table1", table1)
val table2 = util.tableEnv.sqlQuery("SELECT a, b, c FROM MyTable WHERE a >= 0")
- util.tableEnv.registerTable("table2", table2)
+ util.tableEnv.createTemporaryView("table2", table2)
val table3 = util.tableEnv.sqlQuery("SELECT a AS a2, c FROM table2 WHERE b >= 5")
- util.tableEnv.registerTable("table3", table3)
+ util.tableEnv.createTemporaryView("table3", table3)
val table4 = util.tableEnv.sqlQuery("SELECT a AS a3, c AS c1 FROM table2 WHERE b < 5")
- util.tableEnv.registerTable("table4", table4)
+ util.tableEnv.createTemporaryView("table4", table4)
val table5 = util.tableEnv.sqlQuery("SELECT a1, b, c AS c2 from table1, table3 WHERE a1 = a2")
- util.tableEnv.registerTable("table5", table5)
+ util.tableEnv.createTemporaryView("table5", table5)
val table6 = util.tableEnv.sqlQuery("SELECT a3, b as b1, c1 FROM table4, table5 WHERE a1 = a3")
- util.tableEnv.registerTable("table6", table6)
+ util.tableEnv.createTemporaryView("table6", table6)
val table7 = util.tableEnv.sqlQuery("SELECT a1, b1, c1 FROM table1, table6 WHERE a1 = a3")
val appendSink = util.createAppendTableSink(Array("a", "b", "c"), Array(INT, LONG, STRING))
@@ -133,7 +133,7 @@ class DagOptimizationTest extends TableTestBase {
RelNodeBlockPlanBuilder.TABLE_OPTIMIZER_REUSE_OPTIMIZE_BLOCK_WITH_DIGEST_ENABLED,
Boolean.box(true))
val table1 = util.tableEnv.sqlQuery("SELECT SUM(a) AS sum_a, c FROM MyTable GROUP BY c")
- util.tableEnv.registerTable("table1", table1)
+ util.tableEnv.createTemporaryView("table1", table1)
val table2 = util.tableEnv.sqlQuery("SELECT SUM(sum_a) AS total_sum FROM table1")
val table3 = util.tableEnv.sqlQuery("SELECT MIN(sum_a) AS total_min FROM table1")
@@ -161,9 +161,9 @@ class DagOptimizationTest extends TableTestBase {
util.addTableSource[(Int, Long, String, Double, Boolean)]("MyTable2", 'a, 'b, 'c, 'd, 'e)
val table1 = util.tableEnv.sqlQuery("SELECT a as a1, b as b1 FROM MyTable WHERE a <= 10")
- util.tableEnv.registerTable("table1", table1)
+ util.tableEnv.createTemporaryView("table1", table1)
val table2 = util.tableEnv.sqlQuery("SELECT a, b1 from table1, MyTable2 where a = a1")
- util.tableEnv.registerTable("table2", table2)
+ util.tableEnv.createTemporaryView("table2", table2)
val table3 = util.tableEnv.sqlQuery("SELECT * FROM table1 UNION ALL SELECT * FROM table2")
val appendSink1 = util.createAppendTableSink(Array("a", "b1"), Array(INT, LONG))
@@ -190,9 +190,9 @@ class DagOptimizationTest extends TableTestBase {
util.addTableSource[(Int, Long, String, Double, Boolean)]("MyTable2", 'a, 'b, 'c, 'd, 'e)
val table1 = util.tableEnv.sqlQuery("SELECT a AS a1, b AS b1 FROM MyTable WHERE a <= 10")
- util.tableEnv.registerTable("table1", table1)
+ util.tableEnv.createTemporaryView("table1", table1)
val table2 = util.tableEnv.sqlQuery("SELECT a, b1 FROM table1, MyTable2 WHERE a = a1")
- util.tableEnv.registerTable("table2", table2)
+ util.tableEnv.createTemporaryView("table2", table2)
val table3 = util.tableEnv.sqlQuery("SELECT * FROM table1 UNION ALL SELECT * FROM table2")
val appendSink1 = util.createAppendTableSink(Array("a", "b1"), Array(INT, LONG))
@@ -214,15 +214,15 @@ class DagOptimizationTest extends TableTestBase {
def testMultiSinks4(): Unit = {
val stmtSet = util.tableEnv.createStatementSet()
val table1 = util.tableEnv.sqlQuery("SELECT a as a1, b FROM MyTable WHERE a <= 10")
- util.tableEnv.registerTable("table1", table1)
+ util.tableEnv.createTemporaryView("table1", table1)
val table2 = util.tableEnv.sqlQuery("SELECT a, b, c FROM MyTable WHERE a >= 0")
- util.tableEnv.registerTable("table2", table2)
+ util.tableEnv.createTemporaryView("table2", table2)
val table3 = util.tableEnv.sqlQuery("SELECT a as a2, c FROM table2 WHERE b >= 5")
- util.tableEnv.registerTable("table3", table3)
+ util.tableEnv.createTemporaryView("table3", table3)
val table4 = util.tableEnv.sqlQuery("SELECT a as a3, c as c1 FROM table2 WHERE b < 5")
- util.tableEnv.registerTable("table4", table4)
+ util.tableEnv.createTemporaryView("table4", table4)
val table5 = util.tableEnv.sqlQuery("SELECT a1, b, c as c2 FROM table1, table3 WHERE a1 = a2")
- util.tableEnv.registerTable("table5", table5)
+ util.tableEnv.createTemporaryView("table5", table5)
val table6 = util.tableEnv.sqlQuery("SELECT a1, b, c1 FROM table4, table5 WHERE a1 = a3")
val appendSink1 = util.createAppendTableSink(Array("a1", "b", "c2"), Array(INT, LONG, STRING))
@@ -249,7 +249,7 @@ class DagOptimizationTest extends TableTestBase {
// test with non-deterministic udf
util.tableEnv.registerFunction("random_udf", new NonDeterministicUdf())
val table1 = util.tableEnv.sqlQuery("SELECT random_udf(a) AS a, c FROM MyTable")
- util.tableEnv.registerTable("table1", table1)
+ util.tableEnv.createTemporaryView("table1", table1)
val table2 = util.tableEnv.sqlQuery("SELECT SUM(a) AS total_sum FROM table1")
val table3 = util.tableEnv.sqlQuery("SELECT MIN(a) AS total_min FROM table1")
@@ -281,20 +281,20 @@ class DagOptimizationTest extends TableTestBase {
|WHERE b >= UNIX_TIMESTAMP('${startTime}')
""".stripMargin
val table1 = util.tableEnv.sqlQuery(sqlQuery1)
- util.tableEnv.registerTable("table1", table1)
+ util.tableEnv.createTemporaryView("table1", table1)
val sqlQuery2 =
"SELECT a, b, c1 AS c FROM table1, LATERAL TABLE(split(c)) AS T(c1) WHERE c <> '' "
val table2 = util.tableEnv.sqlQuery(sqlQuery2)
- util.tableEnv.registerTable("table2", table2)
+ util.tableEnv.createTemporaryView("table2", table2)
val sqlQuery3 = "SELECT a, b, COUNT(DISTINCT c) AS total_c FROM table2 GROUP BY a, b"
val table3 = util.tableEnv.sqlQuery(sqlQuery3)
- util.tableEnv.registerTable("table3", table3)
+ util.tableEnv.createTemporaryView("table3", table3)
val sqlQuery4 = "SELECT a, total_c FROM table3 UNION ALL SELECT a, 0 AS total_c FROM table1"
val table4 = util.tableEnv.sqlQuery(sqlQuery4)
- util.tableEnv.registerTable("table4", table4)
+ util.tableEnv.createTemporaryView("table4", table4)
val sqlQuery5 = "SELECT * FROM table4 WHERE a > 50"
val table5 = util.tableEnv.sqlQuery(sqlQuery5)
@@ -324,7 +324,7 @@ class DagOptimizationTest extends TableTestBase {
val table =
util.tableEnv.sqlQuery("SELECT a, c FROM MyTable UNION ALL SELECT d, f FROM MyTable1")
- util.tableEnv.registerTable("TempTable", table)
+ util.tableEnv.createTemporaryView("TempTable", table)
val table1 = util.tableEnv.sqlQuery("SELECT SUM(a) AS total_sum FROM TempTable")
val upsertSink = util.createUpsertTableSink(Array(), Array("total_sum"), Array(INT))
@@ -363,7 +363,7 @@ class DagOptimizationTest extends TableTestBase {
|SELECT a, c FROM MyTable2
""".stripMargin
val table = util.tableEnv.sqlQuery(sqlQuery1)
- util.tableEnv.registerTable("TempTable", table)
+ util.tableEnv.createTemporaryView("TempTable", table)
val table1 = util.tableEnv.sqlQuery("SELECT SUM(a) AS total_sum FROM TempTable")
val retractSink1 = util.createRetractTableSink(Array("total_sum"), Array(INT))
@@ -400,7 +400,7 @@ class DagOptimizationTest extends TableTestBase {
val sqlQuery1 = "SELECT a, c FROM MyTable UNION ALL SELECT d, f FROM MyTable1"
val table = util.tableEnv.sqlQuery(sqlQuery1)
- util.tableEnv.registerTable("TempTable", table)
+ util.tableEnv.createTemporaryView("TempTable", table)
val appendSink = util.createAppendTableSink(Array("a", "c"), Array(INT, STRING))
util.tableEnv
@@ -410,7 +410,7 @@ class DagOptimizationTest extends TableTestBase {
val sqlQuery2 = "SELECT a, c FROM TempTable UNION ALL SELECT a, c FROM MyTable2"
val table1 = util.tableEnv.sqlQuery(sqlQuery2)
- util.tableEnv.registerTable("TempTable1", table1)
+ util.tableEnv.createTemporaryView("TempTable1", table1)
val table2 = util.tableEnv.sqlQuery("SELECT SUM(a) AS total_sum FROM TempTable1")
val retractSink = util.createRetractTableSink(Array("total_sum"), Array(INT))
@@ -446,7 +446,7 @@ class DagOptimizationTest extends TableTestBase {
|SELECT a, c FROM MyTable2
""".stripMargin
val table = util.tableEnv.sqlQuery(sqlQuery)
- util.tableEnv.registerTable("TempTable", table)
+ util.tableEnv.createTemporaryView("TempTable", table)
val table1 = util.tableEnv.sqlQuery("SELECT SUM(a) AS total_sum FROM TempTable")
val upsertSink = util.createUpsertTableSink(Array(), Array("total_sum"), Array(INT))
@@ -484,7 +484,7 @@ class DagOptimizationTest extends TableTestBase {
def testUpdateAsRetractConsumedAtSinkBlock(): Unit = {
val stmtSet = util.tableEnv.createStatementSet()
val table = util.tableEnv.sqlQuery("SELECT a, b, c FROM MyTable")
- util.tableEnv.registerTable("TempTable", table)
+ util.tableEnv.createTemporaryView("TempTable", table)
val sqlQuery =
s"""
@@ -524,7 +524,7 @@ class DagOptimizationTest extends TableTestBase {
|WHERE rank_num <= 10
""".stripMargin
val table = util.tableEnv.sqlQuery(sqlQuery)
- util.tableEnv.registerTable("TempTable", table)
+ util.tableEnv.createTemporaryView("TempTable", table)
val table1 = util.tableEnv.sqlQuery("SELECT a FROM TempTable WHERE a > 6")
val retractSink = util.createRetractTableSink(Array("a"), Array(INT))
@@ -551,7 +551,7 @@ class DagOptimizationTest extends TableTestBase {
Boolean.box(false))
val table1 = util.tableEnv.sqlQuery("SELECT a, b, c FROM MyTable WHERE c LIKE '%hello%'")
- util.tableEnv.registerTable("TempTable1", table1)
+ util.tableEnv.createTemporaryView("TempTable1", table1)
val appendSink = util.createAppendTableSink(Array("a", "b", "c"), Array(INT, LONG, STRING))
util.tableEnv
.asInstanceOf[TableEnvironmentInternal]
@@ -559,7 +559,7 @@ class DagOptimizationTest extends TableTestBase {
stmtSet.addInsert("appendSink", table1)
val table2 = util.tableEnv.sqlQuery("SELECT a, b, c FROM MyTable WHERE c LIKE '%world%'")
- util.tableEnv.registerTable("TempTable2", table2)
+ util.tableEnv.createTemporaryView("TempTable2", table2)
val sqlQuery =
"""
@@ -571,7 +571,7 @@ class DagOptimizationTest extends TableTestBase {
|GROUP BY b
""".stripMargin
val table3 = util.tableEnv.sqlQuery(sqlQuery)
- util.tableEnv.registerTable("TempTable3", table3)
+ util.tableEnv.createTemporaryView("TempTable3", table3)
val table4 = util.tableEnv.sqlQuery("SELECT b, cnt FROM TempTable3 WHERE b < 4")
val retractSink = util.createRetractTableSink(Array("b", "cnt"), Array(LONG, LONG))
@@ -598,7 +598,7 @@ class DagOptimizationTest extends TableTestBase {
Boolean.box(false))
val table1 = util.tableEnv.sqlQuery("SELECT a, b, c FROM MyTable WHERE c LIKE '%hello%'")
- util.tableEnv.registerTable("TempTable1", table1)
+ util.tableEnv.createTemporaryView("TempTable1", table1)
val appendSink = util.createAppendTableSink(Array("a", "b", "c"), Array(INT, LONG, STRING))
util.tableEnv
.asInstanceOf[TableEnvironmentInternal]
@@ -606,7 +606,7 @@ class DagOptimizationTest extends TableTestBase {
stmtSet.addInsert("appendSink", table1)
val table2 = util.tableEnv.sqlQuery("SELECT a, b, c FROM MyTable WHERE c LIKE '%world%'")
- util.tableEnv.registerTable("TempTable2", table2)
+ util.tableEnv.createTemporaryView("TempTable2", table2)
val sqlQuery1 =
"""
@@ -615,7 +615,7 @@ class DagOptimizationTest extends TableTestBase {
|SELECT * FROM TempTable2
""".stripMargin
val table3 = util.tableEnv.sqlQuery(sqlQuery1)
- util.tableEnv.registerTable("TempTable3", table3)
+ util.tableEnv.createTemporaryView("TempTable3", table3)
val table4 = util.tableEnv.sqlQuery("SELECT * FROM TempTable3 WHERE b >= 5")
val retractSink1 = util.createRetractTableSink(Array("a", "b", "c"), Array(INT, LONG, STRING))
@@ -625,7 +625,7 @@ class DagOptimizationTest extends TableTestBase {
stmtSet.addInsert("retractSink1", table4)
val table5 = util.tableEnv.sqlQuery("SELECT b, count(a) as cnt FROM TempTable3 GROUP BY b")
- util.tableEnv.registerTable("TempTable4", table5)
+ util.tableEnv.createTemporaryView("TempTable4", table5)
val table6 = util.tableEnv.sqlQuery("SELECT b, cnt FROM TempTable4 WHERE b < 4")
val retractSink2 = util.createRetractTableSink(Array("b", "cnt"), Array(LONG, LONG))
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/LegacySinkTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/LegacySinkTest.scala
index 33fff26a2b6..074e6bde45a 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/LegacySinkTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/LegacySinkTest.scala
@@ -141,7 +141,7 @@ class LegacySinkTest extends TableTestBase {
def testRetractAndUpsertSink(): Unit = {
val stmtSet = util.tableEnv.createStatementSet()
val table = util.tableEnv.sqlQuery("SELECT b, COUNT(a) AS cnt FROM MyTable GROUP BY b")
- util.tableEnv.registerTable("TempTable", table)
+ util.tableEnv.createTemporaryView("TempTable", table)
val table1 = util.tableEnv.sqlQuery("SELECT b, cnt FROM TempTable WHERE b < 4")
val retractSink = util.createRetractTableSink(Array("b", "cnt"), Array(LONG, LONG))
@@ -164,7 +164,7 @@ class LegacySinkTest extends TableTestBase {
def testUpsertAndUpsertSink(): Unit = {
val stmtSet = util.tableEnv.createStatementSet()
val table = util.tableEnv.sqlQuery("SELECT b, COUNT(a) AS cnt FROM MyTable GROUP BY b")
- util.tableEnv.registerTable("TempTable", table)
+ util.tableEnv.createTemporaryView("TempTable", table)
val table1 = util.tableEnv.sqlQuery(
"SELECT cnt, COUNT(b) AS frequency FROM TempTable WHERE b < 4 GROUP BY cnt")
@@ -192,7 +192,7 @@ class LegacySinkTest extends TableTestBase {
val table =
util.tableEnv.sqlQuery("SELECT a, b FROM MyTable UNION ALL SELECT d, e FROM MyTable2")
- util.tableEnv.registerTable("TempTable", table)
+ util.tableEnv.createTemporaryView("TempTable", table)
val appendSink = util.createAppendTableSink(Array("a", "b"), Array(INT, LONG))
util.tableEnv
@@ -202,7 +202,7 @@ class LegacySinkTest extends TableTestBase {
val table1 =
util.tableEnv.sqlQuery("SELECT a, b FROM TempTable UNION ALL SELECT i, j FROM MyTable3")
- util.tableEnv.registerTable("TempTable1", table1)
+ util.tableEnv.createTemporaryView("TempTable1", table1)
val table2 = util.tableEnv.sqlQuery("SELECT SUM(a) AS total_sum FROM TempTable1")
val retractSink = util.createRetractTableSink(Array("total_sum"), Array(INT))
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/MatchRecognizeTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/MatchRecognizeTest.scala
index 48c97a256cc..f1e3755ea17 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/MatchRecognizeTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/MatchRecognizeTest.scala
@@ -132,7 +132,7 @@ class MatchRecognizeTest extends TableTestBase {
|) AS T
|""".stripMargin
val table = util.tableEnv.sqlQuery(sqlQuery)
- util.tableEnv.registerTable("T", table)
+ util.tableEnv.createTemporaryView("T", table)
val sqlQuery1 =
s"""
|SELECT *
@@ -162,7 +162,7 @@ class MatchRecognizeTest extends TableTestBase {
|) AS T
|""".stripMargin
val table = util.tableEnv.sqlQuery(sqlQuery)
- util.tableEnv.registerTable("T", table)
+ util.tableEnv.createTemporaryView("T", table)
val sqlQuery1 =
"""
|SELECT
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/table/validation/LegacyTableSinkValidationTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/table/validation/LegacyTableSinkValidationTest.scala
index f7183a47bc7..f92823021f0 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/table/validation/LegacyTableSinkValidationTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/table/validation/LegacyTableSinkValidationTest.scala
@@ -101,7 +101,7 @@ class LegacyTableSinkValidationTest extends TableTestBase {
val tEnv = StreamTableEnvironment.create(env, TableTestUtil.STREAM_SETTING)
val sourceTable = env.fromCollection(TestData.tupleData3).toTable(tEnv, 'a, 'b, 'c)
- tEnv.registerTable("source", sourceTable)
+ tEnv.createTemporaryView("source", sourceTable)
val resultTable = tEnv.sqlQuery("select a, b, c, b as d from source")
val sinkSchema = TableSchema
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/utils/FlinkRelOptUtilTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/utils/FlinkRelOptUtilTest.scala
index a1c9ae53c08..52506a638fc 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/utils/FlinkRelOptUtilTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/utils/FlinkRelOptUtilTest.scala
@@ -58,7 +58,7 @@ class FlinkRelOptUtilTest {
val tableEnv = StreamTableEnvironment.create(env, TableTestUtil.STREAM_SETTING)
val table = env.fromElements[(Int, Long, String)]().toTable(tableEnv, 'a, 'b, 'c)
- tableEnv.registerTable("MyTable", table)
+ tableEnv.createTemporaryView("MyTable", table)
val sqlQuery =
"""
@@ -110,7 +110,7 @@ class FlinkRelOptUtilTest {
val tableEnv = StreamTableEnvironment.create(env, TableTestUtil.STREAM_SETTING)
val table = env.fromElements[(Int, Long, String)]().toTable(tableEnv, 'a, 'b, 'c)
- tableEnv.registerTable("MyTable", table)
+ tableEnv.createTemporaryView("MyTable", table)
val sqlQuery =
"""
@@ -163,7 +163,7 @@ class FlinkRelOptUtilTest {
@Test
def testGetDigestWithDynamicFunctionView(): Unit = {
val view = tableEnv.sqlQuery("SELECT id AS random FROM MyTable ORDER BY rand() LIMIT 1")
- tableEnv.registerTable("MyView", view)
+ tableEnv.createTemporaryView("MyView", view)
val table = tableEnv.sqlQuery("""
|(SELECT * FROM MyView)
|INTERSECT
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/CalcITCase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/CalcITCase.scala
index 57d86e22d41..983cefc80c8 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/CalcITCase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/CalcITCase.scala
@@ -1080,7 +1080,7 @@ class CalcITCase extends BatchTestBase {
((1, 1), "1"),
((2, 2), "2")
))
- tEnv.registerTable("MyTable", table)
+ tEnv.createTemporaryView("MyTable", table)
checkResult(
"SELECT * FROM MyTable",
@@ -1102,7 +1102,7 @@ class CalcITCase extends BatchTestBase {
(2L, "2")
),
"a, b")
- tEnv.registerTable("MyTable", table)
+ tEnv.createTemporaryView("MyTable", table)
checkResult(
"select * from (select MAP[a,b], a from MyTable)",
@@ -1133,7 +1133,7 @@ class CalcITCase extends BatchTestBase {
(2L, "2")
),
"a, b")
- tEnv.registerTable("MyTable", table)
+ tEnv.createTemporaryView("MyTable", table)
checkResult(
"select * from (select ARRAY[a,cast(b as BIGINT)], a from MyTable)",
Seq(
@@ -1159,7 +1159,7 @@ class CalcITCase extends BatchTestBase {
registerFunction("splitUDF1", splitUDF1)
val t1 = BatchTableEnvUtil.fromCollection(tEnv, data, "a, b, c")
- tEnv.registerTable("T1", t1)
+ tEnv.createTemporaryView("T1", t1)
// uses SQL escaping (be aware that even Scala multi-line strings parse backslash!)
checkResult(
s"""
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/table/CalcITCase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/table/CalcITCase.scala
index 83b741867b0..5f221edb2ec 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/table/CalcITCase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/table/CalcITCase.scala
@@ -372,7 +372,7 @@ class CalcITCase extends BatchTestBase {
UserDefinedFunctionTestUtils.setJobParameters(env, Map("string.value" -> "ABC"))
val ds = CollectionBatchExecTable.getSmall3TupleDataSet(tEnv, "a, b, c")
- tEnv.registerTable("t1", ds)
+ tEnv.createTemporaryView("t1", ds)
val sqlQuery = "SELECT c FROM t1 where RichFunc2(c)='ABC#Hello'"
@@ -391,7 +391,7 @@ class CalcITCase extends BatchTestBase {
registerFunction("RichFunc3", new RichFunc3)
val ds = CollectionBatchExecTable.getSmall3TupleDataSet(tEnv, "a, b, c")
- tEnv.registerTable("t1", ds)
+ tEnv.createTemporaryView("t1", ds)
val sqlQuery = "SELECT c FROM t1 where RichFunc3(c)=true"
@@ -409,7 +409,7 @@ class CalcITCase extends BatchTestBase {
UserDefinedFunctionTestUtils.setJobParameters(env, Map("string.value" -> "Abc"))
val ds = CollectionBatchExecTable.getSmall3TupleDataSet(tEnv, "a, b, c")
- tEnv.registerTable("t1", ds)
+ tEnv.createTemporaryView("t1", ds)
val sqlQuery = "SELECT c FROM t1 where " +
"RichFunc2(c)='Abc#Hello' or RichFunc1(a)=3 and b=2"
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/harness/OverAggregateHarnessTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/harness/OverAggregateHarnessTest.scala
index cedfd412bd2..56d07ecf630 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/harness/OverAggregateHarnessTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/harness/OverAggregateHarnessTest.scala
@@ -126,7 +126,7 @@ class OverAggregateHarnessTest(mode: StateBackendMode) extends HarnessTestBase(m
: (KeyedOneInputStreamOperatorTestHarness[RowData, RowData, RowData], Array[LogicalType]) = {
val data = new mutable.MutableList[(Long, String, Long)]
val t = env.fromCollection(data).toTable(tEnv, 'currtime, 'b, 'c, 'proctime.proctime)
- tEnv.registerTable("T", t)
+ tEnv.createTemporaryView("T", t)
val sql =
"""
@@ -159,7 +159,7 @@ class OverAggregateHarnessTest(mode: StateBackendMode) extends HarnessTestBase(m
val data = new mutable.MutableList[(Long, String, Long)]
val t = env.fromCollection(data).toTable(tEnv, 'currtime, 'b, 'c, 'proctime.proctime)
- tEnv.registerTable("T", t)
+ tEnv.createTemporaryView("T", t)
val sql =
"""
@@ -263,7 +263,7 @@ class OverAggregateHarnessTest(mode: StateBackendMode) extends HarnessTestBase(m
val data = new mutable.MutableList[(Long, String, Long)]
val t = env.fromCollection(data).toTable(tEnv, 'currtime, 'b, 'c, 'proctime.proctime)
- tEnv.registerTable("T", t)
+ tEnv.createTemporaryView("T", t)
val sql =
"""
@@ -352,7 +352,7 @@ class OverAggregateHarnessTest(mode: StateBackendMode) extends HarnessTestBase(m
val data = new mutable.MutableList[(Long, String, Long)]
val t = env.fromCollection(data).toTable(tEnv, 'rowtime.rowtime, 'b, 'c)
- tEnv.registerTable("T", t)
+ tEnv.createTemporaryView("T", t)
val sql =
"""
@@ -449,7 +449,7 @@ class OverAggregateHarnessTest(mode: StateBackendMode) extends HarnessTestBase(m
val data = new mutable.MutableList[(Long, String, Long)]
val t = env.fromCollection(data).toTable(tEnv, 'rowtime.rowtime, 'b, 'c)
- tEnv.registerTable("T", t)
+ tEnv.createTemporaryView("T", t)
val sql =
"""
@@ -576,7 +576,7 @@ class OverAggregateHarnessTest(mode: StateBackendMode) extends HarnessTestBase(m
val data = new mutable.MutableList[(Long, String, Long)]
val t = env.fromCollection(data).toTable(tEnv, 'rowtime.rowtime, 'b, 'c)
- tEnv.registerTable("T", t)
+ tEnv.createTemporaryView("T", t)
val sql =
"""
@@ -698,7 +698,7 @@ class OverAggregateHarnessTest(mode: StateBackendMode) extends HarnessTestBase(m
val data = new mutable.MutableList[(Long, String, Long)]
val t = env.fromCollection(data).toTable(tEnv, 'rowtime.rowtime, 'b, 'c)
- tEnv.registerTable("T", t)
+ tEnv.createTemporaryView("T", t)
val sql =
"""
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/AggregateITCase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/AggregateITCase.scala
index 1e6cf31ed7b..81fa9674e58 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/AggregateITCase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/AggregateITCase.scala
@@ -78,7 +78,7 @@ class AggregateITCase(aggMode: AggMode, miniBatch: MiniBatchMode, backend: State
data.+=((3, 3))
val t = failingDataSource(data).toTable(tEnv, 'a, 'b)
- tEnv.registerTable("T", t)
+ tEnv.createTemporaryView("T", t)
val t1 = tEnv.sqlQuery(
"select sum(a), avg(a), min(a), count(a), count(1) from T where a > 9999 group by b")
@@ -97,7 +97,7 @@ class AggregateITCase(aggMode: AggMode, miniBatch: MiniBatchMode, backend: State
data.+=((3, 3))
val t = failingDataSource(data).toTable(tEnv, 'a, 'b)
- tEnv.registerTable("T", t)
+ tEnv.createTemporaryView("T", t)
tEnv.registerFunction("pojoFunc", MyToPojoFunc)
val t1 =
@@ -118,7 +118,7 @@ class AggregateITCase(aggMode: AggMode, miniBatch: MiniBatchMode, backend: State
data.+=((3, 3))
val t = failingDataSource(data).toTable(tEnv, 'a, 'b)
- tEnv.registerTable("T", t)
+ tEnv.createTemporaryView("T", t)
val t1 =
tEnv.sqlQuery("select sum(a), avg(a), min(a), count(a), count(1) from T where a > 9999")
@@ -141,7 +141,7 @@ class AggregateITCase(aggMode: AggMode, miniBatch: MiniBatchMode, backend: State
data.+=((3, 3))
val t = failingDataSource(data).toTable(tEnv, 'a, 'b)
- tEnv.registerTable("T", t)
+ tEnv.createTemporaryView("T", t)
val t1 = tEnv.sqlQuery("select sum(a), avg(a), min(a), count(a), count(1) from T")
val sink = new TestingRetractSink
@@ -163,7 +163,7 @@ class AggregateITCase(aggMode: AggMode, miniBatch: MiniBatchMode, backend: State
"GROUP BY b"
val t = failingDataSource(TestData.tupleData3).toTable(tEnv, 'a, 'b, 'c)
- tEnv.registerTable("MyTable", t)
+ tEnv.createTemporaryView("MyTable", t)
val result = tEnv.sqlQuery(sqlQuery).toRetractStream[Row]
val sink = new TestingRetractSink
@@ -322,7 +322,7 @@ class AggregateITCase(aggMode: AggMode, miniBatch: MiniBatchMode, backend: State
// 5, 2
val t = failingDataSource(data).toTable(tEnv, 'a, 'b, 'c)
- tEnv.registerTable("T", t)
+ tEnv.createTemporaryView("T", t)
val sql =
"""
@@ -358,7 +358,7 @@ class AggregateITCase(aggMode: AggMode, miniBatch: MiniBatchMode, backend: State
}
}
val t = failingDataSource(Random.shuffle(data)).toTable(tEnv, 'a, 'b)
- tEnv.registerTable("T", t)
+ tEnv.createTemporaryView("T", t)
val distincts = for (i <- 0 until 100) yield {
s"count(distinct a) filter (where b = $i)"
@@ -402,7 +402,7 @@ class AggregateITCase(aggMode: AggMode, miniBatch: MiniBatchMode, backend: State
data.+=((17, 5L, "B"))
val t = failingDataSource(data).toTable(tEnv, 'a, 'b, 'c)
- tEnv.registerTable("T", t)
+ tEnv.createTemporaryView("T", t)
tEnv.createTemporarySystemFunction("CntNullNonNull", new CountNullNonNull)
val t1 = tEnv.sqlQuery("SELECT b, count(*), CntNullNonNull(DISTINCT c) FROM T GROUP BY b")
@@ -435,7 +435,7 @@ class AggregateITCase(aggMode: AggMode, miniBatch: MiniBatchMode, backend: State
env.setParallelism(1)
t = failingDataSource(data).toTable(tEnv, 'a)
- tEnv.registerTable("T", t)
+ tEnv.createTemporaryView("T", t)
t = tEnv.sqlQuery("select sum(cast(a as decimal(32, 8))) from T")
sink = new TestingRetractSink
@@ -549,7 +549,7 @@ class AggregateITCase(aggMode: AggMode, miniBatch: MiniBatchMode, backend: State
env.setParallelism(1)
t = failingDataSource(data).toTable(tEnv, 'a)
- tEnv.registerTable("T", t)
+ tEnv.createTemporaryView("T", t)
t = tEnv.sqlQuery("select avg(cast(a as decimal(32, 8))) from T")
sink = new TestingRetractSink
@@ -579,7 +579,7 @@ class AggregateITCase(aggMode: AggMode, miniBatch: MiniBatchMode, backend: State
data.+=((12, 5L, "B"))
val t = failingDataSource(data).toTable(tEnv, 'a, 'b, 'c)
- tEnv.registerTable("T", t)
+ tEnv.createTemporaryView("T", t)
val t1 = tEnv.sqlQuery("SELECT b, count(c), sum(a) FROM T GROUP BY b")
val sink = new TestingRetractSink
@@ -606,7 +606,7 @@ class AggregateITCase(aggMode: AggMode, miniBatch: MiniBatchMode, backend: State
data.+=((12, 5L, "B"))
val t = failingDataSource(data).toTable(tEnv, 'a, 'b, 'c)
- tEnv.registerTable("T", t)
+ tEnv.createTemporaryView("T", t)
val sql =
s"""
|select
@@ -644,7 +644,7 @@ class AggregateITCase(aggMode: AggMode, miniBatch: MiniBatchMode, backend: State
data.+=((12, 5L, "B"))
val t = failingDataSource(data).toTable(tEnv, 'a, 'b, 'c)
- tEnv.registerTable("T", t)
+ tEnv.createTemporaryView("T", t)
val sql =
"""
@@ -669,7 +669,7 @@ class AggregateITCase(aggMode: AggMode, miniBatch: MiniBatchMode, backend: State
@Test
def testUnboundedGroupBy(): Unit = {
val t = failingDataSource(TestData.tupleData3).toTable(tEnv, 'a, 'b, 'c)
- tEnv.registerTable("MyTable", t)
+ tEnv.createTemporaryView("MyTable", t)
val sqlQuery = "SELECT b, COUNT(a) FROM MyTable GROUP BY b"
@@ -687,7 +687,7 @@ class AggregateITCase(aggMode: AggMode, miniBatch: MiniBatchMode, backend: State
.assignTimestampsAndWatermarks(
new TimestampAndWatermarkWithOffset[(Long, Int, Int, String, Long)](0L))
.toTable(tEnv, 'rowtime.rowtime, 'a, 'c, 'd, 'e)
- tEnv.registerTable("MyTable", t)
+ tEnv.createTemporaryView("MyTable", t)
val innerSql =
"""
@@ -717,7 +717,7 @@ class AggregateITCase(aggMode: AggMode, miniBatch: MiniBatchMode, backend: State
val t: DataStream[(String, String)] = failingDataSource(dataWithNull)
val streamTable = t.toTable(tEnv, 'x, 'y)
- tEnv.registerTable("T", streamTable)
+ tEnv.createTemporaryView("T", streamTable)
tEnv.executeSql("""
|CREATE VIEW view1 AS
@@ -759,7 +759,7 @@ class AggregateITCase(aggMode: AggMode, miniBatch: MiniBatchMode, backend: State
val t: DataStream[(Int, Int, String)] = failingDataSource(dataWithNull)
val streamTable = t.toTable(tEnv, 'id, 'len, 'content)
- tEnv.registerTable("T", streamTable)
+ tEnv.createTemporaryView("T", streamTable)
val sqlQuery =
s"""
@@ -780,7 +780,7 @@ class AggregateITCase(aggMode: AggMode, miniBatch: MiniBatchMode, backend: State
val t: DataStream[(Int, Int, String)] = failingDataSource(dataWithNull)
val streamTable = t.toTable(tEnv, 'id, 'len, 'content)
- tEnv.registerTable("T", streamTable)
+ tEnv.createTemporaryView("T", streamTable)
val sqlQuery =
s"""
@@ -809,7 +809,7 @@ class AggregateITCase(aggMode: AggMode, miniBatch: MiniBatchMode, backend: State
data.+=((8, 4L, "EF"))
data.+=((8, 4L, null))
val sqlQuery = "SELECT b, LISTAGG(DISTINCT c, '#') FROM MyTable GROUP BY b"
- tEnv.registerTable("MyTable", failingDataSource(data).toTable(tEnv).as("a", "b", "c"))
+ tEnv.createTemporaryView("MyTable", failingDataSource(data).toTable(tEnv).as("a", "b", "c"))
val sink = new TestingRetractSink
tEnv.sqlQuery(sqlQuery).toRetractStream[Row].addSink(sink).setParallelism(1)
env.execute()
@@ -822,7 +822,7 @@ class AggregateITCase(aggMode: AggMode, miniBatch: MiniBatchMode, backend: State
val sqlQuery = "SELECT b, COLLECT(a) FROM MyTable GROUP BY b"
val t = failingDataSource(TestData.tupleData3).toTable(tEnv, 'a, 'b, 'c)
- tEnv.registerTable("MyTable", t)
+ tEnv.createTemporaryView("MyTable", t)
val sink = new TestingRetractSink
tEnv.sqlQuery(sqlQuery).toRetractStream[Row].addSink(sink)
@@ -852,7 +852,7 @@ class AggregateITCase(aggMode: AggMode, miniBatch: MiniBatchMode, backend: State
(5, 3, List(18, "42.6"))
)
- tEnv.registerTable("MyTable", failingDataSource(data).toTable(tEnv, 'a, 'b, 'c))
+ tEnv.createTemporaryView("MyTable", failingDataSource(data).toTable(tEnv, 'a, 'b, 'c))
val sink = new TestingRetractSink
tEnv.sqlQuery(sqlQuery).toRetractStream[Row].addSink(sink)
@@ -889,8 +889,8 @@ class AggregateITCase(aggMode: AggMode, miniBatch: MiniBatchMode, backend: State
data.+=((12, 5L, "B"))
val t = failingDataSource(data).toTable(tEnv, 'a, 'b, 'c)
- tEnv.registerTable("T1", t)
- tEnv.registerTable("T2", t)
+ tEnv.createTemporaryView("T1", t)
+ tEnv.createTemporaryView("T2", t)
val t1 = tEnv.sqlQuery("SELECT * FROM T2 WHERE T2.a < (SELECT count(*) * 0.3 FROM T1)")
val sink = new TestingRetractSink
@@ -902,7 +902,7 @@ class AggregateITCase(aggMode: AggMode, miniBatch: MiniBatchMode, backend: State
// test single value for char type
val tc = tEnv.fromValues(DataTypes.ROW(DataTypes.FIELD("a", DataTypes.CHAR(3))), Row.of("AA"))
- tEnv.registerTable("tc", tc)
+ tEnv.createTemporaryView("tc", tc)
val tr = tEnv.sqlQuery("SELECT * FROM tc WHERE tc.a = (SELECT a FROM tc)")
val sink1 = new TestingRetractSink
tr.toRetractStream[Row].addSink(sink1).setParallelism(1)
@@ -915,7 +915,7 @@ class AggregateITCase(aggMode: AggMode, miniBatch: MiniBatchMode, backend: State
val data = Seq((1, new MyPojo(5, 105)), (1, new MyPojo(6, 11)), (1, new MyPojo(7, 12)))
val t = failingDataSource(data).toTable(tEnv, 'a, 'b)
- tEnv.registerTable("MyTable", t)
+ tEnv.createTemporaryView("MyTable", t)
tEnv.registerFunction("pojoFunc", new MyPojoAggFunction)
tEnv.registerFunction("pojoToInt", MyPojoFunc)
@@ -939,7 +939,7 @@ class AggregateITCase(aggMode: AggMode, miniBatch: MiniBatchMode, backend: State
val rowType = new RowTypeInfo(BigDecimalTypeInfo.of(7, 2))
val t = failingDataSource(data)(rowType).toTable(tEnv, 'd)
- tEnv.registerTable("T", t)
+ tEnv.createTemporaryView("T", t)
val sql =
"""
@@ -1085,9 +1085,9 @@ class AggregateITCase(aggMode: AggMode, miniBatch: MiniBatchMode, backend: State
)
val t1 = failingDataSource(data).toTable(tEnv, 'a, 'b, 'c)
- tEnv.registerTable("T1", t1)
+ tEnv.createTemporaryView("T1", t1)
val t2 = failingDataSource(data).toTable(tEnv, 'a, 'b, 'c)
- tEnv.registerTable("T2", t2)
+ tEnv.createTemporaryView("T2", t2)
val sql =
"""
@@ -1112,7 +1112,7 @@ class AggregateITCase(aggMode: AggMode, miniBatch: MiniBatchMode, backend: State
val data = List((1, 1L, "5", "3"), (1, 22L, "15", "13"), (3, 33L, "25", "23"))
val t = failingDataSource(data).toTable(tEnv, 'id, 's, 's1, 's2)
- tEnv.registerTable("MyTable", t)
+ tEnv.createTemporaryView("MyTable", t)
tEnv.createTemporarySystemFunction("func", classOf[VarArgsAggFunction])
val sql = "SELECT func(s, s1, s2) FROM MyTable"
@@ -1129,7 +1129,7 @@ class AggregateITCase(aggMode: AggMode, miniBatch: MiniBatchMode, backend: State
val data = List((1, 1L, "5", "3"), (1, 22L, "15", "13"), (3, 33L, "25", "23"))
val t = failingDataSource(data).toTable(tEnv, 'id, 's, 's1, 's2)
- tEnv.registerTable("MyTable", t)
+ tEnv.createTemporaryView("MyTable", t)
tEnv.createTemporarySystemFunction("func", classOf[VarArgsAggFunction])
val sink = new TestingRetractSink
@@ -1157,7 +1157,7 @@ class AggregateITCase(aggMode: AggMode, miniBatch: MiniBatchMode, backend: State
data.+=((10, 4L, "IJ"))
val t = failingDataSource(data).toTable(tEnv, 'a, 'b, 'c)
- tEnv.registerTable("T", t)
+ tEnv.createTemporaryView("T", t)
val sql =
"""
@@ -1185,7 +1185,7 @@ class AggregateITCase(aggMode: AggMode, miniBatch: MiniBatchMode, backend: State
}
val t = failingDataSource(data).toTable(tEnv, 'a, 'b, 'c)
- tEnv.registerTable("T", t)
+ tEnv.createTemporaryView("T", t)
val sql =
"""
@@ -1225,7 +1225,7 @@ class AggregateITCase(aggMode: AggMode, miniBatch: MiniBatchMode, backend: State
data.+=((2, 3L, "D", true))
val t = failingDataSource(data).toTable(tEnv, 'a, 'b, 'c, 'd)
- tEnv.registerTable("T", t)
+ tEnv.createTemporaryView("T", t)
// test declarative and imperative aggregates
val sql =
"""
@@ -1265,7 +1265,7 @@ class AggregateITCase(aggMode: AggMode, miniBatch: MiniBatchMode, backend: State
val rowType = new RowTypeInfo(BigDecimalTypeInfo.of(7, 2))
val t = failingDataSource(data)(rowType).toTable(tEnv, 'a)
- tEnv.registerTable("T", t)
+ tEnv.createTemporaryView("T", t)
val sql =
"""
@@ -1291,11 +1291,11 @@ class AggregateITCase(aggMode: AggMode, miniBatch: MiniBatchMode, backend: State
(4, 3, (14, "45.2136")),
(5, 3, (18, "42.6"))
)
- tEnv.registerTable("src", env.fromCollection(data).toTable(tEnv, 'a, 'b, 'c))
+ tEnv.createTemporaryView("src", env.fromCollection(data).toTable(tEnv, 'a, 'b, 'c))
val sql = "SELECT a, b, COLLECT(c) as `set` FROM src GROUP BY a, b"
val view1 = tEnv.sqlQuery(sql)
- tEnv.registerTable("v1", view1)
+ tEnv.createTemporaryView("v1", view1)
tEnv.createTemporarySystemFunction("toCompObj", ToCompositeObj)
tEnv.createTemporarySystemFunction("anyToString", AnyToStringFunction)
@@ -1348,7 +1348,7 @@ class AggregateITCase(aggMode: AggMode, miniBatch: MiniBatchMode, backend: State
}
val t = failingDataSource(data).toTable(tEnv, 'a, 'b, 'c)
- tEnv.registerTable("MyTable", t)
+ tEnv.createTemporaryView("MyTable", t)
val sink = new TestingRetractSink
tEnv.sqlQuery(sqlQuery).toRetractStream[Row].addSink(sink)
@@ -1369,7 +1369,7 @@ class AggregateITCase(aggMode: AggMode, miniBatch: MiniBatchMode, backend: State
}
val t = failingDataSource(data).toTable(tEnv, 'a, 'b, 'c)
- tEnv.registerTable("MyTable", t)
+ tEnv.createTemporaryView("MyTable", t)
val sink = new TestingRetractSink
tEnv.sqlQuery(sqlQuery).toRetractStream[Row].addSink(sink)
@@ -1391,7 +1391,7 @@ class AggregateITCase(aggMode: AggMode, miniBatch: MiniBatchMode, backend: State
data.+=((11000, 1L, "Hello"))
val t = failingDataSource(data).toTable(tEnv, 'a, 'b, 'c)
- tEnv.registerTable("MyTable", t)
+ tEnv.createTemporaryView("MyTable", t)
val sink = new TestingRetractSink
tEnv.sqlQuery(sqlQuery).toRetractStream[Row].addSink(sink)
@@ -1409,7 +1409,7 @@ class AggregateITCase(aggMode: AggMode, miniBatch: MiniBatchMode, backend: State
val data = Seq[(Int, Int)]((1, 1), (2, 2))
val t = failingDataSource(data).toTable(tEnv, 'a, 'b)
- tEnv.registerTable("MyTable", t)
+ tEnv.createTemporaryView("MyTable", t)
val sink = new TestingRetractSink
tEnv.sqlQuery(sqlQuery).toRetractStream[Row].addSink(sink)
@@ -1425,7 +1425,7 @@ class AggregateITCase(aggMode: AggMode, miniBatch: MiniBatchMode, backend: State
val data = (0 until 100).map(i => ("1", "1", s"${i % 50}", "1")).toList
// use BinaryRowData source here for StringData reuse
val t = failingBinaryRowSource(data).toTable(tEnv, 'a, 'b, 'c, 'd)
- tEnv.registerTable("src", t)
+ tEnv.createTemporaryView("src", t)
val sql =
s"""
@@ -1460,7 +1460,7 @@ class AggregateITCase(aggMode: AggMode, miniBatch: MiniBatchMode, backend: State
@Test
def testDistinctWithMultiFilter(): Unit = {
val t = failingDataSource(TestData.tupleData3).toTable(tEnv).as("a", "b", "c")
- tEnv.registerTable("MyTable", t)
+ tEnv.createTemporaryView("MyTable", t)
val sqlQuery =
s"""
@@ -1497,7 +1497,7 @@ class AggregateITCase(aggMode: AggMode, miniBatch: MiniBatchMode, backend: State
data.+=((3, 2L, "Hello world"))
val t = failingDataSource(data).toTable(tEnv, 'a, 'b, 'c)
- tEnv.registerTable("T", t)
+ tEnv.createTemporaryView("T", t)
val t1 = tEnv.sqlQuery(
"select a from (select b, max(a) as a, count(*), max(c) as c from T group by b) T1")
@@ -1533,7 +1533,7 @@ class AggregateITCase(aggMode: AggMode, miniBatch: MiniBatchMode, backend: State
data.+=((5, 3L, "C"))
val t = failingDataSource(data).toTable(tEnv, 'a, 'b, 'c)
- tEnv.registerTable("MyTable", t)
+ tEnv.createTemporaryView("MyTable", t)
val tableSink = new TestingUpsertTableSink(Array(0))
.configure(Array[String]("c", "bMax"), Array[TypeInformation[_]](Types.STRING, Types.LONG))
@@ -1628,7 +1628,7 @@ class AggregateITCase(aggMode: AggMode, miniBatch: MiniBatchMode, backend: State
)
val tableA = failingDataSource(empsData)
.toTable(tEnv, 'empno, 'name, 'deptno, 'gender, 'city, 'empid, 'age, 'slacker, 'manager)
- tEnv.registerTable("emps", tableA)
+ tEnv.createTemporaryView("emps", tableA)
val sql =
s"""
|select
@@ -1664,7 +1664,7 @@ class AggregateITCase(aggMode: AggMode, miniBatch: MiniBatchMode, backend: State
)
val tableA = failingDataSource(empsData)
.toTable(tEnv, 'empno, 'name, 'deptno, 'gender, 'city, 'empid, 'age, 'slacker, 'manager)
- tEnv.registerTable("emps", tableA)
+ tEnv.createTemporaryView("emps", tableA)
val sql =
s"""
|select
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/AggregateRemoveITCase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/AggregateRemoveITCase.scala
index c243cc93de9..519943d52bc 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/AggregateRemoveITCase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/AggregateRemoveITCase.scala
@@ -276,7 +276,7 @@ class AggregateRemoveITCase(aggMode: AggMode, minibatch: MiniBatchMode, backend:
data.+=((6, 2))
val t = failingDataSource(data).toTable(tEnv, 'a, 'b)
- tEnv.registerTable("T1", t)
+ tEnv.createTemporaryView("T1", t)
val t1 = tEnv.sqlQuery("""
|select sum(b) from
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/AsyncLookupJoinITCase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/AsyncLookupJoinITCase.scala
index c2e8e2e606b..acf77b46d7d 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/AsyncLookupJoinITCase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/AsyncLookupJoinITCase.scala
@@ -300,7 +300,7 @@ class AsyncLookupJoinITCase(
val sql1 = "SELECT max(id) as id, PROCTIME() as proctime FROM src AS T group by len"
val table1 = tEnv.sqlQuery(sql1)
- tEnv.registerTable("t1", table1)
+ tEnv.createTemporaryView("t1", table1)
val sql2 = "SELECT t1.id, D.name, D.age FROM t1 LEFT JOIN user_table " +
"for system_time as of t1.proctime AS D ON t1.id = D.id"
@@ -326,7 +326,7 @@ class AsyncLookupJoinITCase(
val sql1 = "SELECT max(id) as id, PROCTIME() as proctime FROM src AS T group by len"
val table1 = tEnv.sqlQuery(sql1)
- tEnv.registerTable("t1", table1)
+ tEnv.createTemporaryView("t1", table1)
val sql2 = "SELECT t1.id, D.name, D.age FROM t1 LEFT JOIN user_table " +
"for system_time as of t1.proctime AS D ON t1.id = D.id"
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/CalcITCase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/CalcITCase.scala
index 886a862d867..bd4cdf3e9c3 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/CalcITCase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/CalcITCase.scala
@@ -95,7 +95,7 @@ class CalcITCase extends StreamingTestBase {
val ds = env.fromCollection(data)
val t = ds.toTable(tEnv, 'a, 'b)
- tEnv.registerTable("MyTableRow", t)
+ tEnv.createTemporaryView("MyTableRow", t)
val outputType = InternalTypeInfo.ofFields(new IntType(), new BooleanType())
@@ -130,7 +130,7 @@ class CalcITCase extends StreamingTestBase {
val ds = env.fromCollection(data)
val t = ds.toTable(tEnv, 'a, 'b, 'c)
- tEnv.registerTable("MyTableRow", t)
+ tEnv.createTemporaryView("MyTableRow", t)
val outputType = InternalTypeInfo.ofFields(new IntType(), new IntType(), new BigIntType())
@@ -157,7 +157,7 @@ class CalcITCase extends StreamingTestBase {
val ds = env.fromCollection(data)
val t = ds.toTable(tEnv, 'a, 'b, 'c)
- tEnv.registerTable("MyTableRow", t)
+ tEnv.createTemporaryView("MyTableRow", t)
val outputType =
InternalTypeInfo.ofFields(VarCharType.STRING_TYPE, VarCharType.STRING_TYPE, new IntType())
@@ -190,7 +190,7 @@ class CalcITCase extends StreamingTestBase {
val ds = env.fromCollection(data)
val t = ds.toTable(tEnv, 'a, 'b, 'c)
- tEnv.registerTable("MyTableRow", t)
+ tEnv.createTemporaryView("MyTableRow", t)
val result = tEnv.sqlQuery(sqlQuery).toAppendStream[Row]
val sink = new TestingAppendSink
@@ -215,7 +215,7 @@ class CalcITCase extends StreamingTestBase {
val ds = env.fromCollection(data)
val t = ds.toTable(tEnv, 'a, 'b, 'c)
- tEnv.registerTable("MyTableRow", t)
+ tEnv.createTemporaryView("MyTableRow", t)
val result = tEnv.sqlQuery(sqlQuery).toAppendStream[Row]
val sink = new TestingAppendSink
@@ -233,7 +233,7 @@ class CalcITCase extends StreamingTestBase {
val t = env
.fromCollection(TestData.smallTupleData3)
.toTable(tEnv, 'a, 'b, 'c)
- tEnv.registerTable("MyTableRow", t)
+ tEnv.createTemporaryView("MyTableRow", t)
val result = tEnv.sqlQuery(sqlQuery).toAppendStream[Row]
val sink = new TestingAppendSink
@@ -251,7 +251,7 @@ class CalcITCase extends StreamingTestBase {
val t = env
.fromCollection(TestData.smallTupleData3)
.toTable(tEnv, 'a, 'b, 'c)
- tEnv.registerTable("MyTableRow", t)
+ tEnv.createTemporaryView("MyTableRow", t)
val result = tEnv.sqlQuery(sqlQuery).toAppendStream[Row]
val sink = new TestingAppendSink
@@ -275,7 +275,7 @@ class CalcITCase extends StreamingTestBase {
)),
'_1,
'_2)
- tEnv.registerTable("MyTable", table)
+ tEnv.createTemporaryView("MyTable", table)
val result = tEnv.sqlQuery(sqlQuery)
val sink = TestSinkUtil.configureSink(result, new TestingAppendTableSink())
@@ -293,7 +293,7 @@ class CalcITCase extends StreamingTestBase {
val t = env
.fromCollection(TestData.tupleData3)
.toTable(tEnv, 'a, 'b, 'c)
- tEnv.registerTable("MyTable", t)
+ tEnv.createTemporaryView("MyTable", t)
val result = tEnv.sqlQuery(sqlQuery).toAppendStream[Row]
val sink = new TestingAppendSink
@@ -331,7 +331,7 @@ class CalcITCase extends StreamingTestBase {
val t = env
.fromCollection(TestData.tupleData3)
.toTable(tEnv, 'a, 'b, 'c)
- tEnv.registerTable("MyTable", t)
+ tEnv.createTemporaryView("MyTable", t)
val result = tEnv.sqlQuery(sqlQuery).toAppendStream[Row]
val sink = new TestingAppendSink
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/CorrelateITCase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/CorrelateITCase.scala
index 4ce1cfc598d..cb5501c04ff 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/CorrelateITCase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/CorrelateITCase.scala
@@ -51,16 +51,16 @@ class CorrelateITCase extends StreamingTestBase {
val data2 = List((1, "abc-bcd"), (1, "hhh"), (1, "xxx"))
val t1 = env.fromCollection(data).toTable(tEnv, 'a, 'b, 'c)
- tEnv.registerTable("T1", t1)
+ tEnv.createTemporaryView("T1", t1)
val t2 = env.fromCollection(data2).toTable(tEnv, 'a, 'c)
- tEnv.registerTable("T2", t2)
+ tEnv.createTemporaryView("T2", t2)
val query1 = "SELECT a, v FROM T1, lateral table(STRING_SPLIT(c, '-')) as T(v)"
- tEnv.registerTable("TMP1", tEnv.sqlQuery(query1))
+ tEnv.createTemporaryView("TMP1", tEnv.sqlQuery(query1))
val query2 = "SELECT a, v FROM T2, lateral table(STRING_SPLIT(c, '-')) as T(v)"
- tEnv.registerTable("TMP2", tEnv.sqlQuery(query2))
+ tEnv.createTemporaryView("TMP2", tEnv.sqlQuery(query2))
val sql =
"""
@@ -106,7 +106,7 @@ class CorrelateITCase extends StreamingTestBase {
val data = List((1, 2, "abc-bcd"), (1, 2, "hhh"), (1, 2, "xxx"))
val t1 = env.fromCollection(data).toTable(tEnv, 'a, 'b, 'c)
- tEnv.registerTable("T1", t1)
+ tEnv.createTemporaryView("T1", t1)
tEnv.registerFunction("str_split", new StringSplit())
val query = "SELECT * FROM T1, LATERAL TABLE(str_split('Jack,John', ',')) as T0(d)"
@@ -142,7 +142,7 @@ class CorrelateITCase extends StreamingTestBase {
val data = List((1, 2, "abc-bcd"), (1, 2, "hhh"), (1, 2, "xxx"))
val t1 = env.fromCollection(data).toTable(tEnv, 'a, 'b, 'c)
- tEnv.registerTable("T1", t1)
+ tEnv.createTemporaryView("T1", t1)
tEnv.registerFunction("str_split", new NonDeterministicTableFunc())
val query = "SELECT * FROM T1, LATERAL TABLE(str_split('Jack#John')) as T0(d)"
@@ -159,7 +159,7 @@ class CorrelateITCase extends StreamingTestBase {
val data = List((1, 2, "abc-bcd"), (1, 2, "hhh"), (1, 2, "xxx"))
val t1 = env.fromCollection(data).toTable(tEnv, 'a, 'b, 'c)
- tEnv.registerTable("T1", t1)
+ tEnv.createTemporaryView("T1", t1)
// UdfWithOpen checks open method is opened, and add a '$' prefix to the given string
tEnv.registerFunction("func", new UdfWithOpen)
@@ -186,7 +186,7 @@ class CorrelateITCase extends StreamingTestBase {
data.+=(("2", "2,L", "B,C"))
val t = env.fromCollection(data).toTable(tEnv, 'a, 'b, 'c)
- tEnv.registerTable("T1", t)
+ tEnv.createTemporaryView("T1", t)
tEnv.registerFunction("str_split", new StringSplit())
val sink1 = new TestingAppendSink
val sink2 = new TestingAppendSink
@@ -225,7 +225,7 @@ class CorrelateITCase extends StreamingTestBase {
val sink = new TestingAppendSink
- tEnv.registerTable("MyTable", in)
+ tEnv.createTemporaryView("MyTable", in)
tEnv.registerFunction("rfFunc", new RF)
tEnv.registerFunction("tfFunc", new TableFunc7)
tEnv
@@ -243,10 +243,10 @@ class CorrelateITCase extends StreamingTestBase {
val data = List((1, 2, "3018-06-10|2018-06-03"), (1, 2, "2018-06-01"), (1, 2, "2018-06-02"))
val t1 = env.fromCollection(data).toTable(tEnv, 'a, 'b, 'c)
- tEnv.registerTable("T1", t1)
+ tEnv.createTemporaryView("T1", t1)
val query1 = "SELECT a, v FROM T1, lateral table(STRING_SPLIT(c, '|')) as T(v)"
- tEnv.registerTable("TMP1", tEnv.sqlQuery(query1))
+ tEnv.createTemporaryView("TMP1", tEnv.sqlQuery(query1))
val sql =
"""
@@ -267,7 +267,7 @@ class CorrelateITCase extends StreamingTestBase {
val data = List((1, 2, ""), (1, 3, ""))
val t1 = env.fromCollection(data).toTable(tEnv, 'a, 'b, 'c)
- tEnv.registerTable("T1", t1)
+ tEnv.createTemporaryView("T1", t1)
val sql = "SELECT * FROM T1 left join lateral table(STRING_SPLIT(c, '|')) as T(v) on true"
@@ -285,7 +285,7 @@ class CorrelateITCase extends StreamingTestBase {
val data = List((1, 2, "3018-06-10|2018-06-03"), (1, 2, "2018-06-01"), (1, 2, "2018-06-02"))
val t1 = env.fromCollection(data).toTable(tEnv, 'a, 'b, 'c)
- tEnv.registerTable("T1", t1)
+ tEnv.createTemporaryView("T1", t1)
val sql = "SELECT v FROM T1, lateral table(STRING_SPLIT(c, '|')) as T(v)"
@@ -303,7 +303,7 @@ class CorrelateITCase extends StreamingTestBase {
val data = List((1, 2, "3018-06-10|2018-06-03"), (1, 2, "2018-06-01"), (1, 2, "2018-06-02"))
val t1 = env.fromCollection(data).toTable(tEnv, 'a, 'b, 'c)
- tEnv.registerTable("T1", t1)
+ tEnv.createTemporaryView("T1", t1)
val sql = "SELECT a, v FROM T1, lateral table(STRING_SPLIT(c, '|')) as T(v)"
@@ -321,7 +321,7 @@ class CorrelateITCase extends StreamingTestBase {
val data = List((1, 2, "a"), (1, 3, ""))
val t1 = env.fromCollection(data).toTable(tEnv, 'a, 'b, 'c)
- tEnv.registerTable("T1", t1)
+ tEnv.createTemporaryView("T1", t1)
val sql = "SELECT v FROM T1, lateral table(STRING_SPLIT(c, '|')) as T(v)"
@@ -339,7 +339,7 @@ class CorrelateITCase extends StreamingTestBase {
val data = List((1, 2, ""), (1, 3, ""))
val t1 = env.fromCollection(data).toTable(tEnv, 'a, 'b, 'c)
- tEnv.registerTable("T1", t1)
+ tEnv.createTemporaryView("T1", t1)
val sql = "SELECT v FROM T1 left join lateral table(STRING_SPLIT(c, '|')) as T(v) on true"
@@ -358,7 +358,7 @@ class CorrelateITCase extends StreamingTestBase {
val data = List((1, 2, "a"), (1, 3, ""))
val t1 = env.fromCollection(data).toTable(tEnv, 'a, 'b, 'c)
- tEnv.registerTable("T1", t1)
+ tEnv.createTemporaryView("T1", t1)
val sql = "SELECT a, v FROM T1, lateral table(STRING_SPLIT(c, '|')) as T(v)"
@@ -376,7 +376,7 @@ class CorrelateITCase extends StreamingTestBase {
val data = List((1, 2, ""), (1, 3, ""))
val t1 = env.fromCollection(data).toTable(tEnv, 'a, 'b, 'c)
- tEnv.registerTable("T1", t1)
+ tEnv.createTemporaryView("T1", t1)
val sql = "SELECT b, v FROM T1 left join lateral table(STRING_SPLIT(c, '|')) as T(v) on true"
@@ -401,7 +401,7 @@ class CorrelateITCase extends StreamingTestBase {
// val tEnv = TableEnvironment.getTableEnvironment(env)
//
// val t1 = env.fromCollection(data).toTable(tEnv, 'a, 'b, 'c)
-// tEnv.registerTable("T1", t1)
+// tEnv.createTemporaryView("T1", t1)
//
// val sql = "SELECT count(*) FROM T1, lateral table(STRING_SPLIT(c, '|')) as T(v)"
//
@@ -423,7 +423,7 @@ class CorrelateITCase extends StreamingTestBase {
// val tEnv = TableEnvironment.getTableEnvironment(env)
//
// val t1 = env.fromCollection(data).toTable(tEnv, 'a, 'b, 'c)
-// tEnv.registerTable("T1", t1)
+// tEnv.createTemporaryView("T1", t1)
//
// val sql =
// "SELECT count(*) FROM T1 left join lateral table(STRING_SPLIT(c, '|')) as T(v) on " +
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/DeduplicateITCase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/DeduplicateITCase.scala
index a94a701bd88..a0f8ab4a477 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/DeduplicateITCase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/DeduplicateITCase.scala
@@ -58,7 +58,7 @@ class DeduplicateITCase(miniBatch: MiniBatchMode, mode: StateBackendMode)
def testFirstRowOnProctime(): Unit = {
val t = failingDataSource(TestData.tupleData3)
.toTable(tEnv, 'a, 'b, 'c, 'proctime.proctime)
- tEnv.registerTable("T", t)
+ tEnv.createTemporaryView("T", t)
val sql =
"""
@@ -88,7 +88,7 @@ class DeduplicateITCase(miniBatch: MiniBatchMode, mode: StateBackendMode)
@Test
def testFirstRowOnBuiltinProctime(): Unit = {
val t = failingDataSource(TestData.tupleData3).toTable(tEnv, 'a, 'b, 'c)
- tEnv.registerTable("T", t)
+ tEnv.createTemporaryView("T", t)
val sql =
"""
@@ -119,7 +119,7 @@ class DeduplicateITCase(miniBatch: MiniBatchMode, mode: StateBackendMode)
def testLastRowOnProctime(): Unit = {
val t = failingDataSource(TestData.tupleData3)
.toTable(tEnv, 'a, 'b, 'c, 'proctime.proctime)
- tEnv.registerTable("T", t)
+ tEnv.createTemporaryView("T", t)
val sql =
"""
@@ -149,7 +149,7 @@ class DeduplicateITCase(miniBatch: MiniBatchMode, mode: StateBackendMode)
@Test
def testLastRowOnBuiltinProctime(): Unit = {
val t = failingDataSource(TestData.tupleData3).toTable(tEnv, 'a, 'b, 'c)
- tEnv.registerTable("T", t)
+ tEnv.createTemporaryView("T", t)
val sql =
"""
@@ -182,7 +182,7 @@ class DeduplicateITCase(miniBatch: MiniBatchMode, mode: StateBackendMode)
.fromCollection(rowtimeTestData)
.assignTimestampsAndWatermarks(new RowtimeExtractor)
.toTable(tEnv, 'a, 'b, 'c, 'rowtime.rowtime())
- tEnv.registerTable("T", t)
+ tEnv.createTemporaryView("T", t)
createSinkTable("rowtime_sink")
val sql =
@@ -221,7 +221,7 @@ class DeduplicateITCase(miniBatch: MiniBatchMode, mode: StateBackendMode)
.fromCollection(rowtimeTestData)
.assignTimestampsAndWatermarks(new RowtimeExtractor)
.toTable(tEnv, 'a, 'b, 'c, 'rowtime.rowtime())
- tEnv.registerTable("T", t)
+ tEnv.createTemporaryView("T", t)
createSinkTable("rowtime_sink")
val sql =
@@ -254,7 +254,7 @@ class DeduplicateITCase(miniBatch: MiniBatchMode, mode: StateBackendMode)
.fromCollection(rowtimeTestData)
.assignTimestampsAndWatermarks(new RowtimeExtractor)
.toTable(tEnv, 'a, 'b, 'c, 'rowtime.rowtime())
- tEnv.registerTable("T", t)
+ tEnv.createTemporaryView("T", t)
tEnv.executeSql(s"""
|CREATE TABLE rowtime_sink (
| cnt BIGINT
@@ -290,7 +290,7 @@ class DeduplicateITCase(miniBatch: MiniBatchMode, mode: StateBackendMode)
.fromCollection(rowtimeTestData)
.assignTimestampsAndWatermarks(new RowtimeExtractor)
.toTable(tEnv, 'a, 'b, 'c, 'rowtime.rowtime())
- tEnv.registerTable("T", t)
+ tEnv.createTemporaryView("T", t)
createSinkTable("rowtime_sink")
val sql =
@@ -333,7 +333,7 @@ class DeduplicateITCase(miniBatch: MiniBatchMode, mode: StateBackendMode)
.fromCollection(rowtimeTestData)
.assignTimestampsAndWatermarks(new RowtimeExtractor)
.toTable(tEnv, 'a, 'b, 'c, 'rowtime.rowtime())
- tEnv.registerTable("T", t)
+ tEnv.createTemporaryView("T", t)
createSinkTable("rowtime_sink")
val sql =
@@ -368,7 +368,7 @@ class DeduplicateITCase(miniBatch: MiniBatchMode, mode: StateBackendMode)
.fromCollection(rowtimeTestData)
.assignTimestampsAndWatermarks(new RowtimeExtractor)
.toTable(tEnv, 'a, 'b, 'c, 'rowtime.rowtime())
- tEnv.registerTable("T", t)
+ tEnv.createTemporaryView("T", t)
tEnv.executeSql(s"""
|CREATE TABLE rowtime_sink (
| cnt BIGINT
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/GroupWindowITCase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/GroupWindowITCase.scala
index 6da8b3fe71b..ed3d86bda95 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/GroupWindowITCase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/GroupWindowITCase.scala
@@ -289,7 +289,7 @@ class GroupWindowITCase(mode: StateBackendMode, useTimestampLtz: Boolean)
.assignTimestampsAndWatermarks(
new TimestampAndWatermarkWithOffset[(Long, Int, String, String)](10L))
val table = stream.toTable(tEnv, 'rowtime.rowtime, 'int, 'string, 'name)
- tEnv.registerTable("T1", table)
+ tEnv.createTemporaryView("T1", table)
val sql =
"""
@@ -341,7 +341,7 @@ class GroupWindowITCase(mode: StateBackendMode, useTimestampLtz: Boolean)
val stream = failingDataSource(data)
.assignTimestampsAndWatermarks(new TimestampAndWatermarkWithOffset[(Long, Int, String)](0L))
val table = stream.toTable(tEnv, 'long, 'int, 'string, 'rowtime.rowtime)
- tEnv.registerTable("T1", table)
+ tEnv.createTemporaryView("T1", table)
tEnv.createTemporarySystemFunction("weightAvgFun", classOf[WeightedAvg])
val sql =
@@ -566,7 +566,7 @@ class GroupWindowITCase(mode: StateBackendMode, useTimestampLtz: Boolean)
val stream = failingDataSource(sessionWindowTestData)
.assignTimestampsAndWatermarks(new TimestampAndWatermarkWithOffset[(Long, Int, String)](10L))
val table = stream.toTable(tEnv, 'a, 'b, 'c, 'rowtime.rowtime)
- tEnv.registerTable("MyTable", table)
+ tEnv.createTemporaryView("MyTable", table)
val sqlQuery =
"""
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/IntervalJoinITCase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/IntervalJoinITCase.scala
index 45070003c6c..571dda02dd1 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/IntervalJoinITCase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/IntervalJoinITCase.scala
@@ -64,16 +64,16 @@ class IntervalJoinITCase(mode: StateBackendMode) extends StreamingWithStateTestB
data2.+=((2, 2L, "HeHe"))
val tmp1 = env.fromCollection(data1).toTable(tEnv, 'a, 'b, 'c, 'proctime.proctime)
- tEnv.registerTable("TmpT1", tmp1)
+ tEnv.createTemporaryView("TmpT1", tmp1)
val subquery1 = "SELECT IF(a = 1, CAST(NULL AS INT), a) as a, b, c, proctime FROM TmpT1"
val t1 = tEnv.sqlQuery(subquery1)
- tEnv.registerTable("T1", t1)
+ tEnv.createTemporaryView("T1", t1)
val tmp2 = env.fromCollection(data2).toTable(tEnv, 'a, 'b, 'c, 'proctime.proctime)
- tEnv.registerTable("TmpT2", tmp2)
+ tEnv.createTemporaryView("TmpT2", tmp2)
val subquery2 = "SELECT IF(a = 1, CAST(NULL AS INT), a) as a, b, c, proctime FROM TmpT2"
val t2 = tEnv.sqlQuery(subquery2)
- tEnv.registerTable("T2", t2)
+ tEnv.createTemporaryView("T2", t2)
val sink = new TestingAppendSink
val result = tEnv.sqlQuery(sqlQuery).toAppendStream[Row]
@@ -107,16 +107,16 @@ class IntervalJoinITCase(mode: StateBackendMode) extends StreamingWithStateTestB
data2.+=((2, 2L, "HeHe"))
val tmp1 = env.fromCollection(data1).toTable(tEnv, 'a, 'b, 'c, 'proctime.proctime)
- tEnv.registerTable("TmpT1", tmp1)
+ tEnv.createTemporaryView("TmpT1", tmp1)
val subquery1 = "SELECT IF(a = 1, CAST(NULL AS INT), a) as a, b, c, proctime FROM TmpT1"
val t1 = tEnv.sqlQuery(subquery1)
- tEnv.registerTable("T1", t1)
+ tEnv.createTemporaryView("T1", t1)
val tmp2 = env.fromCollection(data2).toTable(tEnv, 'a, 'b, 'c, 'proctime.proctime)
- tEnv.registerTable("TmpT2", tmp2)
+ tEnv.createTemporaryView("TmpT2", tmp2)
val subquery2 = "SELECT IF(a = 1, CAST(NULL AS INT), a) as a, b, c, proctime FROM TmpT2"
val t2 = tEnv.sqlQuery(subquery2)
- tEnv.registerTable("T2", t2)
+ tEnv.createTemporaryView("T2", t2)
val sink = new TestingAppendSink
val result = tEnv.sqlQuery(sqlQuery).toAppendStream[Row]
@@ -158,8 +158,8 @@ class IntervalJoinITCase(mode: StateBackendMode) extends StreamingWithStateTestB
val t1 = env.fromCollection(data1).toTable(tEnv, 'a, 'b, 'c, 'proctime.proctime)
val t2 = env.fromCollection(data2).toTable(tEnv, 'a, 'b, 'c, 'proctime.proctime)
- tEnv.registerTable("T1", t1)
- tEnv.registerTable("T2", t2)
+ tEnv.createTemporaryView("T1", t1)
+ tEnv.createTemporaryView("T2", t2)
val sink = new TestingAppendSink
val result = tEnv.sqlQuery(sqlQuery).toAppendStream[Row]
@@ -209,8 +209,8 @@ class IntervalJoinITCase(mode: StateBackendMode) extends StreamingWithStateTestB
.assignTimestampsAndWatermarks(new Row3WatermarkExtractor2)
.toTable(tEnv, 'key, 'id, 'rowtime.rowtime)
- tEnv.registerTable("T1", t1)
- tEnv.registerTable("T2", t2)
+ tEnv.createTemporaryView("T1", t1)
+ tEnv.createTemporaryView("T2", t2)
val sink = new TestingAppendSink
val result = tEnv.sqlQuery(sqlQuery).toAppendStream[Row]
result.addSink(sink)
@@ -264,8 +264,8 @@ class IntervalJoinITCase(mode: StateBackendMode) extends StreamingWithStateTestB
.assignTimestampsAndWatermarks(new Row3WatermarkExtractor2)
.toTable(tEnv, 'key, 'id, 'rowtime.rowtime)
- tEnv.registerTable("T1", t1)
- tEnv.registerTable("T2", t2)
+ tEnv.createTemporaryView("T1", t1)
+ tEnv.createTemporaryView("T2", t2)
val sink = new TestingAppendSink
val result = tEnv.sqlQuery(sqlQuery).toAppendStream[Row]
result.addSink(sink)
@@ -319,8 +319,8 @@ class IntervalJoinITCase(mode: StateBackendMode) extends StreamingWithStateTestB
.assignTimestampsAndWatermarks(new Row3WatermarkExtractor2)
.toTable(tEnv, 'key, 'id, 'rowtime.rowtime)
- tEnv.registerTable("T1", t1)
- tEnv.registerTable("T2", t2)
+ tEnv.createTemporaryView("T1", t1)
+ tEnv.createTemporaryView("T2", t2)
val sink = new TestingAppendSink
val result = tEnv.sqlQuery(sqlQuery).toAppendStream[Row]
result.addSink(sink)
@@ -387,8 +387,8 @@ class IntervalJoinITCase(mode: StateBackendMode) extends StreamingWithStateTestB
.assignTimestampsAndWatermarks(new Row3WatermarkExtractor2)
.toTable(tEnv, 'key, 'id, 'rowtime.rowtime)
- tEnv.registerTable("T1", t1)
- tEnv.registerTable("T2", t2)
+ tEnv.createTemporaryView("T1", t1)
+ tEnv.createTemporaryView("T2", t2)
val sink = new TestingRetractSink
val result = tEnv.sqlQuery(sqlQuery).toRetractStream[Row]
result.addSink(sink)
@@ -512,8 +512,8 @@ class IntervalJoinITCase(mode: StateBackendMode) extends StreamingWithStateTestB
.assignTimestampsAndWatermarks(new Row4WatermarkExtractor)
.toTable(tEnv, 'a, 'b, 'c, 'rowtime.rowtime)
- tEnv.registerTable("T1", t1)
- tEnv.registerTable("T2", t2)
+ tEnv.createTemporaryView("T1", t1)
+ tEnv.createTemporaryView("T2", t2)
val sink = new TestingAppendSink
val result = tEnv.sqlQuery(sqlQuery).toAppendStream[Row]
result.addSink(sink)
@@ -564,8 +564,8 @@ class IntervalJoinITCase(mode: StateBackendMode) extends StreamingWithStateTestB
.assignTimestampsAndWatermarks(new Row4WatermarkExtractor)
.toTable(tEnv, 'a, 'b, 'c, 'rowtime.rowtime)
- tEnv.registerTable("T1", t1)
- tEnv.registerTable("T2", t2)
+ tEnv.createTemporaryView("T1", t1)
+ tEnv.createTemporaryView("T2", t2)
val sink = new TestingAppendSink
val result = tEnv.sqlQuery(sqlQuery).toAppendStream[Row]
result.addSink(sink)
@@ -615,8 +615,8 @@ class IntervalJoinITCase(mode: StateBackendMode) extends StreamingWithStateTestB
.assignTimestampsAndWatermarks(new Row3WatermarkExtractor2)
.toTable(tEnv, 'key, 'id, 'rowtime.rowtime)
- tEnv.registerTable("T1", t1)
- tEnv.registerTable("T2", t2)
+ tEnv.createTemporaryView("T1", t1)
+ tEnv.createTemporaryView("T2", t2)
val sink = new TestingAppendSink
val t_r = tEnv.sqlQuery(sqlQuery)
@@ -669,8 +669,8 @@ class IntervalJoinITCase(mode: StateBackendMode) extends StreamingWithStateTestB
.assignTimestampsAndWatermarks(new Row3WatermarkExtractor2)
.toTable(tEnv, 'key, 'id, 'rowtime.rowtime)
- tEnv.registerTable("T1", t1)
- tEnv.registerTable("T2", t2)
+ tEnv.createTemporaryView("T1", t1)
+ tEnv.createTemporaryView("T2", t2)
val sink = new TestingAppendSink
val result = tEnv.sqlQuery(sqlQuery).toAppendStream[Row]
@@ -714,8 +714,8 @@ class IntervalJoinITCase(mode: StateBackendMode) extends StreamingWithStateTestB
.fromCollection(data2)
.toTable(tEnv, 'a, 'b, 'c, 'proctime.proctime)
- tEnv.registerTable("T1", t1)
- tEnv.registerTable("T2", t2)
+ tEnv.createTemporaryView("T1", t1)
+ tEnv.createTemporaryView("T2", t2)
val sink = new TestingAppendSink
val result = tEnv.sqlQuery(sqlQuery).toAppendStream[Row]
@@ -765,8 +765,8 @@ class IntervalJoinITCase(mode: StateBackendMode) extends StreamingWithStateTestB
.assignTimestampsAndWatermarks(new Row3WatermarkExtractor2)
.toTable(tEnv, 'key, 'id, 'rowtime.rowtime)
- tEnv.registerTable("T1", t1)
- tEnv.registerTable("T2", t2)
+ tEnv.createTemporaryView("T1", t1)
+ tEnv.createTemporaryView("T2", t2)
val sink = new TestingAppendSink
val result = tEnv.sqlQuery(sqlQuery).toAppendStream[Row]
@@ -821,8 +821,8 @@ class IntervalJoinITCase(mode: StateBackendMode) extends StreamingWithStateTestB
.assignTimestampsAndWatermarks(new Row3WatermarkExtractor2)
.toTable(tEnv, 'key, 'id, 'rowtime.rowtime)
- tEnv.registerTable("T1", t1)
- tEnv.registerTable("T2", t2)
+ tEnv.createTemporaryView("T1", t1)
+ tEnv.createTemporaryView("T2", t2)
val sink = new TestingAppendSink
val result = tEnv.sqlQuery(sqlQuery).toAppendStream[Row]
@@ -864,8 +864,8 @@ class IntervalJoinITCase(mode: StateBackendMode) extends StreamingWithStateTestB
val t1 = env.fromCollection(data1).toTable(tEnv, 'a, 'b, 'c, 'proctime.proctime)
val t2 = env.fromCollection(data2).toTable(tEnv, 'a, 'b, 'c, 'proctime.proctime)
- tEnv.registerTable("T1", t1)
- tEnv.registerTable("T2", t2)
+ tEnv.createTemporaryView("T1", t1)
+ tEnv.createTemporaryView("T2", t2)
val sink = new TestingAppendSink
val result = tEnv.sqlQuery(sqlQuery).toAppendStream[Row]
@@ -912,8 +912,8 @@ class IntervalJoinITCase(mode: StateBackendMode) extends StreamingWithStateTestB
.assignTimestampsAndWatermarks(new Row3WatermarkExtractor2)
.toTable(tEnv, 'key, 'id, 'rowtime.rowtime)
- tEnv.registerTable("T1", t1)
- tEnv.registerTable("T2", t2)
+ tEnv.createTemporaryView("T1", t1)
+ tEnv.createTemporaryView("T2", t2)
val sink = new TestingAppendSink
val result = tEnv.sqlQuery(sqlQuery).toAppendStream[Row]
@@ -965,8 +965,8 @@ class IntervalJoinITCase(mode: StateBackendMode) extends StreamingWithStateTestB
.assignTimestampsAndWatermarks(new Row3WatermarkExtractor2)
.toTable(tEnv, 'key, 'id, 'rowtime.rowtime)
- tEnv.registerTable("T1", t1)
- tEnv.registerTable("T2", t2)
+ tEnv.createTemporaryView("T1", t1)
+ tEnv.createTemporaryView("T2", t2)
val sink = new TestingAppendSink
val result = tEnv.sqlQuery(sqlQuery).toAppendStream[Row]
@@ -1008,8 +1008,8 @@ class IntervalJoinITCase(mode: StateBackendMode) extends StreamingWithStateTestB
val t1 = env.fromCollection(data1).toTable(tEnv, 'a, 'b, 'c, 'proctime.proctime)
val t2 = env.fromCollection(data2).toTable(tEnv, 'a, 'b, 'c, 'proctime.proctime)
- tEnv.registerTable("T1", t1)
- tEnv.registerTable("T2", t2)
+ tEnv.createTemporaryView("T1", t1)
+ tEnv.createTemporaryView("T2", t2)
val sink = new TestingAppendSink
val result = tEnv.sqlQuery(sqlQuery).toAppendStream[Row]
@@ -1056,8 +1056,8 @@ class IntervalJoinITCase(mode: StateBackendMode) extends StreamingWithStateTestB
.assignTimestampsAndWatermarks(new Row3WatermarkExtractor2)
.toTable(tEnv, 'key, 'id, 'rowtime.rowtime)
- tEnv.registerTable("T1", t1)
- tEnv.registerTable("T2", t2)
+ tEnv.createTemporaryView("T1", t1)
+ tEnv.createTemporaryView("T2", t2)
val sink = new TestingAppendSink
val result = tEnv.sqlQuery(sqlQuery).toAppendStream[Row]
@@ -1111,8 +1111,8 @@ class IntervalJoinITCase(mode: StateBackendMode) extends StreamingWithStateTestB
.assignTimestampsAndWatermarks(new Row3WatermarkExtractor2)
.toTable(tEnv, 'key, 'id, 'rowtime.rowtime)
- tEnv.registerTable("T1", t1)
- tEnv.registerTable("T2", t2)
+ tEnv.createTemporaryView("T1", t1)
+ tEnv.createTemporaryView("T2", t2)
val sink = new TestingAppendSink
val result = tEnv.sqlQuery(sqlQuery).toAppendStream[Row]
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/JoinITCase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/JoinITCase.scala
index ac59bc2623f..c94a67f8a18 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/JoinITCase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/JoinITCase.scala
@@ -66,8 +66,8 @@ class JoinITCase(state: StateBackendMode) extends StreamingWithStateTestBase(sta
.toTable(tEnv, 'a1, 'a2, 'a3)
val tableB = failingDataSource(TestData.tupleData5)
.toTable(tEnv, 'b1, 'b2, 'b3, 'b4, 'b5)
- tEnv.registerTable("A", tableA)
- tEnv.registerTable("B", tableB)
+ tEnv.createTemporaryView("A", tableA)
+ tEnv.createTemporaryView("B", tableB)
val dataId1 = TestValuesTableFactory.registerData(TestData.data2_1)
tEnv.executeSql(s"""
@@ -165,8 +165,8 @@ class JoinITCase(state: StateBackendMode) extends StreamingWithStateTestBase(sta
val table1 = failingDataSource(data1).toTable(tEnv, 'a1, 'a2, 'a3)
val table2 = failingDataSource(data2).toTable(tEnv, 'b1, 'b2, 'b3)
- tEnv.registerTable("a", table1)
- tEnv.registerTable("b", table2)
+ tEnv.createTemporaryView("a", table1)
+ tEnv.createTemporaryView("b", table2)
val sqlQuery = "SELECT * FROM a, b WHERE (a1 = 1 AND b1 = 3) OR (a1 = 2 AND b3 is null) "
@@ -200,8 +200,8 @@ class JoinITCase(state: StateBackendMode) extends StreamingWithStateTestBase(sta
val t1 = failingDataSource(data1).toTable(tEnv, 'a, 'b, 'c)
val t2 = failingDataSource(data2).toTable(tEnv, 'a, 'b, 'c)
- tEnv.registerTable("T1", t1)
- tEnv.registerTable("T2", t2)
+ tEnv.createTemporaryView("T1", t1)
+ tEnv.createTemporaryView("T2", t2)
val sqlQuery =
"""
@@ -250,8 +250,8 @@ class JoinITCase(state: StateBackendMode) extends StreamingWithStateTestBase(sta
val t1 = failingDataSource(data1).toTable(tEnv, 'a, 'b, 'c)
val t2 = failingDataSource(data2).toTable(tEnv, 'a, 'b, 'c)
- tEnv.registerTable("T1", t1)
- tEnv.registerTable("T2", t2)
+ tEnv.createTemporaryView("T1", t1)
+ tEnv.createTemporaryView("T2", t2)
val sqlQuery =
"""
@@ -300,8 +300,8 @@ class JoinITCase(state: StateBackendMode) extends StreamingWithStateTestBase(sta
def testInnerJoin(): Unit = {
val ds1 = failingDataSource(tuple3Data).toTable(tEnv, 'a, 'b, 'c)
val ds2 = failingDataSource(smallTuple5Data).toTable(tEnv, 'd, 'e, 'f, 'g, 'h)
- tEnv.registerTable("ds1", ds1)
- tEnv.registerTable("ds2", ds2)
+ tEnv.createTemporaryView("ds1", ds1)
+ tEnv.createTemporaryView("ds2", ds2)
val query = "SELECT b, c, e, g FROM ds1 JOIN ds2 ON b = e"
val sink = new TestingAppendSink
@@ -357,8 +357,8 @@ class JoinITCase(state: StateBackendMode) extends StreamingWithStateTestBase(sta
val t1 = failingDataSource(data1).toTable(tEnv, 'a1, 'b1, 'c1, 'd1)
val t2 = failingDataSource(data1).toTable(tEnv, 'a2, 'b2, 'c2, 'd2)
- tEnv.registerTable("Table3", t1)
- tEnv.registerTable("Table5", t2)
+ tEnv.createTemporaryView("Table3", t1)
+ tEnv.createTemporaryView("Table5", t2)
val sqlQuery = "SELECT a1, a1, c2 FROM Table3 INNER JOIN Table5 ON d1 = d2 where d1 is true"
@@ -376,8 +376,8 @@ class JoinITCase(state: StateBackendMode) extends StreamingWithStateTestBase(sta
val ds1 = failingDataSource(TestData.tupleData3).toTable(tEnv, 'a, 'b, 'c)
val ds2 = failingDataSource(TestData.tupleData5).toTable(tEnv, 'd, 'e, 'f, 'g, 'h)
- tEnv.registerTable("Table3", ds1)
- tEnv.registerTable("Table5", ds2)
+ tEnv.createTemporaryView("Table3", ds1)
+ tEnv.createTemporaryView("Table5", ds2)
val sink = new TestingRetractSink
tEnv.sqlQuery(sqlQuery).toRetractStream[Row].addSink(sink).setParallelism(1)
@@ -393,8 +393,8 @@ class JoinITCase(state: StateBackendMode) extends StreamingWithStateTestBase(sta
val ds1 = failingDataSource(TestData.tupleData3).toTable(tEnv, 'a, 'b, 'c)
val ds2 = failingDataSource(TestData.tupleData5).toTable(tEnv, 'd, 'e, 'f, 'g, 'h)
- tEnv.registerTable("Table3", ds1)
- tEnv.registerTable("Table5", ds2)
+ tEnv.createTemporaryView("Table3", ds1)
+ tEnv.createTemporaryView("Table5", ds2)
val sink = new TestingRetractSink
tEnv.sqlQuery(sqlQuery).toRetractStream[Row].addSink(sink).setParallelism(1)
@@ -441,8 +441,8 @@ class JoinITCase(state: StateBackendMode) extends StreamingWithStateTestBase(sta
def testLeftOuterJoin(): Unit = {
val ds1 = failingDataSource(tuple3Data).toTable(tEnv, 'a, 'b, 'c)
val ds2 = failingDataSource(dataCannotBeJoin).toTable(tEnv, 'd, 'e, 'f, 'g, 'h)
- tEnv.registerTable("ds1", ds1)
- tEnv.registerTable("ds2", ds2)
+ tEnv.createTemporaryView("ds1", ds1)
+ tEnv.createTemporaryView("ds2", ds2)
val query = "SELECT b, c, e, g FROM ds1 LEFT OUTER JOIN ds2 ON b = e"
val sink = new TestingRetractSink
@@ -457,8 +457,8 @@ class JoinITCase(state: StateBackendMode) extends StreamingWithStateTestBase(sta
def testLeftOuterJoinWithRetraction(): Unit = {
val ds1 = failingDataSource(tuple3Data).toTable(tEnv, 'a, 'b, 'c)
val ds2 = failingDataSource(smallTuple5Data).toTable(tEnv, 'd, 'e, 'f, 'g, 'h)
- tEnv.registerTable("ds1", ds1)
- tEnv.registerTable("ds2", ds2)
+ tEnv.createTemporaryView("ds1", ds1)
+ tEnv.createTemporaryView("ds2", ds2)
val query = "SELECT b, c, e, g FROM ds1 LEFT OUTER JOIN ds2 ON b = e"
val sink = new TestingRetractSink
@@ -478,8 +478,8 @@ class JoinITCase(state: StateBackendMode) extends StreamingWithStateTestBase(sta
val table1 = failingDataSource(data1).toTable(tEnv, 'pk, 'a)
val table2 = failingDataSource(data2).toTable(tEnv, 'pk, 'a)
- tEnv.registerTable("ds1", table1)
- tEnv.registerTable("ds2", table2)
+ tEnv.createTemporaryView("ds1", table1)
+ tEnv.createTemporaryView("ds2", table2)
val sql =
"""
@@ -544,8 +544,8 @@ class JoinITCase(state: StateBackendMode) extends StreamingWithStateTestBase(sta
val ds1 = failingDataSource(TestData.smallTupleData3).toTable(tEnv, 'a, 'b, 'c)
val ds2 = failingDataSource(TestData.tupleData5).toTable(tEnv, 'd, 'e, 'f, 'g, 'h)
- tEnv.registerTable("Table3", ds1)
- tEnv.registerTable("Table5", ds2)
+ tEnv.createTemporaryView("Table3", ds1)
+ tEnv.createTemporaryView("Table5", ds2)
val sink = new TestingRetractSink
tEnv.sqlQuery(sqlQuery).toRetractStream[Row].addSink(sink).setParallelism(1)
@@ -1121,8 +1121,8 @@ class JoinITCase(state: StateBackendMode) extends StreamingWithStateTestBase(sta
val t1 = failingDataSource(data1).toTable(tEnv, 'a, 'b)
val t2 = failingDataSource(data2).toTable(tEnv, 'a, 'b)
- tEnv.registerTable("T1", t1)
- tEnv.registerTable("T2", t2)
+ tEnv.createTemporaryView("T1", t1)
+ tEnv.createTemporaryView("T2", t2)
val sqlQuery =
"""
@@ -1164,8 +1164,8 @@ class JoinITCase(state: StateBackendMode) extends StreamingWithStateTestBase(sta
val t1 = failingDataSource(data1).toTable(tEnv, 'a, 'b)
val t2 = failingDataSource(data2).toTable(tEnv, 'a, 'b)
- tEnv.registerTable("T1", t1)
- tEnv.registerTable("T2", t2)
+ tEnv.createTemporaryView("T1", t1)
+ tEnv.createTemporaryView("T2", t2)
val sqlQuery =
"""
@@ -1207,8 +1207,8 @@ class JoinITCase(state: StateBackendMode) extends StreamingWithStateTestBase(sta
val t1 = failingDataSource(data1).toTable(tEnv, 'a, 'b)
val t2 = failingDataSource(data2).toTable(tEnv, 'a, 'b)
- tEnv.registerTable("T1", t1)
- tEnv.registerTable("T2", t2)
+ tEnv.createTemporaryView("T1", t1)
+ tEnv.createTemporaryView("T2", t2)
val sqlQuery =
"""
@@ -1249,8 +1249,8 @@ class JoinITCase(state: StateBackendMode) extends StreamingWithStateTestBase(sta
val t1 = failingDataSource(data1).toTable(tEnv, 'a, 'b)
val t2 = failingDataSource(data2).toTable(tEnv, 'a, 'b)
- tEnv.registerTable("T1", t1)
- tEnv.registerTable("T2", t2)
+ tEnv.createTemporaryView("T1", t1)
+ tEnv.createTemporaryView("T2", t2)
val sqlQuery =
"""
@@ -1292,8 +1292,8 @@ class JoinITCase(state: StateBackendMode) extends StreamingWithStateTestBase(sta
val t1 = failingDataSource(data1).toTable(tEnv, 'a, 'b)
val t2 = failingDataSource(data2).toTable(tEnv, 'a, 'b)
- tEnv.registerTable("T1", t1)
- tEnv.registerTable("T2", t2)
+ tEnv.createTemporaryView("T1", t1)
+ tEnv.createTemporaryView("T2", t2)
val sqlQuery =
"""
@@ -1337,8 +1337,8 @@ class JoinITCase(state: StateBackendMode) extends StreamingWithStateTestBase(sta
val t1 = failingDataSource(data1).toTable(tEnv, 'a, 'b)
val t2 = failingDataSource(data2).toTable(tEnv, 'a, 'b)
- tEnv.registerTable("T1", t1)
- tEnv.registerTable("T2", t2)
+ tEnv.createTemporaryView("T1", t1)
+ tEnv.createTemporaryView("T2", t2)
val sqlQuery =
"""
@@ -1383,9 +1383,9 @@ class JoinITCase(state: StateBackendMode) extends StreamingWithStateTestBase(sta
data2.+=((3, -3L))
val t1 = failingDataSource(data1).toTable(tEnv, 'a, 'b)
- tEnv.registerTable("T1", t1)
+ tEnv.createTemporaryView("T1", t1)
val t2 = failingDataSource(data2).toTable(tEnv, 'a, 'c)
- tEnv.registerTable("T2", t2)
+ tEnv.createTemporaryView("T2", t2)
val t3 = tEnv.sqlQuery("select T1.a, b, c from T1, T2 WHERE T1.a = T2.a")
val sink = new TestingRetractSink
@@ -1406,8 +1406,8 @@ class JoinITCase(state: StateBackendMode) extends StreamingWithStateTestBase(sta
val t1 = failingDataSource(data).toTable(tEnv, 'a, 'b, 'c)
val t2 = failingDataSource(data).toTable(tEnv, 'd, 'e, 'f)
- tEnv.registerTable("T1", t1)
- tEnv.registerTable("T2", t2)
+ tEnv.createTemporaryView("T1", t1)
+ tEnv.createTemporaryView("T2", t2)
val sql =
"""
@@ -1427,8 +1427,8 @@ class JoinITCase(state: StateBackendMode) extends StreamingWithStateTestBase(sta
val ds1 = failingDataSource(TestData.smallTupleData3).toTable(tEnv, 'a, 'b, 'c)
val ds2 = failingDataSource(TestData.tupleData5).toTable(tEnv, 'd, 'e, 'f, 'g, 'h)
- tEnv.registerTable("T3", ds1)
- tEnv.registerTable("T5", ds2)
+ tEnv.createTemporaryView("T3", ds1)
+ tEnv.createTemporaryView("T5", ds2)
tEnv.registerFunction("funcWithOpen", new FuncWithOpen)
val sql = "SELECT c, g FROM T3 join T5 on funcWithOpen(a + d) where b = e"
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/Limit0RemoveITCase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/Limit0RemoveITCase.scala
index ee72bd5913b..7119667894e 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/Limit0RemoveITCase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/Limit0RemoveITCase.scala
@@ -32,7 +32,7 @@ class Limit0RemoveITCase extends StreamingTestBase() {
def testSimpleLimitRemove(): Unit = {
val ds = env.fromCollection(Seq(1, 2, 3, 4, 5, 6))
val table = ds.toTable(tEnv, 'a)
- tEnv.registerTable("MyTable", table)
+ tEnv.createTemporaryView("MyTable", table)
val sql = "SELECT * FROM MyTable LIMIT 0"
@@ -48,7 +48,7 @@ class Limit0RemoveITCase extends StreamingTestBase() {
def testLimitRemoveWithOrderBy(): Unit = {
val ds = env.fromCollection(Seq(1, 2, 3, 4, 5, 6))
val table = ds.toTable(tEnv, 'a)
- tEnv.registerTable("MyTable", table)
+ tEnv.createTemporaryView("MyTable", table)
val sql = "SELECT * FROM MyTable ORDER BY a LIMIT 0"
@@ -64,7 +64,7 @@ class Limit0RemoveITCase extends StreamingTestBase() {
def testLimitRemoveWithSelect(): Unit = {
val ds = env.fromCollection(Seq(1, 2, 3, 4, 5, 6))
val table = ds.toTable(tEnv, 'a)
- tEnv.registerTable("MyTable", table)
+ tEnv.createTemporaryView("MyTable", table)
val sql = "select a2 from (select cast(a as int) a2 from MyTable limit 0)"
@@ -80,11 +80,11 @@ class Limit0RemoveITCase extends StreamingTestBase() {
def testLimitRemoveWithIn(): Unit = {
val ds1 = env.fromCollection(Seq(1, 2, 3, 4, 5, 6))
val table1 = ds1.toTable(tEnv, 'a)
- tEnv.registerTable("MyTable1", table1)
+ tEnv.createTemporaryView("MyTable1", table1)
val ds2 = env.fromCollection(Seq(1, 2, 3))
val table2 = ds2.toTable(tEnv, 'a)
- tEnv.registerTable("MyTable2", table2)
+ tEnv.createTemporaryView("MyTable2", table2)
val sql = "SELECT * FROM MyTable1 WHERE a IN (SELECT a FROM MyTable2 LIMIT 0)"
@@ -100,11 +100,11 @@ class Limit0RemoveITCase extends StreamingTestBase() {
def testLimitRemoveWithNotIn(): Unit = {
val ds1 = env.fromCollection(Seq(1, 2, 3, 4, 5, 6))
val table1 = ds1.toTable(tEnv, 'a)
- tEnv.registerTable("MyTable1", table1)
+ tEnv.createTemporaryView("MyTable1", table1)
val ds2 = env.fromCollection(Seq(1, 2, 3))
val table2 = ds2.toTable(tEnv, 'a)
- tEnv.registerTable("MyTable2", table2)
+ tEnv.createTemporaryView("MyTable2", table2)
val sql = "SELECT * FROM MyTable1 WHERE a NOT IN (SELECT a FROM MyTable2 LIMIT 0)"
@@ -121,11 +121,11 @@ class Limit0RemoveITCase extends StreamingTestBase() {
def testLimitRemoveWithExists(): Unit = {
val ds1 = env.fromCollection(Seq(1, 2, 3, 4, 5, 6))
val table1 = ds1.toTable(tEnv, 'a)
- tEnv.registerTable("MyTable1", table1)
+ tEnv.createTemporaryView("MyTable1", table1)
val ds2 = env.fromCollection(Seq(1, 2, 3))
val table2 = ds2.toTable(tEnv, 'a)
- tEnv.registerTable("MyTable2", table2)
+ tEnv.createTemporaryView("MyTable2", table2)
val sql = "SELECT * FROM MyTable1 WHERE EXISTS (SELECT a FROM MyTable2 LIMIT 0)"
@@ -141,11 +141,11 @@ class Limit0RemoveITCase extends StreamingTestBase() {
def testLimitRemoveWithNotExists(): Unit = {
val ds1 = env.fromCollection(Seq(1, 2, 3, 4, 5, 6))
val table1 = ds1.toTable(tEnv, 'a)
- tEnv.registerTable("MyTable1", table1)
+ tEnv.createTemporaryView("MyTable1", table1)
val ds2 = env.fromCollection(Seq(1, 2, 3))
val table2 = ds2.toTable(tEnv, 'a)
- tEnv.registerTable("MyTable2", table2)
+ tEnv.createTemporaryView("MyTable2", table2)
val sql = "SELECT * FROM MyTable1 WHERE NOT EXISTS (SELECT a FROM MyTable2 LIMIT 0)"
@@ -162,11 +162,11 @@ class Limit0RemoveITCase extends StreamingTestBase() {
def testLimitRemoveWithJoin(): Unit = {
val ds1 = env.fromCollection(Seq(1, 2, 3, 4, 5, 6))
val table1 = ds1.toTable(tEnv, 'a1)
- tEnv.registerTable("MyTable1", table1)
+ tEnv.createTemporaryView("MyTable1", table1)
val ds2 = env.fromCollection(Seq(1, 2, 3))
val table2 = ds2.toTable(tEnv, 'a2)
- tEnv.registerTable("MyTable2", table2)
+ tEnv.createTemporaryView("MyTable2", table2)
val sql = "SELECT a1 FROM MyTable1 INNER JOIN (SELECT a2 FROM MyTable2 LIMIT 0) ON true"
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/LimitITCase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/LimitITCase.scala
index 3dd81b7bd18..31b0376babd 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/LimitITCase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/LimitITCase.scala
@@ -43,7 +43,7 @@ class LimitITCase(mode: StateBackendMode) extends StreamingWithStateTestBase(mod
("fruit", 5, 22))
val ds = failingDataSource(data).toTable(tEnv, 'category, 'shopId, 'num)
- tEnv.registerTable("T", ds)
+ tEnv.createTemporaryView("T", ds)
val sql = "SELECT * FROM T LIMIT 4"
@@ -66,7 +66,7 @@ class LimitITCase(mode: StateBackendMode) extends StreamingWithStateTestBase(mod
("fruit", 5, 22))
val ds = failingDataSource(data).toTable(tEnv, 'category, 'shopId, 'num)
- tEnv.registerTable("T", ds)
+ tEnv.createTemporaryView("T", ds)
val sql = "SELECT * FROM T LIMIT 4 OFFSET 2"
@@ -90,7 +90,7 @@ class LimitITCase(mode: StateBackendMode) extends StreamingWithStateTestBase(mod
("fruit", 5, 22))
val t = env.fromCollection(data).toTable(tEnv, 'category, 'shopId, 'num)
- tEnv.registerTable("T", t)
+ tEnv.createTemporaryView("T", t)
val sql = "SELECT * FROM T OFFSET 2"
thrown.expect(classOf[TableException])
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/LookupJoinITCase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/LookupJoinITCase.scala
index 08dec3d4321..9fdbe7b4b23 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/LookupJoinITCase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/LookupJoinITCase.scala
@@ -671,7 +671,7 @@ class LookupJoinITCase(legacyTableSource: Boolean, cacheType: LookupCacheType)
val sql1 = "SELECT max(id) as id, PROCTIME() as proctime FROM src AS T group by len"
val table1 = tEnv.sqlQuery(sql1)
- tEnv.registerTable("t1", table1)
+ tEnv.createTemporaryView("t1", table1)
val sql2 = "SELECT t1.id, D.name, D.age FROM t1 LEFT JOIN user_table " +
"for system_time as of t1.proctime AS D ON t1.id = D.id"
@@ -693,7 +693,7 @@ class LookupJoinITCase(legacyTableSource: Boolean, cacheType: LookupCacheType)
val sql1 = "SELECT max(id) as id, PROCTIME() as proctime FROM src AS T group by len"
val table1 = tEnv.sqlQuery(sql1)
- tEnv.registerTable("t1", table1)
+ tEnv.createTemporaryView("t1", table1)
val sql2 = "SELECT t1.id, D.name, D.age FROM t1 LEFT JOIN user_table " +
"for system_time as of t1.proctime AS D ON D.id = 3"
@@ -716,7 +716,7 @@ class LookupJoinITCase(legacyTableSource: Boolean, cacheType: LookupCacheType)
val sql1 = "SELECT max(id) as id, PROCTIME() as proctime FROM src AS T group by len"
val table1 = tEnv.sqlQuery(sql1)
- tEnv.registerTable("t1", table1)
+ tEnv.createTemporaryView("t1", table1)
val sql2 = "SELECT t1.id FROM t1 LEFT JOIN user_table " +
"for system_time as of t1.proctime AS D ON D.id = 3"
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/MatchRecognizeITCase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/MatchRecognizeITCase.scala
index 2c79726f7fc..4516ffb0839 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/MatchRecognizeITCase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/MatchRecognizeITCase.scala
@@ -63,7 +63,7 @@ class MatchRecognizeITCase(backend: StateBackendMode) extends StreamingWithState
data.+=((9, "h"))
val t = env.fromCollection(data).toTable(tEnv, 'id, 'name, 'proctime.proctime)
- tEnv.registerTable("MyTable", t)
+ tEnv.createTemporaryView("MyTable", t)
val sqlQuery =
s"""
@@ -109,7 +109,7 @@ class MatchRecognizeITCase(backend: StateBackendMode) extends StreamingWithState
data.+=((9, null, null))
val t = env.fromCollection(data).toTable(tEnv, 'id, 'name, 'nullField, 'proctime.proctime)
- tEnv.registerTable("MyTable", t)
+ tEnv.createTemporaryView("MyTable", t)
val sqlQuery =
s"""
@@ -161,7 +161,7 @@ class MatchRecognizeITCase(backend: StateBackendMode) extends StreamingWithState
val t = env
.fromCollection(data)
.toTable(tEnv, 'id, 'name, 'key1, 'key2, 'proctime.proctime)
- tEnv.registerTable("MyTable", t)
+ tEnv.createTemporaryView("MyTable", t)
val sqlQuery =
s"""
@@ -222,7 +222,7 @@ class MatchRecognizeITCase(backend: StateBackendMode) extends StreamingWithState
val t = env
.addSource(new EventTimeSourceFunction[(Int, Int, String, Int)](data))
.toTable(tEnv, 'secondaryOrder, 'ternaryOrder, 'name, 'id, 'rowtime.rowtime)
- tEnv.registerTable("MyTable", t)
+ tEnv.createTemporaryView("MyTable", t)
val sqlQuery =
s"""
@@ -278,7 +278,7 @@ class MatchRecognizeITCase(backend: StateBackendMode) extends StreamingWithState
.fromCollection(data)
.assignAscendingTimestamps(e => e._2)
.toTable(tEnv, 'symbol, 'rowtime.rowtime, 'price, 'tax)
- tEnv.registerTable("Ticker", t)
+ tEnv.createTemporaryView("Ticker", t)
val sqlQuery =
s"""
@@ -332,7 +332,7 @@ class MatchRecognizeITCase(backend: StateBackendMode) extends StreamingWithState
.assignAscendingTimestamps(tickerEvent => tickerEvent._2)
.toTable(tEnv, 'symbol, 'rowtime.rowtime, 'price, 'tax)
- tEnv.registerTable("Ticker", tickerEvents)
+ tEnv.createTemporaryView("Ticker", tickerEvents)
val sqlQuery =
s"""
@@ -450,7 +450,7 @@ class MatchRecognizeITCase(backend: StateBackendMode) extends StreamingWithState
val t = env
.fromCollection(data)
.toTable(tEnv, 'symbol, 'tstamp, 'price, 'tax, 'proctime.proctime)
- tEnv.registerTable("Ticker", t)
+ tEnv.createTemporaryView("Ticker", t)
val sqlQuery =
s"""
@@ -498,7 +498,7 @@ class MatchRecognizeITCase(backend: StateBackendMode) extends StreamingWithState
.assignAscendingTimestamps(tickerEvent => tickerEvent._2)
.setParallelism(env.getParallelism)
.toTable(tEnv, 'symbol, 'rowtime.rowtime, 'price, 'tax)
- tEnv.registerTable("Ticker", t)
+ tEnv.createTemporaryView("Ticker", t)
val sqlQuery =
s"""
@@ -546,7 +546,7 @@ class MatchRecognizeITCase(backend: StateBackendMode) extends StreamingWithState
val t = env
.fromCollection(data)
.toTable(tEnv, 'id, 'symbol, 'tstamp, 'price, 'proctime.proctime)
- tEnv.registerTable("Ticker", t)
+ tEnv.createTemporaryView("Ticker", t)
val sqlQuery =
s"""
@@ -603,7 +603,7 @@ class MatchRecognizeITCase(backend: StateBackendMode) extends StreamingWithState
val t = env
.fromCollection(data)
.toTable(tEnv, 'symbol, 'tstamp, 'price, 'tax, 'proctime.proctime)
- tEnv.registerTable("Ticker", t)
+ tEnv.createTemporaryView("Ticker", t)
val sqlQuery =
s"""
@@ -663,7 +663,7 @@ class MatchRecognizeITCase(backend: StateBackendMode) extends StreamingWithState
val t = env
.fromCollection(data)
.toTable(tEnv, 'id, 'name, 'price, 'rate, 'weight, 'proctime.proctime)
- tEnv.registerTable("MyTable", t)
+ tEnv.createTemporaryView("MyTable", t)
tEnv.createTemporarySystemFunction("weightedAvg", classOf[WeightedAvg])
val sqlQuery =
@@ -726,7 +726,7 @@ class MatchRecognizeITCase(backend: StateBackendMode) extends StreamingWithState
BasicTypeInfo.STRING_TYPE_INFO,
BasicTypeInfo.INT_TYPE_INFO))
.toTable(tEnv, 'id, 'name, 'price, 'proctime.proctime)
- tEnv.registerTable("MyTable", t)
+ tEnv.createTemporaryView("MyTable", t)
tEnv.registerFunction("weightedAvg", new WeightedAvg)
val sqlQuery =
@@ -768,7 +768,7 @@ class MatchRecognizeITCase(backend: StateBackendMode) extends StreamingWithState
data.+=((1, "a"))
val t = env.fromCollection(data).toTable(tEnv, 'id, 'name, 'proctime.proctime)
- tEnv.registerTable("MyTable", t)
+ tEnv.createTemporaryView("MyTable", t)
val sqlQuery =
s"""
@@ -818,7 +818,7 @@ class MatchRecognizeITCase(backend: StateBackendMode) extends StreamingWithState
val t = env
.fromCollection(data)
.toTable(tEnv, 'id, 'name, 'price, 'proctime.proctime)
- tEnv.registerTable("MyTable", t)
+ tEnv.createTemporaryView("MyTable", t)
tEnv.registerFunction("prefix", new PrefixingScalarFunc)
tEnv.registerFunction("countFrom", new RichAggFunc)
val prefix = "PREF"
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/OverAggregateITCase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/OverAggregateITCase.scala
index 61874a2a2b2..246e77b227d 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/OverAggregateITCase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/OverAggregateITCase.scala
@@ -87,7 +87,7 @@ class OverAggregateITCase(mode: StateBackendMode) extends StreamingWithStateTest
.setParallelism(source.parallelism)
.toTable(tEnv, 'a, 'b, 'c, 'rowtime.rowtime)
- tEnv.registerTable("T1", t1)
+ tEnv.createTemporaryView("T1", t1)
val sink = new TestingAppendSink
tEnv.sqlQuery(sqlQuery).toAppendStream[Row].addSink(sink)
@@ -128,7 +128,7 @@ class OverAggregateITCase(mode: StateBackendMode) extends StreamingWithStateTest
.transform("TimeAssigner", new EventTimeProcessOperator[(Int, Long, String)])
.setParallelism(source.parallelism)
.toTable(tEnv, 'a, 'b, 'c, 'rowtime.rowtime)
- tEnv.registerTable("T1", t1)
+ tEnv.createTemporaryView("T1", t1)
val sink = new TestingAppendSink
tEnv.sqlQuery(sqlQuery).toAppendStream[Row].addSink(sink)
env.execute()
@@ -138,7 +138,7 @@ class OverAggregateITCase(mode: StateBackendMode) extends StreamingWithStateTest
def testRowNumberOnOver(): Unit = {
val t = failingDataSource(TestData.tupleData5)
.toTable(tEnv, 'a, 'b, 'c, 'd, 'e, 'proctime.proctime)
- tEnv.registerTable("MyTable", t)
+ tEnv.createTemporaryView("MyTable", t)
val sqlQuery = "SELECT a, ROW_NUMBER() OVER (PARTITION BY a ORDER BY proctime()) FROM MyTable"
val sink = new TestingAppendSink
@@ -168,7 +168,7 @@ class OverAggregateITCase(mode: StateBackendMode) extends StreamingWithStateTest
def testProcTimeBoundedPartitionedRowsOver(): Unit = {
val t = failingDataSource(TestData.tupleData5)
.toTable(tEnv, 'a, 'b, 'c, 'd, 'e, 'proctime.proctime)
- tEnv.registerTable("MyTable", t)
+ tEnv.createTemporaryView("MyTable", t)
val sqlQuery = "SELECT a, " +
" SUM(c) OVER (" +
@@ -203,7 +203,7 @@ class OverAggregateITCase(mode: StateBackendMode) extends StreamingWithStateTest
@Test
def testProcTimeBoundedPartitionedRowsOverWithBulitinProctime(): Unit = {
val t = failingDataSource(TestData.tupleData5).toTable(tEnv, 'a, 'b, 'c, 'd, 'e)
- tEnv.registerTable("MyTable", t)
+ tEnv.createTemporaryView("MyTable", t)
val sqlQuery = "SELECT a, " +
" SUM(c) OVER (" +
@@ -238,7 +238,7 @@ class OverAggregateITCase(mode: StateBackendMode) extends StreamingWithStateTest
@Test
def testProcTimeBoundedPartitionedRowsOverWithBuiltinProctime(): Unit = {
val t = failingDataSource(TestData.tupleData5).toTable(tEnv, 'a, 'b, 'c, 'd, 'e)
- tEnv.registerTable("MyTable", t)
+ tEnv.createTemporaryView("MyTable", t)
val sqlQuery = "SELECT a, " +
" SUM(c) OVER (" +
@@ -274,7 +274,7 @@ class OverAggregateITCase(mode: StateBackendMode) extends StreamingWithStateTest
def testProcTimeBoundedNonPartitionedRowsOver(): Unit = {
val t = failingDataSource(TestData.tupleData5)
.toTable(tEnv, 'a, 'b, 'c, 'd, 'e, 'proctime.proctime)
- tEnv.registerTable("MyTable", t)
+ tEnv.createTemporaryView("MyTable", t)
val sqlQuery = "SELECT a, " +
" first_value(d) OVER (" +
@@ -315,7 +315,7 @@ class OverAggregateITCase(mode: StateBackendMode) extends StreamingWithStateTest
def testProcTimeUnboundedPartitionedRangeOver(): Unit = {
val t1 = failingDataSource(data).toTable(tEnv, 'a, 'b, 'c, 'proctime.proctime)
- tEnv.registerTable("T1", t1)
+ tEnv.createTemporaryView("T1", t1)
val sqlQuery = "SELECT " +
"c, " +
@@ -347,7 +347,7 @@ class OverAggregateITCase(mode: StateBackendMode) extends StreamingWithStateTest
def testProcTimeUnboundedPartitionedRowsOver(): Unit = {
val t1 = failingDataSource(data).toTable(tEnv, 'a, 'b, 'c, 'proctime.proctime)
- tEnv.registerTable("T1", t1)
+ tEnv.createTemporaryView("T1", t1)
val sql =
"""
@@ -387,7 +387,7 @@ class OverAggregateITCase(mode: StateBackendMode) extends StreamingWithStateTest
def testProcTimeUnboundedNonPartitionedRangeOver(): Unit = {
val t1 = failingDataSource(data).toTable(tEnv, 'a, 'b, 'c, 'proctime.proctime)
- tEnv.registerTable("T1", t1)
+ tEnv.createTemporaryView("T1", t1)
val sqlQuery = "SELECT " +
"c, " +
@@ -416,7 +416,7 @@ class OverAggregateITCase(mode: StateBackendMode) extends StreamingWithStateTest
def testProcTimeUnboundedNonPartitionedRowsOver(): Unit = {
val t1 = failingDataSource(data).toTable(tEnv, 'a, 'b, 'c, 'proctime.proctime)
- tEnv.registerTable("T1", t1)
+ tEnv.createTemporaryView("T1", t1)
val sqlQuery = "SELECT " +
"listagg(distinct c, '|') " +
@@ -483,7 +483,7 @@ class OverAggregateITCase(mode: StateBackendMode) extends StreamingWithStateTest
.setParallelism(source.parallelism)
.toTable(tEnv, 'a, 'b, 'c, 'rowtime.rowtime)
- tEnv.registerTable("T1", t1)
+ tEnv.createTemporaryView("T1", t1)
tEnv.registerFunction("LTCNT", new LargerThanCount)
val sqlQuery = "SELECT " +
@@ -555,7 +555,7 @@ class OverAggregateITCase(mode: StateBackendMode) extends StreamingWithStateTest
.setParallelism(source.parallelism)
.toTable(tEnv, 'a, 'b, 'c, 'rowtime.rowtime)
- tEnv.registerTable("T1", t1)
+ tEnv.createTemporaryView("T1", t1)
tEnv.registerFunction("LTCNT", new LargerThanCount)
val sqlQuery = "SELECT " +
@@ -634,7 +634,7 @@ class OverAggregateITCase(mode: StateBackendMode) extends StreamingWithStateTest
.setParallelism(source.parallelism)
.toTable(tEnv, 'a, 'b, 'c, 'rowtime.rowtime)
- tEnv.registerTable("T1", t1)
+ tEnv.createTemporaryView("T1", t1)
val sqlQuery = "SELECT " +
" c, b, " +
@@ -703,7 +703,7 @@ class OverAggregateITCase(mode: StateBackendMode) extends StreamingWithStateTest
.setParallelism(source.parallelism)
.toTable(tEnv, 'a, 'b, 'c, 'rowtime.rowtime)
- tEnv.registerTable("T1", t1)
+ tEnv.createTemporaryView("T1", t1)
val sqlQuery = "SELECT " +
"c, a, " +
@@ -776,7 +776,7 @@ class OverAggregateITCase(mode: StateBackendMode) extends StreamingWithStateTest
.setParallelism(source.parallelism)
.toTable(tEnv, 'a, 'b, 'c, 'rowtime.rowtime)
- tEnv.registerTable("T1", t1)
+ tEnv.createTemporaryView("T1", t1)
tEnv.registerFunction("LTCNT", new LargerThanCount)
val sink = new TestingAppendSink
@@ -844,7 +844,7 @@ class OverAggregateITCase(mode: StateBackendMode) extends StreamingWithStateTest
.setParallelism(source.parallelism)
.toTable(tEnv, 'a, 'b, 'c, 'rowtime.rowtime)
- tEnv.registerTable("T1", t1)
+ tEnv.createTemporaryView("T1", t1)
tEnv.registerFunction("LTCNT", new LargerThanCount)
val sink = new TestingAppendSink
@@ -904,7 +904,7 @@ class OverAggregateITCase(mode: StateBackendMode) extends StreamingWithStateTest
.setParallelism(source.parallelism)
.toTable(tEnv, 'a, 'b, 'c, 'rowtime.rowtime)
- tEnv.registerTable("T1", t1)
+ tEnv.createTemporaryView("T1", t1)
val sink = new TestingAppendSink
tEnv.sqlQuery(sqlQuery).toAppendStream[Row].addSink(sink).setParallelism(1)
@@ -961,7 +961,7 @@ class OverAggregateITCase(mode: StateBackendMode) extends StreamingWithStateTest
.setParallelism(source.parallelism)
.toTable(tEnv, 'a, 'b, 'c, 'rowtime.rowtime)
- tEnv.registerTable("T1", t1)
+ tEnv.createTemporaryView("T1", t1)
val sink = new TestingAppendSink
tEnv.sqlQuery(sqlQuery).toAppendStream[Row].addSink(sink).setParallelism(1)
@@ -1030,7 +1030,7 @@ class OverAggregateITCase(mode: StateBackendMode) extends StreamingWithStateTest
.setParallelism(source.parallelism)
.toTable(tEnv, 'a, 'b, 'c, 'rowtime.rowtime)
- tEnv.registerTable("T1", t1)
+ tEnv.createTemporaryView("T1", t1)
val sink = new TestingAppendSink
tEnv.sqlQuery(sqlQuery).toAppendStream[Row].addSink(sink)
@@ -1059,7 +1059,7 @@ class OverAggregateITCase(mode: StateBackendMode) extends StreamingWithStateTest
def testProcTimeDistinctUnboundedPartitionedRowsOver(): Unit = {
val t = failingDataSource(TestData.tupleData5)
.toTable(tEnv, 'a, 'b, 'c, 'd, 'e, 'proctime.proctime)
- tEnv.registerTable("MyTable", t)
+ tEnv.createTemporaryView("MyTable", t)
val sqlQuery = "SELECT a, " +
" COUNT(e) OVER (" +
@@ -1226,7 +1226,7 @@ class OverAggregateITCase(mode: StateBackendMode) extends StreamingWithStateTest
def testProcTimeDistinctBoundedPartitionedRowsOver(): Unit = {
val t = failingDataSource(TestData.tupleData5)
.toTable(tEnv, 'a, 'b, 'c, 'd, 'e, 'proctime.proctime)
- tEnv.registerTable("MyTable", t)
+ tEnv.createTemporaryView("MyTable", t)
val sqlQuery = "SELECT a, " +
" SUM(DISTINCT e) OVER (" +
@@ -1281,7 +1281,7 @@ class OverAggregateITCase(mode: StateBackendMode) extends StreamingWithStateTest
env.setParallelism(1)
val table = env.fromCollection(data).toTable(tEnv, 'a, 'b, 'proctime.proctime)
- tEnv.registerTable("MyTable", table)
+ tEnv.createTemporaryView("MyTable", table)
tEnv.registerFunction("PairCount", new CountPairs)
val sqlQuery = "SELECT a, b, " +
@@ -1320,7 +1320,7 @@ class OverAggregateITCase(mode: StateBackendMode) extends StreamingWithStateTest
env.setParallelism(1)
val rowType = new RowTypeInfo(BigDecimalTypeInfo.of(38, 18))
val t = failingDataSource(data)(rowType).toTable(tEnv, 'd, 'proctime.proctime)
- tEnv.registerTable("T", t)
+ tEnv.createTemporaryView("T", t)
val sqlQuery = "select sum(d) over (ORDER BY proctime rows between unbounded preceding " +
"and current row) from T"
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/RankITCase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/RankITCase.scala
index d04ed000f11..fad0b904725 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/RankITCase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/RankITCase.scala
@@ -50,7 +50,7 @@ class RankITCase(mode: StateBackendMode) extends StreamingWithStateTestBase(mode
("fruit", 5, 22))
val ds = failingDataSource(data).toTable(tEnv, 'category, 'shopId, 'num)
- tEnv.registerTable("T", ds)
+ tEnv.createTemporaryView("T", ds)
val sql =
"""
@@ -91,7 +91,7 @@ class RankITCase(mode: StateBackendMode) extends StreamingWithStateTestBase(mode
("fruit", 5, 22))
val ds = failingDataSource(data).toTable(tEnv, 'category, 'shopId, 'num)
- tEnv.registerTable("T", ds)
+ tEnv.createTemporaryView("T", ds)
val sql =
s"""
@@ -121,7 +121,7 @@ class RankITCase(mode: StateBackendMode) extends StreamingWithStateTestBase(mode
("fruit", 5, 22))
val ds = failingDataSource(data).toTable(tEnv, 'category, 'shopId, 'num)
- tEnv.registerTable("T", ds)
+ tEnv.createTemporaryView("T", ds)
val sql =
"""
@@ -175,7 +175,7 @@ class RankITCase(mode: StateBackendMode) extends StreamingWithStateTestBase(mode
)
val ds = failingDataSource(data).toTable(tEnv, 'category, 'shopId, 'num)
- tEnv.registerTable("T", ds)
+ tEnv.createTemporaryView("T", ds)
val sql =
"""
@@ -234,7 +234,7 @@ class RankITCase(mode: StateBackendMode) extends StreamingWithStateTestBase(mode
)
val ds = failingDataSource(data).toTable(tEnv, 'category, 'shopId, 'price)
- tEnv.registerTable("T", ds)
+ tEnv.createTemporaryView("T", ds)
val sql =
"""
@@ -276,7 +276,7 @@ class RankITCase(mode: StateBackendMode) extends StreamingWithStateTestBase(mode
("fruit", 5, 22))
val ds = failingDataSource(data).toTable(tEnv, 'category, 'shopId, 'num)
- tEnv.registerTable("T", ds)
+ tEnv.createTemporaryView("T", ds)
val sql =
"""
@@ -323,7 +323,7 @@ class RankITCase(mode: StateBackendMode) extends StreamingWithStateTestBase(mode
val ds = env.fromCollection(data)
val t = ds.toTable(tEnv, 'category, 'shopId, 'num)
- tEnv.registerTable("T", t)
+ tEnv.createTemporaryView("T", t)
val subquery =
"""
@@ -369,7 +369,7 @@ class RankITCase(mode: StateBackendMode) extends StreamingWithStateTestBase(mode
("fruit", 5, 22))
val ds = failingDataSource(data).toTable(tEnv, 'category, 'shopId, 'num)
- tEnv.registerTable("T", ds)
+ tEnv.createTemporaryView("T", ds)
val sql =
"""
@@ -411,7 +411,7 @@ class RankITCase(mode: StateBackendMode) extends StreamingWithStateTestBase(mode
("fruit", 5, 22))
val ds = failingDataSource(data).toTable(tEnv, 'category, 'shopId, 'num)
- tEnv.registerTable("T", ds)
+ tEnv.createTemporaryView("T", ds)
val sql =
"""
@@ -448,7 +448,7 @@ class RankITCase(mode: StateBackendMode) extends StreamingWithStateTestBase(mode
("fruit", 5, 22))
val ds = failingDataSource(data).toTable(tEnv, 'category, 'shopId, 'num)
- tEnv.registerTable("T", ds)
+ tEnv.createTemporaryView("T", ds)
val sql =
"""
@@ -496,7 +496,7 @@ class RankITCase(mode: StateBackendMode) extends StreamingWithStateTestBase(mode
)
val ds = failingDataSource(data).toTable(tEnv, 'category, 'shopId, 'sellId)
- tEnv.registerTable("T", ds)
+ tEnv.createTemporaryView("T", ds)
val sql =
"""
@@ -566,7 +566,7 @@ class RankITCase(mode: StateBackendMode) extends StreamingWithStateTestBase(mode
)
val ds = failingDataSource(data).toTable(tEnv, 'category, 'shopId, 'sellId)
- tEnv.registerTable("T", ds)
+ tEnv.createTemporaryView("T", ds)
val sql =
s"""
@@ -611,7 +611,7 @@ class RankITCase(mode: StateBackendMode) extends StreamingWithStateTestBase(mode
env.setParallelism(1)
val ds = failingDataSource(data).toTable(tEnv, 'cate, 'shopId, 'sells)
- tEnv.registerTable("T", ds)
+ tEnv.createTemporaryView("T", ds)
val sql =
"""
@@ -696,7 +696,7 @@ class RankITCase(mode: StateBackendMode) extends StreamingWithStateTestBase(mode
env.setParallelism(1)
val ds = failingDataSource(data).toTable(tEnv, 'cate, 'shopId, 'sells)
- tEnv.registerTable("T", ds)
+ tEnv.createTemporaryView("T", ds)
val sql =
"""
@@ -768,7 +768,7 @@ class RankITCase(mode: StateBackendMode) extends StreamingWithStateTestBase(mode
)
val ds = failingDataSource(data).toTable(tEnv, 'category, 'shopId, 'sellId, 'topSize)
- tEnv.registerTable("T", ds)
+ tEnv.createTemporaryView("T", ds)
val sql =
"""
@@ -826,7 +826,7 @@ class RankITCase(mode: StateBackendMode) extends StreamingWithStateTestBase(mode
env.setParallelism(1)
val ds = failingDataSource(data).toTable(tEnv, 'category, 'shopId, 'num)
- tEnv.registerTable("T", ds)
+ tEnv.createTemporaryView("T", ds)
val sql =
"""
@@ -895,7 +895,7 @@ class RankITCase(mode: StateBackendMode) extends StreamingWithStateTestBase(mode
env.setParallelism(1)
val ds = failingDataSource(data).toTable(tEnv, 'category, 'shopId, 'sellId)
- tEnv.registerTable("T", ds)
+ tEnv.createTemporaryView("T", ds)
val sql =
"""
@@ -976,7 +976,7 @@ class RankITCase(mode: StateBackendMode) extends StreamingWithStateTestBase(mode
env.setParallelism(1)
val ds = failingDataSource(data).toTable(tEnv, 'category, 'shopId, 'sellId)
- tEnv.registerTable("T", ds)
+ tEnv.createTemporaryView("T", ds)
val sql =
"""
@@ -1042,7 +1042,7 @@ class RankITCase(mode: StateBackendMode) extends StreamingWithStateTestBase(mode
env.setParallelism(1)
val ds = failingDataSource(data).toTable(tEnv, 'category, 'shopId, 'num)
- tEnv.registerTable("T", ds)
+ tEnv.createTemporaryView("T", ds)
val sql =
"""
@@ -1092,7 +1092,7 @@ class RankITCase(mode: StateBackendMode) extends StreamingWithStateTestBase(mode
env.setParallelism(1)
val ds = failingDataSource(data).toTable(tEnv, 'category, 'shopId, 'num)
- tEnv.registerTable("T", ds)
+ tEnv.createTemporaryView("T", ds)
val subquery =
s"""
@@ -1103,7 +1103,7 @@ class RankITCase(mode: StateBackendMode) extends StreamingWithStateTestBase(mode
|""".stripMargin
val t1 = tEnv.sqlQuery(subquery)
- tEnv.registerTable("MyView", t1)
+ tEnv.createTemporaryView("MyView", t1)
}
registerView()
@@ -1158,7 +1158,7 @@ class RankITCase(mode: StateBackendMode) extends StreamingWithStateTestBase(mode
env.setParallelism(1)
val ds = failingDataSource(data).toTable(tEnv, 'category, 'shopId, 'num)
- tEnv.registerTable("T", ds)
+ tEnv.createTemporaryView("T", ds)
val subquery =
s"""
@@ -1168,7 +1168,7 @@ class RankITCase(mode: StateBackendMode) extends StreamingWithStateTestBase(mode
|""".stripMargin
val t1 = tEnv.sqlQuery(subquery)
- tEnv.registerTable("MyView", t1)
+ tEnv.createTemporaryView("MyView", t1)
val table1 = tEnv.sqlQuery(
s"""
@@ -1222,7 +1222,7 @@ class RankITCase(mode: StateBackendMode) extends StreamingWithStateTestBase(mode
env.setParallelism(1)
val ds = failingDataSource(data).toTable(tEnv, 'category, 'shopId, 'num)
- tEnv.registerTable("T", ds)
+ tEnv.createTemporaryView("T", ds)
val subquery =
s"""
@@ -1232,7 +1232,7 @@ class RankITCase(mode: StateBackendMode) extends StreamingWithStateTestBase(mode
|""".stripMargin
val t1 = tEnv.sqlQuery(subquery)
- tEnv.registerTable("MyView", t1)
+ tEnv.createTemporaryView("MyView", t1)
val table1 = tEnv.sqlQuery(
s"""
@@ -1330,7 +1330,7 @@ class RankITCase(mode: StateBackendMode) extends StreamingWithStateTestBase(mode
("fruit", "aws", 5, 22))
val ds = failingDataSource(data).toTable(tEnv, 'category, 'seller, 'shopId, 'num)
- tEnv.registerTable("T", ds)
+ tEnv.createTemporaryView("T", ds)
val sql =
"""
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/SemiAntiJoinStreamITCase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/SemiAntiJoinStreamITCase.scala
index 2a8b05a7f0c..16d3ab4b83e 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/SemiAntiJoinStreamITCase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/SemiAntiJoinStreamITCase.scala
@@ -38,8 +38,8 @@ class SemiAntiJoinStreamITCase(state: StateBackendMode) extends StreamingWithSta
.toTable(tEnv, 'a1, 'a2, 'a3)
val tableB = failingDataSource(TestData.tupleData5)
.toTable(tEnv, 'b1, 'b2, 'b3, 'b4, 'b5)
- tEnv.registerTable("A", tableA)
- tEnv.registerTable("B", tableB)
+ tEnv.createTemporaryView("A", tableA)
+ tEnv.createTemporaryView("B", tableB)
}
val data = List(
@@ -68,8 +68,8 @@ class SemiAntiJoinStreamITCase(state: StateBackendMode) extends StreamingWithSta
def testGenericSemiJoin(): Unit = {
val ds1 = failingDataSource(data2).toTable(tEnv, 'a, 'b, 'c)
val ds2 = failingDataSource(data).toTable(tEnv, 'd, 'e, 'f, 'g, 'h)
- tEnv.registerTable("ds1", ds1)
- tEnv.registerTable("ds2", ds2)
+ tEnv.createTemporaryView("ds1", ds1)
+ tEnv.createTemporaryView("ds2", ds2)
val query = "SELECT a, b, c FROM ds1 WHERE a in (SELECT d from ds2 WHERE d < 3)"
val sink = new TestingRetractSink
@@ -108,8 +108,8 @@ class SemiAntiJoinStreamITCase(state: StateBackendMode) extends StreamingWithSta
val ds1 = failingDataSource(leftTable).toTable(tEnv, 'a, 'b)
val ds2 = failingDataSource(rightTable).toTable(tEnv, 'c, 'd)
- tEnv.registerTable("ds1", ds1)
- tEnv.registerTable("ds2", ds2)
+ tEnv.createTemporaryView("ds1", ds1)
+ tEnv.createTemporaryView("ds2", ds2)
val query = "SELECT a FROM ds1 WHERE a in (SELECT sum(c) from ds2 GROUP BY d)"
val sink = new TestingRetractSink
@@ -139,10 +139,10 @@ class SemiAntiJoinStreamITCase(state: StateBackendMode) extends StreamingWithSta
)
val ds1 = failingDataSource(tableData).toTable(tEnv, 'a, 'b)
val ds2 = failingDataSource(tableData).toTable(tEnv, 'c, 'd)
- tEnv.registerTable("ds1", ds1)
- tEnv.registerTable("ds2", ds2)
+ tEnv.createTemporaryView("ds1", ds1)
+ tEnv.createTemporaryView("ds2", ds2)
val ds3 = tEnv.sqlQuery("SELECT sum(a) as a FROM ds1 GROUP BY b")
- tEnv.registerTable("ds3", ds3)
+ tEnv.createTemporaryView("ds3", ds3)
val query = "SELECT a FROM ds3 WHERE a in (SELECT sum(c) from ds2 GROUP BY d)"
val result = tEnv.sqlQuery(query)
@@ -158,8 +158,8 @@ class SemiAntiJoinStreamITCase(state: StateBackendMode) extends StreamingWithSta
def testGenericAntiJoin(): Unit = {
val ds1 = failingDataSource(data).toTable(tEnv, 'a, 'b, 'c, 'd, 'e)
val ds2 = failingDataSource(data2).toTable(tEnv, 'f, 'g, 'h)
- tEnv.registerTable("ds1", ds1)
- tEnv.registerTable("ds2", ds2)
+ tEnv.createTemporaryView("ds1", ds1)
+ tEnv.createTemporaryView("ds2", ds2)
val query = "SELECT c FROM ds1 WHERE NOT EXISTS (SELECT * from ds2 WHERE b = g)"
val sink = new TestingRetractSink
@@ -198,10 +198,10 @@ class SemiAntiJoinStreamITCase(state: StateBackendMode) extends StreamingWithSta
val ds1 = failingDataSource(leftTable).toTable(tEnv, 'a, 'b)
val ds2 = failingDataSource(rightTable).toTable(tEnv, 'c, 'd)
- tEnv.registerTable("ds1", ds1)
- tEnv.registerTable("ds2", ds2)
+ tEnv.createTemporaryView("ds1", ds1)
+ tEnv.createTemporaryView("ds2", ds2)
val ds3 = tEnv.sqlQuery("SELECT SUM(c) as c FROM ds2 GROUP BY d")
- tEnv.registerTable("ds3", ds3)
+ tEnv.createTemporaryView("ds3", ds3)
val query = "SELECT * FROM ds1 WHERE NOT EXISTS (SELECT c from ds3 WHERE a = c)"
val result = tEnv.sqlQuery(query)
@@ -252,12 +252,12 @@ class SemiAntiJoinStreamITCase(state: StateBackendMode) extends StreamingWithSta
val ds1 = failingDataSource(leftTable).toTable(tEnv, 'a, 'b)
val ds2 = failingDataSource(rightTable).toTable(tEnv, 'c, 'd)
- tEnv.registerTable("ds1", ds1)
- tEnv.registerTable("ds2", ds2)
+ tEnv.createTemporaryView("ds1", ds1)
+ tEnv.createTemporaryView("ds2", ds2)
val ds3 = tEnv.sqlQuery("SELECT SUM(c) as c FROM ds2 GROUP BY d")
- tEnv.registerTable("ds3", ds3)
+ tEnv.createTemporaryView("ds3", ds3)
val ds4 = tEnv.sqlQuery("SELECT SUM(a) as a, b FROM ds1 GROUP BY b")
- tEnv.registerTable("ds4", ds4)
+ tEnv.createTemporaryView("ds4", ds4)
val query = "SELECT * FROM ds4 WHERE NOT EXISTS (SELECT c from ds3 WHERE a = c)"
val result = tEnv.sqlQuery(query)
@@ -418,8 +418,8 @@ class SemiAntiJoinStreamITCase(state: StateBackendMode) extends StreamingWithSta
val ds1 = failingDataSource(data1).toTable(tEnv, 'pk, 'a)
val ds2 = failingDataSource(data2).toTable(tEnv, 'pk, 'a)
- tEnv.registerTable("ds1", ds1)
- tEnv.registerTable("ds2", ds2)
+ tEnv.createTemporaryView("ds1", ds1)
+ tEnv.createTemporaryView("ds2", ds2)
val sql =
"""
@@ -448,8 +448,8 @@ class SemiAntiJoinStreamITCase(state: StateBackendMode) extends StreamingWithSta
val ds1 = failingDataSource(data1).toTable(tEnv, 'deptno, 'dname, 'loc)
val ds2 = failingDataSource(data2).toTable(tEnv, 'empno, 'ename, 'deptno)
- tEnv.registerTable("scott_dept", ds1)
- tEnv.registerTable("scott_emp", ds2)
+ tEnv.createTemporaryView("scott_dept", ds1)
+ tEnv.createTemporaryView("scott_emp", ds2)
val sql =
"""
@@ -475,8 +475,8 @@ class SemiAntiJoinStreamITCase(state: StateBackendMode) extends StreamingWithSta
val ds1 = failingDataSource(data1).toTable(tEnv, 'pk, 'a)
val ds2 = failingDataSource(data2).toTable(tEnv, 'pk, 'a)
- tEnv.registerTable("ds1", ds1)
- tEnv.registerTable("ds2", ds2)
+ tEnv.createTemporaryView("ds1", ds1)
+ tEnv.createTemporaryView("ds2", ds2)
val sql =
"""
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/SetOperatorsITCase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/SetOperatorsITCase.scala
index 214d55c8454..655d56e9b3e 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/SetOperatorsITCase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/SetOperatorsITCase.scala
@@ -40,8 +40,8 @@ class SetOperatorsITCase(mode: StateBackendMode) extends StreamingWithStateTestB
.toTable(tEnv, 'a1, 'a2, 'a3)
val tableB = failingDataSource(TestData.tupleData3)
.toTable(tEnv, 'b1, 'b2, 'b3)
- tEnv.registerTable("A", tableA)
- tEnv.registerTable("B", tableB)
+ tEnv.createTemporaryView("A", tableA)
+ tEnv.createTemporaryView("B", tableB)
val sqlQuery = "SELECT a1, a2, a3 from A INTERSECT SELECT b1, b2, b3 from B"
@@ -71,8 +71,8 @@ class SetOperatorsITCase(mode: StateBackendMode) extends StreamingWithStateTestB
val t1 = failingDataSource(data1).toTable(tEnv, 'a1, 'a2, 'a3)
val t2 = failingDataSource(data2).toTable(tEnv, 'b1, 'b2, 'b3)
- tEnv.registerTable("T1", t1)
- tEnv.registerTable("T2", t2)
+ tEnv.createTemporaryView("T1", t1)
+ tEnv.createTemporaryView("T2", t2)
val sqlQuery = "SELECT a3 from T1 EXCEPT SELECT b3 from T2"
@@ -92,8 +92,8 @@ class SetOperatorsITCase(mode: StateBackendMode) extends StreamingWithStateTestB
def testIntersectAll(): Unit = {
val t1 = failingDataSource(Seq(1, 1, 1, 2, 2)).toTable(tEnv, 'c)
val t2 = failingDataSource(Seq(1, 2, 2, 2, 3)).toTable(tEnv, 'c)
- tEnv.registerTable("T1", t1)
- tEnv.registerTable("T2", t2)
+ tEnv.createTemporaryView("T1", t1)
+ tEnv.createTemporaryView("T2", t2)
val sqlQuery = "SELECT c FROM T1 INTERSECT ALL SELECT c FROM T2"
@@ -107,9 +107,9 @@ class SetOperatorsITCase(mode: StateBackendMode) extends StreamingWithStateTestB
@Test
def testMinusAll(): Unit = {
val tableA = failingDataSource(TestData.smallTupleData3).toTable(tEnv, 'a, 'b, 'c)
- tEnv.registerTable("tableA", tableA)
+ tEnv.createTemporaryView("tableA", tableA)
val tableB = failingDataSource(Seq((1, 1L, "Hi"), (1, 1L, "Hi"))).toTable(tEnv, 'a, 'b, 'c)
- tEnv.registerTable("tableB", tableB)
+ tEnv.createTemporaryView("tableB", tableB)
val t1 = "SELECT * FROM tableA"
val t2 = "SELECT * FROM tableB"
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/SortITCase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/SortITCase.scala
index 7ced31ebd79..94d3fa2378e 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/SortITCase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/SortITCase.scala
@@ -46,7 +46,7 @@ class SortITCase(mode: StateBackendMode) extends StreamingWithStateTestBase(mode
data.+=(("5", "1"))
val da = env.fromCollection(data).toTable(tEnv, 'a1, 'a2)
- tEnv.registerTable("a", da)
+ tEnv.createTemporaryView("a", da)
thrown.expect(classOf[TableException])
thrown.expectMessage("Sort on a non-time-attribute field is not supported.")
@@ -63,7 +63,7 @@ class SortITCase(mode: StateBackendMode) extends StreamingWithStateTestBase(mode
data.+=(("5", "1"))
val da = failingDataSource(data).toTable(tEnv, 'a1, 'a2)
- tEnv.registerTable("a", da)
+ tEnv.createTemporaryView("a", da)
val sink = new TestingRetractSink
tEnv.getConfig
@@ -88,7 +88,7 @@ class SortITCase(mode: StateBackendMode) extends StreamingWithStateTestBase(mode
data.+=(("5", "1"))
val da = failingDataSource(data).toTable(tEnv, 'a1, 'a2)
- tEnv.registerTable("a", da)
+ tEnv.createTemporaryView("a", da)
val sink = new TestingRetractSink
tEnv.getConfig.set(
@@ -114,7 +114,7 @@ class SortITCase(mode: StateBackendMode) extends StreamingWithStateTestBase(mode
data.+=(("1", "2"))
val da = failingDataSource(data).toTable(tEnv, 'a1, 'a2)
- tEnv.registerTable("a", da)
+ tEnv.createTemporaryView("a", da)
val sink = new TestingRetractSink
tEnv.getConfig
@@ -139,7 +139,7 @@ class SortITCase(mode: StateBackendMode) extends StreamingWithStateTestBase(mode
data.+=(("2", "2"))
val da = failingDataSource(data).toTable(tEnv, 'a1, 'a2)
- tEnv.registerTable("a", da)
+ tEnv.createTemporaryView("a", da)
val sink = new TestingRetractSink
tEnv.getConfig
@@ -170,7 +170,7 @@ class SortITCase(mode: StateBackendMode) extends StreamingWithStateTestBase(mode
data.+=(("6", "2"))
val da = failingDataSource(data).toTable(tEnv, 'a1, 'a2)
- tEnv.registerTable("a", da)
+ tEnv.createTemporaryView("a", da)
val sink = new TestingRetractSink
tEnv.getConfig
@@ -199,7 +199,7 @@ class SortITCase(mode: StateBackendMode) extends StreamingWithStateTestBase(mode
data.+=((4, 5))
val da = failingDataSource(data).toTable(tEnv, 'a1)
- tEnv.registerTable("a", da)
+ tEnv.createTemporaryView("a", da)
val sink = new TestingRetractSink
tEnv.getConfig
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/SortLimitITCase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/SortLimitITCase.scala
index f175d43e90f..e914094e615 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/SortLimitITCase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/SortLimitITCase.scala
@@ -43,7 +43,7 @@ class SortLimitITCase(mode: StateBackendMode) extends StreamingWithStateTestBase
("fruit", 5, 22))
val ds = failingDataSource(data).toTable(tEnv, 'category, 'shopId, 'num)
- tEnv.registerTable("T", ds)
+ tEnv.createTemporaryView("T", ds)
val sql = "SELECT * FROM T ORDER BY num DESC LIMIT 2"
@@ -60,7 +60,7 @@ class SortLimitITCase(mode: StateBackendMode) extends StreamingWithStateTestBase
val data = List((1, 1), (1, 2), (1, 3), (2, 2), (2, 3), (2, 4), (3, 3), (3, 4), (3, 5))
val ds = failingDataSource(data).toTable(tEnv, 'a, 'b)
- tEnv.registerTable("T", ds)
+ tEnv.createTemporaryView("T", ds)
// We use max here to ensure the usage of update rank
val sql = "SELECT a, max(b) FROM T GROUP BY a ORDER BY a LIMIT 2"
@@ -78,7 +78,7 @@ class SortLimitITCase(mode: StateBackendMode) extends StreamingWithStateTestBase
val data = List((1, 1), (1, 2), (1, 3), (2, 2), (2, 3), (2, 4), (3, 3), (3, 4), (3, 5))
val ds = failingDataSource(data).toTable(tEnv, 'a, 'b)
- tEnv.registerTable("T", ds)
+ tEnv.createTemporaryView("T", ds)
// We use max here to ensure the usage of update rank
val sql = "SELECT a, max(b) FROM T GROUP BY a ORDER BY a LIMIT 2 OFFSET 1"
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/SplitAggregateITCase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/SplitAggregateITCase.scala
index 8e723ea3831..3e6b11bfd1f 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/SplitAggregateITCase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/SplitAggregateITCase.scala
@@ -82,7 +82,7 @@ class SplitAggregateITCase(
)
val t = failingDataSource(data).toTable(tEnv, 'a, 'b, 'c)
- tEnv.registerTable("T", t)
+ tEnv.createTemporaryView("T", t)
}
@Test
@@ -409,7 +409,7 @@ class SplitAggregateITCase(
def testUvWithRetraction(): Unit = {
val data = (0 until 1000).map(i => (s"${i % 10}", s"${i % 100}", s"$i")).toList
val t = failingDataSource(data).toTable(tEnv, 'a, 'b, 'c)
- tEnv.registerTable("src", t)
+ tEnv.createTemporaryView("src", t)
val sql =
s"""
@@ -440,7 +440,7 @@ class SplitAggregateITCase(
val data = (0 until 100).map(i => ("1", "1", s"${i % 50}", "1")).toList
// use BinaryRowData source here for StringData reuse
val t = failingBinaryRowSource(data).toTable(tEnv, 'a, 'b, 'c, 'd)
- tEnv.registerTable("src", t)
+ tEnv.createTemporaryView("src", t)
val sql =
s"""
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/TemporalSortITCase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/TemporalSortITCase.scala
index 5244a60b32c..262afbcae77 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/TemporalSortITCase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/TemporalSortITCase.scala
@@ -59,7 +59,7 @@ class TemporalSortITCase(mode: StateBackendMode) extends StreamingWithStateTestB
.assignTimestampsAndWatermarks(
new TimestampAndWatermarkWithOffset[(Long, Long, String, Int)](10L))
.toTable(tEnv, 'rowtime.rowtime, 'key, 'str, 'int)
- tEnv.registerTable("T", t)
+ tEnv.createTemporaryView("T", t)
val sqlQuery = "SELECT key, str, `int` FROM T ORDER BY rowtime"
@@ -105,7 +105,7 @@ class TemporalSortITCase(mode: StateBackendMode) extends StreamingWithStateTestB
new TimestampAndWatermarkWithOffset[(Long, Long, String, Int)](10L))
.setParallelism(env.getParallelism)
.toTable(tEnv, 'rowtime.rowtime, 'key, 'str, 'int)
- tEnv.registerTable("T", t)
+ tEnv.createTemporaryView("T", t)
val sqlQuery = "SELECT key, str, `int` FROM T ORDER BY rowtime"
@@ -244,7 +244,7 @@ class TemporalSortITCase(mode: StateBackendMode) extends StreamingWithStateTestB
def testProcTimeOrderBy(): Unit = {
val t = failingDataSource(TestData.tupleData3)
.toTable(tEnv, 'a, 'b, 'c, 'proctime.proctime)
- tEnv.registerTable("T", t)
+ tEnv.createTemporaryView("T", t)
val sql = "SELECT a, b, c FROM T ORDER BY proctime"
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/TemporalTableFunctionJoinITCase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/TemporalTableFunctionJoinITCase.scala
index 1dd5a9a04c1..ff325d35328 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/TemporalTableFunctionJoinITCase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/TemporalTableFunctionJoinITCase.scala
@@ -85,8 +85,8 @@ class TemporalTableFunctionJoinITCase(state: StateBackendMode)
.fromCollection(ratesHistoryData)
.toTable(tEnv, 'currency, 'rate, 'proctime.proctime)
- tEnv.registerTable("Orders", orders)
- tEnv.registerTable("RatesHistory", ratesHistory)
+ tEnv.createTemporaryView("Orders", orders)
+ tEnv.createTemporaryView("RatesHistory", ratesHistory)
tEnv.createTemporarySystemFunction(
"Rates",
ratesHistory.createTemporalTableFunction($"proctime", $"currency"))
@@ -152,14 +152,14 @@ class TemporalTableFunctionJoinITCase(state: StateBackendMode)
.fromCollection(ratesHistoryData)
.toTable(tEnv, 'currency, 'rate, 'proctime.proctime)
- tEnv.registerTable("Orders1", orders1)
- tEnv.registerTable("Orders2", orders2)
- tEnv.registerTable("RatesHistory", ratesHistory)
+ tEnv.createTemporaryView("Orders1", orders1)
+ tEnv.createTemporaryView("Orders2", orders2)
+ tEnv.createTemporaryView("RatesHistory", ratesHistory)
tEnv.registerFunction(
"Rates",
ratesHistory.createTemporalTableFunction($"proctime", $"currency"))
- tEnv.registerTable(
+ tEnv.createTemporaryView(
"Orders",
tEnv.sqlQuery("SELECT * FROM Orders1 UNION ALL SELECT * FROM Orders2"))
val result = tEnv.sqlQuery(sqlQuery).toAppendStream[Row]
@@ -210,9 +210,9 @@ class TemporalTableFunctionJoinITCase(state: StateBackendMode)
.assignTimestampsAndWatermarks(new TimestampExtractor[(String, Long, Timestamp)]())
.toTable(tEnv, 'currency, 'rate, 'rowtime.rowtime)
- tEnv.registerTable("Orders", orders)
- tEnv.registerTable("RatesHistory", ratesHistory)
- tEnv.registerTable(
+ tEnv.createTemporaryView("Orders", orders)
+ tEnv.createTemporaryView("RatesHistory", ratesHistory)
+ tEnv.createTemporaryView(
"FilteredRatesHistory",
tEnv.sqlQuery("SELECT * FROM RatesHistory WHERE rate > 110"))
tEnv.createTemporarySystemFunction(
@@ -220,7 +220,7 @@ class TemporalTableFunctionJoinITCase(state: StateBackendMode)
tEnv
.from("FilteredRatesHistory")
.createTemporalTableFunction($"rowtime", $"currency"))
- tEnv.registerTable("TemporalJoinResult", tEnv.sqlQuery(sqlQuery))
+ tEnv.createTemporaryView("TemporalJoinResult", tEnv.sqlQuery(sqlQuery))
// Scan from registered table to test for interplay between
// LogicalCorrelateToTemporalTableJoinRule and TableScanRule
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/UnnestITCase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/UnnestITCase.scala
index 81214a4cf2e..c5adba19b1f 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/UnnestITCase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/UnnestITCase.scala
@@ -47,7 +47,7 @@ class UnnestITCase(mode: StateBackendMode) extends StreamingWithStateTestBase(mo
(3, Array(18, 42), Array(Array(1), Array(45)))
)
val t = env.fromCollection(data).toTable(tEnv, 'a, 'b, 'c)
- tEnv.registerTable("T", t)
+ tEnv.createTemporaryView("T", t)
val sqlQuery = "SELECT a, b, s FROM T, UNNEST(T.b) AS A (s)"
val result = tEnv.sqlQuery(sqlQuery).toAppendStream[Row]
@@ -73,7 +73,7 @@ class UnnestITCase(mode: StateBackendMode) extends StreamingWithStateTestBase(mo
(3, Array(18, 42), Array(Array(1), Array(45)))
)
val t = env.fromCollection(data).toTable(tEnv, 'a, 'b, 'c)
- tEnv.registerTable("T", t)
+ tEnv.createTemporaryView("T", t)
val sqlQuery = "SELECT a, s FROM T, UNNEST(T.c) AS A (s)"
val result = tEnv.sqlQuery(sqlQuery).toAppendStream[Row]
@@ -93,7 +93,7 @@ class UnnestITCase(mode: StateBackendMode) extends StreamingWithStateTestBase(mo
(3, Array((18, "42.6")))
)
val t = env.fromCollection(data).toTable(tEnv, 'a, 'b)
- tEnv.registerTable("T", t)
+ tEnv.createTemporaryView("T", t)
val sqlQuery = "SELECT a, b, s, t FROM T, UNNEST(T.b) AS A (s, t) WHERE s > 13"
val result = tEnv.sqlQuery(sqlQuery).toAppendStream[Row]
@@ -114,7 +114,7 @@ class UnnestITCase(mode: StateBackendMode) extends StreamingWithStateTestBase(mo
(4, 3, (14, "45.2136")),
(5, 3, (18, "42.6")))
val t = env.fromCollection(data).toTable(tEnv, 'a, 'b, 'c)
- tEnv.registerTable("T", t)
+ tEnv.createTemporaryView("T", t)
val sqlQuery =
"""
@@ -146,7 +146,7 @@ class UnnestITCase(mode: StateBackendMode) extends StreamingWithStateTestBase(mo
)
val t = env.fromCollection(data).toTable(tEnv, 'a, 'b, 'c)
- tEnv.registerTable("T", t)
+ tEnv.createTemporaryView("T", t)
val sqlQuery =
"""
@@ -174,7 +174,7 @@ class UnnestITCase(mode: StateBackendMode) extends StreamingWithStateTestBase(mo
val stream = failingDataSource(data)
.assignTimestampsAndWatermarks(new TimestampAndWatermarkWithOffset[(Long, Int, String)](0L))
val t = stream.toTable(tEnv, 'b, 'a, 'c, 'rowtime.rowtime)
- tEnv.registerTable("T", t)
+ tEnv.createTemporaryView("T", t)
val sqlQuery =
"""
@@ -206,7 +206,7 @@ class UnnestITCase(mode: StateBackendMode) extends StreamingWithStateTestBase(mo
)
val t = env.fromCollection(data).toTable(tEnv, 'a, 'b, 'c)
- tEnv.registerTable("T", t)
+ tEnv.createTemporaryView("T", t)
val sqlQuery = "SELECT a, s FROM T, UNNEST(T.c) as A (s)"
val result = tEnv.sqlQuery(sqlQuery).toAppendStream[Row]
@@ -251,7 +251,7 @@ class UnnestITCase(mode: StateBackendMode) extends StreamingWithStateTestBase(mo
Array[TypeInformation[_]](Types.INT, Types.LONG, Types.MAP(Types.STRING, Types.STRING))
)
val t = env.fromCollection(data).toTable(tEnv, 'a, 'b, 'c)
- tEnv.registerTable("T", t)
+ tEnv.createTemporaryView("T", t)
val sqlQuery = "SELECT a, b, v FROM T CROSS JOIN UNNEST(c) as f (k, v)"
val result = tEnv.sqlQuery(sqlQuery)
@@ -274,7 +274,7 @@ class UnnestITCase(mode: StateBackendMode) extends StreamingWithStateTestBase(mo
)
val t = env.fromCollection(data).toTable(tEnv, 'a, 'b)
- tEnv.registerTable("T", t)
+ tEnv.createTemporaryView("T", t)
val sqlQuery = "SELECT a, b, x, y " +
"FROM " +
@@ -301,7 +301,7 @@ class UnnestITCase(mode: StateBackendMode) extends StreamingWithStateTestBase(mo
(3, Array((18, "42.6")))
)
val t = env.fromCollection(data).toTable(tEnv, 'a, 'b)
- tEnv.registerTable("T", t)
+ tEnv.createTemporaryView("T", t)
val sqlQuery = "SELECT a, b, A._1, A._2 FROM T, UNNEST(T.b) AS A where A._1 > 13"
val result = tEnv.sqlQuery(sqlQuery).toAppendStream[Row]
@@ -321,7 +321,7 @@ class UnnestITCase(mode: StateBackendMode) extends StreamingWithStateTestBase(mo
(3, Array((18, "42.6")))
)
val t = env.fromCollection(data).toTable(tEnv, 'a, 'b)
- tEnv.registerTable("MyTable", t)
+ tEnv.createTemporaryView("MyTable", t)
val sqlQuery =
"""
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/table/LegacyTableSinkITCase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/table/LegacyTableSinkITCase.scala
index 8e6c3420d7f..3164471cb40 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/table/LegacyTableSinkITCase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/table/LegacyTableSinkITCase.scala
@@ -138,7 +138,7 @@ class LegacyTableSinkITCase extends AbstractTestBase {
val t = env
.fromCollection(smallTupleData3)
.toTable(tEnv, 'id, 'num, 'text)
- tEnv.registerTable("src", t)
+ tEnv.createTemporaryView("src", t)
val sink = new TestingAppendTableSink()
tEnv
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/table/TableToDataStreamITCase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/table/TableToDataStreamITCase.scala
index a4dce7d30b0..b2a786bae04 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/table/TableToDataStreamITCase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/table/TableToDataStreamITCase.scala
@@ -148,7 +148,7 @@ final class TableToDataStreamITCase extends StreamingTestBase {
// second to millisecond
.assignAscendingTimestamps(_._1 * 1000L)
val table = ds1.toTable(tEnv, 'ts, 'a, 'rowtime.rowtime)
- tEnv.registerTable("t1", table)
+ tEnv.createTemporaryView("t1", table)
val ds2 = tEnv
.sqlQuery(
@@ -293,7 +293,7 @@ final class TableToDataStreamITCase extends StreamingTestBase {
// second to millisecond
.assignAscendingTimestamps(_._1 * 1000L)
val table = ds1.toTable(tEnv, 'ts, 'a, 'proctime.proctime())
- tEnv.registerTable("t1", table)
+ tEnv.createTemporaryView("t1", table)
val ds2 = tEnv
.sqlQuery(
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/utils/TableTestBase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/utils/TableTestBase.scala
index 0615b66b671..e9d1f0295a3 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/utils/TableTestBase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/utils/TableTestBase.scala
@@ -1185,7 +1185,7 @@ abstract class TableTestUtil(
name)
val operation = new RichTableSourceQueryOperation(identifier, tableSource, statistic)
val table = testingTableEnv.createTable(operation)
- testingTableEnv.registerTable(name, table)
+ testingTableEnv.createTemporaryView(name, table)
testingTableEnv.from(name)
}
@@ -1315,7 +1315,7 @@ case class StreamTableTestUtil(
expr
)
val queryOperation = new PlannerQueryOperation(watermarkAssigner)
- testingTableEnv.registerTable(tableName, testingTableEnv.createTable(queryOperation))
+ testingTableEnv.createTemporaryView(tableName, testingTableEnv.createTable(queryOperation))
}
def buildStreamProgram(firstProgramNameToRemove: String): Unit = {
@@ -1722,7 +1722,7 @@ object TableTestUtil {
statistic.getOrElse(FlinkStatistic.UNKNOWN)
)
val table = tEnv.asInstanceOf[TableEnvironmentImpl].createTable(dataStreamQueryOperation)
- tEnv.registerTable(name, table)
+ tEnv.createTemporaryView(name, table)
}
def readFromResource(path: String): String = {
[flink] 02/02: [FLINK-31598][table] Quotes the UUID table name to fix the failed tests
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 2a3652d856347ddc333a476aacb0e750f6502a7b
Author: Jark Wu <ja...@apache.org>
AuthorDate: Thu Mar 23 23:46:45 2023 +0800
[FLINK-31598][table] Quotes the UUID table name to fix the failed tests
---
.../table/planner/runtime/stream/sql/SemiAntiJoinStreamITCase.scala | 4 ++--
.../scala/org/apache/flink/table/planner/utils/TableTestBase.scala | 5 +++--
2 files changed, 5 insertions(+), 4 deletions(-)
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/SemiAntiJoinStreamITCase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/SemiAntiJoinStreamITCase.scala
index 16d3ab4b83e..057480d3ff1 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/SemiAntiJoinStreamITCase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/SemiAntiJoinStreamITCase.scala
@@ -504,8 +504,8 @@ class SemiAntiJoinStreamITCase(state: StateBackendMode) extends StreamingWithSta
val rTable = List(0, 1)
val ds1 = failingDataSource(lTable).toTable(tEnv, 'a, 'b, 'c)
val ds2 = failingDataSource(rTable).toTable(tEnv, 'd)
- tEnv.registerTable("l", ds1)
- tEnv.registerTable("r", ds2)
+ tEnv.createTemporaryView("l", ds1)
+ tEnv.createTemporaryView("r", ds2)
val query =
"SELECT a + 10, c FROM l WHERE b > 10 AND NOT (c like 'abc' OR NOT EXISTS (SELECT d FROM r))"
val result = tEnv.sqlQuery(query)
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/utils/TableTestBase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/utils/TableTestBase.scala
index e9d1f0295a3..cde16f76502 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/utils/TableTestBase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/utils/TableTestBase.scala
@@ -1177,7 +1177,7 @@ abstract class TableTestUtil(
tableSource: TableSource[_],
statistic: FlinkStatistic): Table = {
// TODO RichTableSourceQueryOperation should be deleted and use registerTableSourceInternal
- // method instead of registerTable method here after unique key in TableSchema is ready
+ // method instead of createTemporaryView method here after unique key in TableSchema is ready
// and setting catalog statistic to TableSourceTable in DatabaseCalciteSchema is ready
val identifier = ObjectIdentifier.of(
testingTableEnv.getCurrentCatalog,
@@ -1722,7 +1722,8 @@ object TableTestUtil {
statistic.getOrElse(FlinkStatistic.UNKNOWN)
)
val table = tEnv.asInstanceOf[TableEnvironmentImpl].createTable(dataStreamQueryOperation)
- tEnv.createTemporaryView(name, table)
+ // the table name is UUID generated which should be quoted to parse safely
+ tEnv.createTemporaryView(s"`$name`", table)
}
def readFromResource(path: String): String = {