This is an automated email from the ASF dual-hosted git repository.
snuyanzin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new f5846010d69 [FLINK-40096][table] Migrate `EXPLAIN` `CTAS`/`RTAS`
Explain tests to Java
f5846010d69 is described below
commit f5846010d6908e5407d78dadf597f1989f08a29c
Author: Ramin Gharib <[email protected]>
AuthorDate: Thu Jul 9 06:58:13 2026 +0200
[FLINK-40096][table] Migrate `EXPLAIN` `CTAS`/`RTAS` Explain tests to Java
---
.../table/planner/plan/stream/sql/ExplainTest.java | 88 +++++++++++++++++-
.../src/test/resources/explain/testExplainCtas.out | 6 +-
...ExplainCtasWithColumnsInCreateAndQueryParts.out | 4 +-
.../planner/plan/stream/sql/TableSinkTest.scala | 101 +--------------------
4 files changed, 89 insertions(+), 110 deletions(-)
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/ExplainTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/ExplainTest.java
index e3b35dd9f6b..e6b4024fae9 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/ExplainTest.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/ExplainTest.java
@@ -174,15 +174,95 @@ public class ExplainTest extends TableTestBase {
+ " MyTable");
}
+ @Test
+ void testExplainCreateTableAsSelect() {
+ verifyExplain(
+ "CREATE TABLE MyCtasTable\n"
+ + " WITH (\n"
+ + " 'connector' = 'values'\n"
+ + ") AS\n"
+ + " SELECT\n"
+ + " `a`,\n"
+ + " `b`\n"
+ + " FROM\n"
+ + " MyTable",
+ "testExplainCtas");
+ }
+
+ @Test
+ void testExplainReplaceTableAsSelect() {
+ // Produces the same plan as CREATE TABLE AS SELECT.
+ verifyExplain(
+ "REPLACE TABLE MyCtasTable\n"
+ + " WITH (\n"
+ + " 'connector' = 'values'\n"
+ + ") AS\n"
+ + " SELECT\n"
+ + " `a`,\n"
+ + " `b`\n"
+ + " FROM\n"
+ + " MyTable",
+ "testExplainCtas");
+ }
+
+ @Test
+ void testExplainCreateTableAsSelectWithColumnsInCreateAndQueryParts() {
+ verifyExplain(
+ "CREATE TABLE MyCtasTable(\n"
+ + " `votes` INT,\n"
+ + " `votes_2x` AS `b` * 2,\n"
+ + " `metadata_col` BIGINT METADATA,\n"
+ + " `virtual_col` STRING METADATA VIRTUAL\n"
+ + ")\n"
+ + " WITH (\n"
+ + " 'connector' = 'values',\n"
+ + " 'readable-metadata' = 'metadata_col:BIGINT,
virtual_col:STRING',\n"
+ + " 'writable-metadata' = 'metadata_col:BIGINT'\n"
+ + ") AS\n"
+ + " SELECT\n"
+ + " `a`,\n"
+ + " `b`\n"
+ + " FROM\n"
+ + " MyTable",
+ "testExplainCtasWithColumnsInCreateAndQueryParts");
+ }
+
+ @Test
+ void testExplainReplaceTableAsSelectWithColumnsInCreateAndQueryParts() {
+ // Produces the same plan as CREATE TABLE AS SELECT.
+ verifyExplain(
+ "REPLACE TABLE MyCtasTable(\n"
+ + " `votes` INT,\n"
+ + " `votes_2x` AS `b` * 2,\n"
+ + " `metadata_col` BIGINT METADATA,\n"
+ + " `virtual_col` STRING METADATA VIRTUAL\n"
+ + ")\n"
+ + " WITH (\n"
+ + " 'connector' = 'values',\n"
+ + " 'readable-metadata' = 'metadata_col:BIGINT,
virtual_col:STRING',\n"
+ + " 'writable-metadata' = 'metadata_col:BIGINT'\n"
+ + ") AS\n"
+ + " SELECT\n"
+ + " `a`,\n"
+ + " `b`\n"
+ + " FROM\n"
+ + " MyTable",
+ "testExplainCtasWithColumnsInCreateAndQueryParts");
+ }
+
private void verifyExplain(final String statement) {
- final String actual = util.getTableEnv().explainSql(statement);
final String displayName = this.testInfo.getDisplayName();
- final String fileName = displayName.substring(0, displayName.length()
- 2) + ".out";
+ verifyExplain(statement, displayName.substring(0, displayName.length()
- 2));
+ }
+
+ private void verifyExplain(final String statement, final String fileName) {
+ final String actual = util.getTableEnv().explainSql(statement);
+ final String fullFileName = fileName + ".out";
if (REGENERATE_FILES) {
- writeToResource(fileName, actual);
+ writeToResource(fullFileName, actual);
return;
}
- final String expected = TableTestUtil.readFromResource("/explain/" +
fileName);
+ final String expected = TableTestUtil.readFromResource("/explain/" +
fullFileName);
assertThat(TableTestUtil.replaceStageId(actual))
.isEqualTo(TableTestUtil.replaceStageId(expected));
}
diff --git
a/flink-table/flink-table-planner/src/test/resources/explain/testExplainCtas.out
b/flink-table/flink-table-planner/src/test/resources/explain/testExplainCtas.out
index 9ebb7197beb..a0dbe7abfd7 100644
---
a/flink-table/flink-table-planner/src/test/resources/explain/testExplainCtas.out
+++
b/flink-table/flink-table-planner/src/test/resources/explain/testExplainCtas.out
@@ -5,10 +5,8 @@
LogicalSink(table=[default_catalog.default_database.MyCtasTable], fields=[a, b])
== Optimized Physical Plan ==
Sink(table=[default_catalog.default_database.MyCtasTable], fields=[a, b])
-+- Calc(select=[a, b])
- +- DataStreamScan(table=[[default_catalog, default_database, MyTable]],
fields=[a, b, c])
++- TableSourceScan(table=[[default_catalog, default_database, MyTable,
project=[a, b], metadata=[]]], fields=[a, b])
== Optimized Execution Plan ==
Sink(table=[default_catalog.default_database.MyCtasTable], fields=[a, b])
-+- Calc(select=[a, b])
- +- DataStreamScan(table=[[default_catalog, default_database, MyTable]],
fields=[a, b, c])
++- TableSourceScan(table=[[default_catalog, default_database, MyTable,
project=[a, b], metadata=[]]], fields=[a, b])
diff --git
a/flink-table/flink-table-planner/src/test/resources/explain/testExplainCtasWithColumnsInCreateAndQueryParts.out
b/flink-table/flink-table-planner/src/test/resources/explain/testExplainCtasWithColumnsInCreateAndQueryParts.out
index eb7db8e3862..c738e804db9 100644
---
a/flink-table/flink-table-planner/src/test/resources/explain/testExplainCtasWithColumnsInCreateAndQueryParts.out
+++
b/flink-table/flink-table-planner/src/test/resources/explain/testExplainCtasWithColumnsInCreateAndQueryParts.out
@@ -6,9 +6,9 @@
LogicalSink(table=[default_catalog.default_database.MyCtasTable], fields=[votes,
== Optimized Physical Plan ==
Sink(table=[default_catalog.default_database.MyCtasTable], fields=[votes, a,
b, metadata_col])
+- Calc(select=[null:INTEGER AS votes, a, b, null:BIGINT AS metadata_col])
- +- DataStreamScan(table=[[default_catalog, default_database, MyTable]],
fields=[a, b, c])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable,
project=[a, b], metadata=[]]], fields=[a, b])
== Optimized Execution Plan ==
Sink(table=[default_catalog.default_database.MyCtasTable], fields=[votes, a,
b, metadata_col])
+- Calc(select=[null:INTEGER AS votes, a, b, null:BIGINT AS metadata_col])
- +- DataStreamScan(table=[[default_catalog, default_database, MyTable]],
fields=[a, b, c])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable,
project=[a, b], metadata=[]]], fields=[a, b])
diff --git
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/TableSinkTest.scala
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/TableSinkTest.scala
index 5201b1f20bc..31f0d6ca6fa 100644
---
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/TableSinkTest.scala
+++
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/TableSinkTest.scala
@@ -26,11 +26,10 @@ import org.apache.flink.table.connector.ChangelogMode
import org.apache.flink.table.connector.source.{DynamicTableSource,
ScanTableSource}
import org.apache.flink.table.data.RowData
import org.apache.flink.table.factories.{DynamicTableFactory,
DynamicTableSourceFactory}
-import org.apache.flink.table.planner.utils.{TableTestBase, TableTestUtil,
TestingTableEnvironment}
+import org.apache.flink.table.planner.utils.{TableTestBase,
TestingTableEnvironment}
import org.assertj.core.api.Assertions
import org.assertj.core.api.Assertions.assertThatThrownBy
-import org.junit.jupiter.api.Assertions.assertEquals
import org.junit.jupiter.api.Test
import java.util
@@ -847,104 +846,6 @@ class TableSinkTest extends TableTestBase {
"Table 'default_catalog.default_database.sink' is a bucketed table and
it supports [HASH, UNKNOWN], but algorithm RANGE was requested.")
}
- @Test
- def testExplainCreateTableAsSelect(): Unit = {
- val actual = util.tableEnv.explainSql("""
- |CREATE TABLE MyCtasTable
- | WITH (
- | 'connector' = 'values'
- |) AS
- | SELECT
- | `a`,
- | `b`
- | FROM
- | MyTable
- |""".stripMargin)
-
- val expected =
TableTestUtil.readFromResource("/explain/testExplainCtas.out")
-
- assertEquals(TableTestUtil.replaceStageId(expected),
TableTestUtil.replaceStageId(actual))
- }
-
- @Test
- def testExplainReplaceTableAsSelect(): Unit = {
- val actual = util.tableEnv.explainSql("""
- |REPLACE TABLE MyCtasTable
- | WITH (
- | 'connector' = 'values'
- |) AS
- | SELECT
- | `a`,
- | `b`
- | FROM
- | MyTable
- |""".stripMargin)
-
- // Same as CTAS
- val expected =
TableTestUtil.readFromResource("/explain/testExplainCtas.out")
-
- assertEquals(TableTestUtil.replaceStageId(expected),
TableTestUtil.replaceStageId(actual))
- }
-
- @Test
- def testExplainCreateTableAsSelectWithColumnsInCreateAndQueryParts(): Unit =
{
- val actual =
- util.tableEnv.explainSql(
- """
- |CREATE TABLE MyCtasTable(
- | `votes` INT,
- | `votes_2x` AS `b` * 2,
- | `metadata_col` BIGINT METADATA,
- | `virtual_col` STRING METADATA VIRTUAL
- |)
- | WITH (
- | 'connector' = 'values',
- | 'readable-metadata' = 'metadata_col:BIGINT, virtual_col:STRING',
- | 'writable-metadata' = 'metadata_col:BIGINT'
- |) AS
- | SELECT
- | `a`,
- | `b`
- | FROM
- | MyTable
- |""".stripMargin)
-
- val expected =
-
TableTestUtil.readFromResource("/explain/testExplainCtasWithColumnsInCreateAndQueryParts.out")
-
- assertEquals(TableTestUtil.replaceStageId(expected),
TableTestUtil.replaceStageId(actual))
- }
-
- @Test
- def testExplainReplaceTableAsSelectWithColumnsInCreateAndQueryParts(): Unit
= {
- val actual =
- util.tableEnv.explainSql(
- """
- |REPLACE TABLE MyCtasTable(
- | `votes` INT,
- | `votes_2x` AS `b` * 2,
- | `metadata_col` BIGINT METADATA,
- | `virtual_col` STRING METADATA VIRTUAL
- |)
- | WITH (
- | 'connector' = 'values',
- | 'readable-metadata' = 'metadata_col:BIGINT, virtual_col:STRING',
- | 'writable-metadata' = 'metadata_col:BIGINT'
- |) AS
- | SELECT
- | `a`,
- | `b`
- | FROM
- | MyTable
- |""".stripMargin)
-
- // Same as CTAS
- val expected =
-
TableTestUtil.readFromResource("/explain/testExplainCtasWithColumnsInCreateAndQueryParts.out")
-
- assertEquals(TableTestUtil.replaceStageId(expected),
TableTestUtil.replaceStageId(actual))
- }
-
@Test
def testInsertOnConflictDoDeduplicate(): Unit = {
util.addTable(s"""
