You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by sn...@apache.org on 2025/11/12 17:09:34 UTC
(flink) 01/02: [FLINK-38674][table] Support schema definition for Materialized tables while `CREATE` operation
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
commit f8e29badfe6531154373db2104707f0296e60880
Author: Sergey Nuyanzin <sn...@gmail.com>
AuthorDate: Sun Nov 9 10:38:16 2025 +0100
[FLINK-38674][table] Support schema definition for Materialized tables while `CREATE` operation
This closes #27222.
---
.../dev/table/materialized-table/statements.md | 48 +++-
.../dev/table/materialized-table/statements.md | 46 +++-
.../src/main/codegen/includes/parserImpls.ftl | 20 +-
.../sql/parser/ddl/SqlCreateMaterializedTable.java | 69 +++--
.../MaterializedTableStatementParserTest.java | 143 ++++++++++-
.../AbstractCreateMaterializedTableConverter.java | 221 ++++++++++++++++
.../SqlCreateMaterializedTableConverter.java | 267 ++++++--------------
.../table/AbstractCreateTableConverter.java | 2 +-
.../converters/table/MergeTableAsUtil.java | 2 +-
.../planner/utils/MaterializedTableUtils.java | 28 ++-
.../operations/SqlCTASNodeToOperationTest.java | 26 +-
.../operations/SqlDdlToOperationConverterTest.java | 8 +-
.../operations/SqlDmlToOperationConverterTest.java | 50 ++--
...erializedTableNodeToOperationConverterTest.java | 277 +++++++++++++++++----
.../operations/SqlModelOperationConverterTest.java | 10 +-
.../operations/SqlNodeToCallOperationTest.java | 4 +-
.../SqlNodeToOperationConversionTestBase.java | 6 +-
.../SqlRTASNodeToOperationConverterTest.java | 30 +--
.../SqlShowToOperationConverterTest.java | 6 +-
19 files changed, 895 insertions(+), 368 deletions(-)
diff --git a/docs/content.zh/docs/dev/table/materialized-table/statements.md b/docs/content.zh/docs/dev/table/materialized-table/statements.md
index a27bce3093f..d65491c9084 100644
--- a/docs/content.zh/docs/dev/table/materialized-table/statements.md
+++ b/docs/content.zh/docs/dev/table/materialized-table/statements.md
@@ -36,7 +36,11 @@ Flink SQL 目前支持以下物化表操作：
```
CREATE MATERIALIZED TABLE [catalog_name.][db_name.]table_name
-[ ([ <table_constraint> ]) ]
+[(
+ { <physical_column_definition> | <metadata_column_definition> | <computed_column_definition> }[ , ...n]
+ [ <watermark_definition> ]
+ [ <table_constraint> ][ , ...n]
+)]
[COMMENT table_comment]
@@ -44,14 +48,29 @@ CREATE MATERIALIZED TABLE [catalog_name.][db_name.]table_name
[WITH (key1=val1, key2=val2, ...)]
-FRESHNESS = INTERVAL '<num>' { SECOND | MINUTE | HOUR | DAY }
+[FRESHNESS = INTERVAL '<num>' { SECOND[S] | MINUTE[S] | HOUR[S] | DAY[S] }]
[REFRESH_MODE = { CONTINUOUS | FULL }]
AS <select_statement>
+<physical_column_definition>:
+ column_name column_type [ <column_constraint> ] [COMMENT column_comment]
+
+<column_constraint>:
+ [CONSTRAINT constraint_name] PRIMARY KEY NOT ENFORCED
+
<table_constraint>:
[CONSTRAINT constraint_name] PRIMARY KEY (column_name, ...) NOT ENFORCED
+
+<metadata_column_definition>:
+ column_name column_type METADATA [ FROM metadata_key ] [ VIRTUAL ]
+
+<computed_column_definition>:
+ column_name AS computed_column_expression [COMMENT column_comment]
+
+<watermark_definition>:
+ WATERMARK FOR rowtime_column_name AS watermark_strategy_expression
```
## PRIMARY KEY
@@ -270,9 +289,32 @@ CREATE MATERIALIZED TABLE my_materialized_table_full
GROUP BY
p.ds, p.product_id, p.product_name
```
+And same materialized table with explicitly specified columns
+
+```sql
+CREATE MATERIALIZED TABLE my_materialized_table_full (
+ ds, product_id, product_name, avg_sale_price, total_quantity)
+ ...
+```
+The order of the columns doesn't need to be the same as in the query,
+Flink will do reordering if required i.e. this will be also valid
+```sql
+CREATE MATERIALIZED TABLE my_materialized_table_full (
+ product_id, product_name, ds, avg_sale_price, total_quantity)
+ ...
+```
+Another way of doing this is putting name and data type
+```sql
+CREATE MATERIALIZED TABLE my_materialized_table_full (
+ ds STRING, product_id STRING, product_name STRING, avg_sale_price DOUBLE, total_quantity BIGINT)
+ ...
+```
+It might happen that types of columns are not the same, in that case implicit casts will be applied.
+If for some of the combinations implicit cast is not supported then there will be validation error thrown.
+Also, it is worth to note that reordering can also be done here.
## 限制
-- 不支持显式指定列名
+- Does not support explicitly specifying physical columns which are not used in the query
- 不支持在 select 查询语句中引用临时表、临时视图或临时函数
# ALTER MATERIALIZED TABLE
diff --git a/docs/content/docs/dev/table/materialized-table/statements.md b/docs/content/docs/dev/table/materialized-table/statements.md
index 07784bb3ede..6e9c0b153b8 100644
--- a/docs/content/docs/dev/table/materialized-table/statements.md
+++ b/docs/content/docs/dev/table/materialized-table/statements.md
@@ -36,7 +36,11 @@ Flink SQL supports the following Materialized Table statements for now:
```
CREATE MATERIALIZED TABLE [catalog_name.][db_name.]table_name
-[ ([ <table_constraint> ]) ]
+[(
+ { <physical_column_definition> | <metadata_column_definition> | <computed_column_definition> }[ , ...n]
+ [ <watermark_definition> ]
+ [ <table_constraint> ][ , ...n]
+)]
[COMMENT table_comment]
@@ -50,8 +54,23 @@ CREATE MATERIALIZED TABLE [catalog_name.][db_name.]table_name
AS <select_statement>
+<physical_column_definition>:
+ column_name column_type [ <column_constraint> ] [COMMENT column_comment]
+
+<column_constraint>:
+ [CONSTRAINT constraint_name] PRIMARY KEY NOT ENFORCED
+
<table_constraint>:
[CONSTRAINT constraint_name] PRIMARY KEY (column_name, ...) NOT ENFORCED
+
+<metadata_column_definition>:
+ column_name column_type METADATA [ FROM metadata_key ] [ VIRTUAL ]
+
+<computed_column_definition>:
+ column_name AS computed_column_expression [COMMENT column_comment]
+
+<watermark_definition>:
+ WATERMARK FOR rowtime_column_name AS watermark_strategy_expression
```
## PRIMARY KEY
@@ -268,10 +287,33 @@ CREATE MATERIALIZED TABLE my_materialized_table_full
GROUP BY
p.ds, p.product_id, p.product_name
```
+And same materialized table with explicitly specified columns
+
+```sql
+CREATE MATERIALIZED TABLE my_materialized_table_full (
+ ds, product_id, product_name, avg_sale_price, total_quantity)
+ ...
+```
+The order of the columns doesn't need to be the same as in the query,
+Flink will do reordering if required i.e. this will be also valid
+```sql
+CREATE MATERIALIZED TABLE my_materialized_table_full (
+ product_id, product_name, ds, avg_sale_price, total_quantity)
+ ...
+```
+Another way of doing this is putting name and data type
+```sql
+CREATE MATERIALIZED TABLE my_materialized_table_full (
+ ds STRING, product_id STRING, product_name STRING, avg_sale_price DOUBLE, total_quantity BIGINT)
+ ...
+```
+It might happen that types of columns are not the same, in that case implicit casts will be applied.
+If for some of the combinations implicit cast is not supported then there will be validation error thrown.
+Also, it is worth to note that reordering can also be done here.
## Limitations
-- Does not support explicitly specifying columns
+- Does not support explicitly specifying physical columns which are not used in the query
- Does not support referring to temporary tables, temporary views, or temporary functions in the select query
# ALTER MATERIALIZED TABLE
diff --git a/flink-table/flink-sql-parser/src/main/codegen/includes/parserImpls.ftl b/flink-table/flink-sql-parser/src/main/codegen/includes/parserImpls.ftl
index 764d0685b97..f7818dc81d8 100644
--- a/flink-table/flink-sql-parser/src/main/codegen/includes/parserImpls.ftl
+++ b/flink-table/flink-sql-parser/src/main/codegen/includes/parserImpls.ftl
@@ -1859,13 +1859,17 @@ SqlCreate SqlCreateMaterializedTable(Span s, boolean replace, boolean isTemporar
final SqlParserPos startPos = s.pos();
SqlIdentifier tableName;
SqlCharStringLiteral comment = null;
- SqlTableConstraint constraint = null;
+ List<SqlTableConstraint> constraints = new ArrayList<SqlTableConstraint>();
+ SqlWatermark watermark = null;
+ SqlNodeList columnList = SqlNodeList.EMPTY;
SqlDistribution distribution = null;
SqlNodeList partitionColumns = SqlNodeList.EMPTY;
SqlNodeList propertyList = SqlNodeList.EMPTY;
SqlNode freshness = null;
SqlLiteral refreshMode = null;
SqlNode asQuery = null;
+ SqlParserPos pos = startPos;
+ boolean isColumnsIdentifiersOnly = false;
}
{
<MATERIALIZED>
@@ -1884,8 +1888,14 @@ SqlCreate SqlCreateMaterializedTable(Span s, boolean replace, boolean isTemporar
<TABLE>
tableName = CompoundIdentifier()
[
- <LPAREN>
- constraint = TableConstraint()
+ <LPAREN> { pos = getPos(); TableCreationContext ctx = new TableCreationContext();}
+ TableColumnsOrIdentifiers(pos, ctx) {
+ pos = pos.plus(getPos());
+ isColumnsIdentifiersOnly = ctx.isColumnsIdentifiersOnly();
+ columnList = new SqlNodeList(ctx.columnList, pos);
+ constraints = ctx.constraints;
+ watermark = ctx.watermark;
+ }
<RPAREN>
]
[
@@ -1939,7 +1949,9 @@ SqlCreate SqlCreateMaterializedTable(Span s, boolean replace, boolean isTemporar
return new SqlCreateMaterializedTable(
startPos.plus(getPos()),
tableName,
- constraint,
+ columnList,
+ constraints,
+ watermark,
comment,
distribution,
partitionColumns,
diff --git a/flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlCreateMaterializedTable.java b/flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlCreateMaterializedTable.java
index 8c7fe2d5089..687226a6220 100644
--- a/flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlCreateMaterializedTable.java
+++ b/flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlCreateMaterializedTable.java
@@ -18,8 +18,11 @@
package org.apache.flink.sql.parser.ddl;
+import org.apache.flink.sql.parser.ExtendedSqlNode;
+import org.apache.flink.sql.parser.SqlConstraintValidator;
import org.apache.flink.sql.parser.SqlUnparseUtils;
import org.apache.flink.sql.parser.ddl.constraint.SqlTableConstraint;
+import org.apache.flink.sql.parser.error.SqlValidateException;
import org.apache.calcite.sql.SqlCharStringLiteral;
import org.apache.calcite.sql.SqlCreate;
@@ -37,21 +40,22 @@ import org.apache.calcite.util.ImmutableNullableList;
import javax.annotation.Nullable;
-import java.util.Collections;
import java.util.List;
import java.util.Optional;
import static java.util.Objects.requireNonNull;
/** CREATE MATERIALIZED TABLE DDL sql call. */
-public class SqlCreateMaterializedTable extends SqlCreate {
+public class SqlCreateMaterializedTable extends SqlCreate implements ExtendedSqlNode {
public static final SqlSpecialOperator OPERATOR =
new SqlSpecialOperator("CREATE MATERIALIZED TABLE", SqlKind.CREATE_TABLE);
private final SqlIdentifier tableName;
- private final @Nullable SqlTableConstraint tableConstraint;
+ private final SqlNodeList columnList;
+
+ private final List<SqlTableConstraint> tableConstraints;
private final @Nullable SqlCharStringLiteral comment;
@@ -59,6 +63,8 @@ public class SqlCreateMaterializedTable extends SqlCreate {
private final SqlNodeList partitionKeyList;
+ private final SqlWatermark watermark;
+
private final SqlNodeList propertyList;
private final @Nullable SqlIntervalLiteral freshness;
@@ -70,7 +76,9 @@ public class SqlCreateMaterializedTable extends SqlCreate {
public SqlCreateMaterializedTable(
SqlParserPos pos,
SqlIdentifier tableName,
- @Nullable SqlTableConstraint tableConstraint,
+ SqlNodeList columnList,
+ List<SqlTableConstraint> tableConstraints,
+ SqlWatermark watermark,
@Nullable SqlCharStringLiteral comment,
@Nullable SqlDistribution distribution,
SqlNodeList partitionKeyList,
@@ -80,7 +88,9 @@ public class SqlCreateMaterializedTable extends SqlCreate {
SqlNode asQuery) {
super(OPERATOR, pos, false, false);
this.tableName = requireNonNull(tableName, "tableName should not be null");
- this.tableConstraint = tableConstraint;
+ this.columnList = columnList;
+ this.tableConstraints = tableConstraints;
+ this.watermark = watermark;
this.comment = comment;
this.distribution = distribution;
this.partitionKeyList =
@@ -100,8 +110,10 @@ public class SqlCreateMaterializedTable extends SqlCreate {
public List<SqlNode> getOperandList() {
return ImmutableNullableList.of(
tableName,
+ columnList,
+ new SqlNodeList(tableConstraints, SqlParserPos.ZERO),
+ watermark,
comment,
- tableConstraint,
partitionKeyList,
propertyList,
freshness,
@@ -116,12 +128,20 @@ public class SqlCreateMaterializedTable extends SqlCreate {
return tableName.names.toArray(new String[0]);
}
- public Optional<SqlCharStringLiteral> getComment() {
- return Optional.ofNullable(comment);
+ public SqlNodeList getColumnList() {
+ return columnList;
+ }
+
+ public List<SqlTableConstraint> getTableConstraints() {
+ return tableConstraints;
+ }
+
+ public Optional<SqlWatermark> getWatermark() {
+ return Optional.ofNullable(watermark);
}
- public Optional<SqlTableConstraint> getTableConstraint() {
- return Optional.ofNullable(tableConstraint);
+ public Optional<SqlCharStringLiteral> getComment() {
+ return Optional.ofNullable(comment);
}
public @Nullable SqlDistribution getDistribution() {
@@ -150,20 +170,33 @@ public class SqlCreateMaterializedTable extends SqlCreate {
return asQuery;
}
+ /** Returns the column constraints plus the table constraints. */
+ public List<SqlTableConstraint> getFullConstraints() {
+ return SqlConstraintValidator.getFullConstraints(tableConstraints, columnList);
+ }
+
+ @Override
+ public void validate() throws SqlValidateException {
+ if (!isSchemaWithColumnsIdentifiersOnly()) {
+ SqlConstraintValidator.validateAndChangeColumnNullability(tableConstraints, columnList);
+ }
+ }
+
+ public boolean isSchemaWithColumnsIdentifiersOnly() {
+ // CREATE MATERIALIZED TABLE supports passing only column identifiers in the column list. If
+ // the first column in the list is an identifier, then we assume the rest of the
+ // columns are identifiers as well.
+ return !getColumnList().isEmpty() && getColumnList().get(0) instanceof SqlIdentifier;
+ }
+
@Override
public void unparse(SqlWriter writer, int leftPrec, int rightPrec) {
writer.keyword("CREATE MATERIALIZED TABLE");
tableName.unparse(writer, leftPrec, rightPrec);
- if (tableConstraint != null) {
- writer.newlineAndIndent();
+ if (!columnList.isEmpty() || !tableConstraints.isEmpty() || watermark != null) {
SqlUnparseUtils.unparseTableSchema(
- writer,
- leftPrec,
- rightPrec,
- SqlNodeList.EMPTY,
- Collections.singletonList(tableConstraint),
- null);
+ writer, leftPrec, rightPrec, columnList, tableConstraints, watermark);
}
if (comment != null) {
diff --git a/flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/MaterializedTableStatementParserTest.java b/flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/MaterializedTableStatementParserTest.java
index d089ecc8079..29f5a52f049 100644
--- a/flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/MaterializedTableStatementParserTest.java
+++ b/flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/MaterializedTableStatementParserTest.java
@@ -39,13 +39,23 @@ class MaterializedTableStatementParserTest {
@ParameterizedTest(name = "{index}: {0}")
@MethodSource("inputForCreateMaterializedTable")
- void testCreateMaterializedTable(Map.Entry<String, String> sqlToExpected) {
+ void testCreateMaterializedTable(
+ final String testName, final Map.Entry<String, String> sqlToExpected) {
final String sql = sqlToExpected.getKey();
final String expected = sqlToExpected.getValue();
sql(sql).ok(expected);
}
+ @Test
+ void testCreateMaterializedTableWithWrongSchema() {
+ final String sql =
+ "CREATE MATERIALIZED TABLE tbl1\n"
+ + "(a, b ^STRING^)\n"
+ + "AS SELECT a, b, h, t m FROM source";
+ sql(sql).fails("(?s).*Encountered \"STRING\" at line 2, column 7.*");
+ }
+
@Test
void testCreateMaterializedTableWithUnsupportedFreshnessInterval() {
final String sql =
@@ -373,17 +383,23 @@ class MaterializedTableStatementParserTest {
private static Stream<Arguments> inputForCreateMaterializedTable() {
return Stream.of(
- Arguments.of(fullExample()),
- Arguments.of(withoutTableConstraint()),
- Arguments.of(withPrimaryKey()),
- Arguments.of(withoutFreshness()));
+ Arguments.of("Full example", fullExample()),
+ Arguments.of("With columns", withColumns()),
+ Arguments.of("With columns and watermarks", withColumnsAndWatermark()),
+ Arguments.of("Without table constraint", withoutTableConstraint()),
+ Arguments.of("With primary key", withPrimaryKey()),
+ Arguments.of("Without freshness", withoutFreshness()),
+ Arguments.of("With column identifiers only", withColumnsIdentifiersOnly()));
}
private static Map.Entry<String, String> fullExample() {
return new AbstractMap.SimpleEntry<>(
"CREATE MATERIALIZED TABLE tbl1\n"
+ "(\n"
- + " PRIMARY KEY (a, b)\n"
+ + " ts timestamp(3),\n"
+ + " id varchar,\n"
+ + " watermark FOR ts AS ts - interval '3' second,\n"
+ + " PRIMARY KEY (id)\n"
+ ")\n"
+ "COMMENT 'table comment'\n"
+ "DISTRIBUTED BY HASH (a) INTO 4 BUCKETS\n"
@@ -394,9 +410,11 @@ class MaterializedTableStatementParserTest {
+ ")\n"
+ "FRESHNESS = INTERVAL '3' MINUTES\n"
+ "AS SELECT a, b, h, t m FROM source",
- "CREATE MATERIALIZED TABLE `TBL1`\n"
- + "(\n"
- + " PRIMARY KEY (`A`, `B`)\n"
+ "CREATE MATERIALIZED TABLE `TBL1` (\n"
+ + " `TS` TIMESTAMP(3),\n"
+ + " `ID` VARCHAR,\n"
+ + " PRIMARY KEY (`ID`),\n"
+ + " WATERMARK FOR `TS` AS (`TS` - INTERVAL '3' SECOND)\n"
+ ")\n"
+ "COMMENT 'table comment'\n"
+ "DISTRIBUTED BY HASH(`A`) INTO 4 BUCKETS\n"
@@ -421,8 +439,7 @@ class MaterializedTableStatementParserTest {
+ "FRESHNESS = INTERVAL '3' MINUTE\n"
+ "REFRESH_MODE = FULL\n"
+ "AS SELECT a, b, h, t m FROM source",
- "CREATE MATERIALIZED TABLE `TBL1`\n"
- + "(\n"
+ "CREATE MATERIALIZED TABLE `TBL1` (\n"
+ " PRIMARY KEY (`A`, `B`)\n"
+ ")\n"
+ "COMMENT 'table comment'\n"
@@ -460,8 +477,7 @@ class MaterializedTableStatementParserTest {
+ " 'kafka.topic' = 'log.test'\n"
+ ")\n"
+ "AS SELECT a, b, h, t m FROM source",
- "CREATE MATERIALIZED TABLE `TBL1`\n"
- + "(\n"
+ "CREATE MATERIALIZED TABLE `TBL1` (\n"
+ " PRIMARY KEY (`A`, `B`)\n"
+ ")\n"
+ "WITH (\n"
@@ -472,4 +488,105 @@ class MaterializedTableStatementParserTest {
+ "SELECT `A`, `B`, `H`, `T` AS `M`\n"
+ "FROM `SOURCE`");
}
+
+ private static Map.Entry<String, String> withColumns() {
+ return new AbstractMap.SimpleEntry<>(
+ "CREATE MATERIALIZED TABLE tbl1\n"
+ + "(\n"
+ + " a INT, b STRING, h INT, m INT\n"
+ + ")\n"
+ + "COMMENT 'table comment'\n"
+ + "DISTRIBUTED BY HASH (a) INTO 4 BUCKETS\n"
+ + "PARTITIONED BY (a, h)\n"
+ + "WITH (\n"
+ + " 'group.id' = 'latest', \n"
+ + " 'kafka.topic' = 'log.test'\n"
+ + ")\n"
+ + "FRESHNESS = INTERVAL '3' MINUTE\n"
+ + "AS SELECT a, b, h, t m FROM source",
+ "CREATE MATERIALIZED TABLE `TBL1` (\n"
+ + " `A` INTEGER,\n"
+ + " `B` STRING,\n"
+ + " `H` INTEGER,\n"
+ + " `M` INTEGER\n"
+ + ")\n"
+ + "COMMENT 'table comment'\n"
+ + "DISTRIBUTED BY HASH(`A`) INTO 4 BUCKETS\n"
+ + "PARTITIONED BY (`A`, `H`)\n"
+ + "WITH (\n"
+ + " 'group.id' = 'latest',\n"
+ + " 'kafka.topic' = 'log.test'\n"
+ + ")\n"
+ + "FRESHNESS = INTERVAL '3' MINUTE\n"
+ + "AS\n"
+ + "SELECT `A`, `B`, `H`, `T` AS `M`\n"
+ + "FROM `SOURCE`");
+ }
+
+ private static Map.Entry<String, String> withColumnsAndWatermark() {
+ return new AbstractMap.SimpleEntry<>(
+ "CREATE MATERIALIZED TABLE tbl1\n"
+ + "(\n"
+ + " ts timestamp(3),\n"
+ + " id varchar,\n"
+ + " watermark FOR ts AS ts - interval '3' second\n"
+ + ")\n"
+ + "COMMENT 'table comment'\n"
+ + "DISTRIBUTED BY HASH (a) INTO 4 BUCKETS\n"
+ + "PARTITIONED BY (a, h)\n"
+ + "WITH (\n"
+ + " 'group.id' = 'latest', \n"
+ + " 'kafka.topic' = 'log.test'\n"
+ + ")\n"
+ + "FRESHNESS = INTERVAL '3' MINUTE\n"
+ + "AS SELECT a, b, h, t m FROM source",
+ "CREATE MATERIALIZED TABLE `TBL1` (\n"
+ + " `TS` TIMESTAMP(3),\n"
+ + " `ID` VARCHAR,\n"
+ + " WATERMARK FOR `TS` AS (`TS` - INTERVAL '3' SECOND)\n"
+ + ")\n"
+ + "COMMENT 'table comment'\n"
+ + "DISTRIBUTED BY HASH(`A`) INTO 4 BUCKETS\n"
+ + "PARTITIONED BY (`A`, `H`)\n"
+ + "WITH (\n"
+ + " 'group.id' = 'latest',\n"
+ + " 'kafka.topic' = 'log.test'\n"
+ + ")\n"
+ + "FRESHNESS = INTERVAL '3' MINUTE\n"
+ + "AS\n"
+ + "SELECT `A`, `B`, `H`, `T` AS `M`\n"
+ + "FROM `SOURCE`");
+ }
+
+ private static Map.Entry<String, String> withColumnsIdentifiersOnly() {
+ return new AbstractMap.SimpleEntry<>(
+ "CREATE MATERIALIZED TABLE tbl1\n"
+ + "(a, b, h, m)\n"
+ + "COMMENT 'table comment'\n"
+ + "DISTRIBUTED BY HASH (a) INTO 4 BUCKETS\n"
+ + "PARTITIONED BY (a, h)\n"
+ + "WITH (\n"
+ + " 'group.id' = 'latest', \n"
+ + " 'kafka.topic' = 'log.test'\n"
+ + ")\n"
+ + "FRESHNESS = INTERVAL '3' MINUTE\n"
+ + "AS SELECT a, b, h, t m FROM source",
+ "CREATE MATERIALIZED TABLE `TBL1` (\n"
+ + " `A`,\n"
+ + " `B`,\n"
+ + " `H`,\n"
+ + " `M`\n"
+ + ")\n"
+ + "COMMENT 'table comment'\n"
+ + "DISTRIBUTED BY HASH(`A`) INTO 4 BUCKETS\n"
+ + "PARTITIONED BY (`A`, `H`)\n"
+ + "WITH (\n"
+ + " 'group.id' = 'latest',\n"
+ + " 'kafka.topic' = 'log.test'\n"
+ + ")\n"
+ + "FRESHNESS = INTERVAL '3' MINUTE\n"
+ + "AS\n"
+ + "SELECT `A`, `B`, `H`, `T` AS `M`\n"
+ + "FROM `SOURCE`");
+ }
}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/AbstractCreateMaterializedTableConverter.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/AbstractCreateMaterializedTableConverter.java
new file mode 100644
index 00000000000..fb61e48b7ef
--- /dev/null
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/AbstractCreateMaterializedTableConverter.java
@@ -0,0 +1,221 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.operations.converters;
+
+import org.apache.flink.sql.parser.ddl.SqlCreateMaterializedTable;
+import org.apache.flink.sql.parser.ddl.SqlRefreshMode;
+import org.apache.flink.table.api.Schema;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.catalog.CatalogMaterializedTable;
+import org.apache.flink.table.catalog.CatalogMaterializedTable.LogicalRefreshMode;
+import org.apache.flink.table.catalog.CatalogMaterializedTable.RefreshMode;
+import org.apache.flink.table.catalog.IntervalFreshness;
+import org.apache.flink.table.catalog.ObjectIdentifier;
+import org.apache.flink.table.catalog.ResolvedCatalogMaterializedTable;
+import org.apache.flink.table.catalog.ResolvedSchema;
+import org.apache.flink.table.catalog.TableDistribution;
+import org.apache.flink.table.catalog.UnresolvedIdentifier;
+import org.apache.flink.table.planner.operations.PlannerQueryOperation;
+import org.apache.flink.table.planner.utils.MaterializedTableUtils;
+import org.apache.flink.table.planner.utils.OperationConverterUtils;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeFamily;
+
+import org.apache.calcite.sql.SqlNode;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.table.api.config.MaterializedTableConfigOptions.DATE_FORMATTER;
+import static org.apache.flink.table.api.config.MaterializedTableConfigOptions.PARTITION_FIELDS;
+
+/**
+ * Abstract class for converting {@link SqlCreateMaterializedTable} and it's children to create
+ * materialized table operations.
+ */
+public abstract class AbstractCreateMaterializedTableConverter<T extends SqlCreateMaterializedTable>
+ implements SqlNodeConverter<T> {
+ /** Context of create table converters while merging source and derived items. */
+ protected interface MergeContext {
+ Schema getMergedSchema();
+
+ Map<String, String> getMergedTableOptions();
+
+ List<String> getMergedPartitionKeys();
+
+ Optional<TableDistribution> getMergedTableDistribution();
+
+ String getMergedDefinitionQuery();
+
+ ResolvedSchema getMergedQuerySchema();
+ }
+
+ protected abstract MergeContext getMergeContext(
+ T sqlCreateMaterializedTable, ConvertContext context);
+
+ protected final Optional<TableDistribution> getDerivedTableDistribution(
+ T sqlCreateMaterializedTable) {
+ return Optional.ofNullable(sqlCreateMaterializedTable.getDistribution())
+ .map(OperationConverterUtils::getDistributionFromSqlDistribution);
+ }
+
+ protected final List<String> getDerivedPartitionKeys(T sqlCreateMaterializedTable) {
+ return OperationConverterUtils.getColumnNames(
+ sqlCreateMaterializedTable.getPartitionKeyList());
+ }
+
+ protected final Map<String, String> getDerivedTableOptions(T sqlCreateMaterializedTable) {
+ return OperationConverterUtils.getProperties(sqlCreateMaterializedTable.getPropertyList());
+ }
+
+ protected final IntervalFreshness getDerivedFreshness(T sqlCreateMaterializedTable) {
+ return Optional.ofNullable(sqlCreateMaterializedTable.getFreshness())
+ .map(MaterializedTableUtils::getMaterializedTableFreshness)
+ .orElse(null);
+ }
+
+ protected final ResolvedSchema getQueryResolvedSchema(
+ T sqlCreateMaterializedTable, ConvertContext context) {
+ SqlNode selectQuery = sqlCreateMaterializedTable.getAsQuery();
+ SqlNode validateQuery = context.getSqlValidator().validate(selectQuery);
+
+ PlannerQueryOperation queryOperation =
+ new PlannerQueryOperation(
+ context.toRelRoot(validateQuery).project(),
+ () -> context.toQuotedSqlString(validateQuery));
+ return queryOperation.getResolvedSchema();
+ }
+
+ protected final LogicalRefreshMode getDerivedLogicalRefreshMode(T sqlCreateMaterializedTable) {
+ SqlRefreshMode sqlRefreshMode =
+ Optional.ofNullable(sqlCreateMaterializedTable.getRefreshMode())
+ .map(mode -> mode.getValueAs(SqlRefreshMode.class))
+ .orElse(null);
+
+ return MaterializedTableUtils.deriveLogicalRefreshMode(sqlRefreshMode);
+ }
+
+ protected final RefreshMode getDerivedRefreshMode(LogicalRefreshMode logicalRefreshMode) {
+ return MaterializedTableUtils.fromLogicalRefreshModeToRefreshMode(logicalRefreshMode);
+ }
+
+ protected final String getDerivedDefinitionQuery(
+ T sqlCreateMaterializedTable, ConvertContext context) {
+ SqlNode selectQuery = sqlCreateMaterializedTable.getAsQuery();
+ SqlNode validatedQuery = context.getSqlValidator().validate(selectQuery);
+ return context.toQuotedSqlString(validatedQuery);
+ }
+
+ protected final String getComment(T sqlCreateMaterializedTable) {
+ return OperationConverterUtils.getComment(sqlCreateMaterializedTable.getComment());
+ }
+
+ protected final ResolvedCatalogMaterializedTable getResolvedCatalogMaterializedTable(
+ T sqlCreateMaterializedTable, ConvertContext context) {
+ final MergeContext mergeContext = getMergeContext(sqlCreateMaterializedTable, context);
+ final List<String> partitionKeys = mergeContext.getMergedPartitionKeys();
+ final Schema schema = mergeContext.getMergedSchema();
+ final ResolvedSchema querySchema = mergeContext.getMergedQuerySchema();
+ final Map<String, String> tableOptions = mergeContext.getMergedTableOptions();
+ verifyPartitioningColumnsExist(querySchema, partitionKeys, tableOptions);
+
+ final TableDistribution distribution =
+ mergeContext.getMergedTableDistribution().orElse(null);
+ final String comment = getComment(sqlCreateMaterializedTable);
+ final String definitionQuery = mergeContext.getMergedDefinitionQuery();
+ final IntervalFreshness intervalFreshness = getDerivedFreshness(sqlCreateMaterializedTable);
+ final LogicalRefreshMode logicalRefreshMode =
+ getDerivedLogicalRefreshMode(sqlCreateMaterializedTable);
+ final RefreshMode refreshMode = getDerivedRefreshMode(logicalRefreshMode);
+ return context.getCatalogManager()
+ .resolveCatalogMaterializedTable(
+ CatalogMaterializedTable.newBuilder()
+ .schema(schema)
+ .comment(comment)
+ .distribution(distribution)
+ .partitionKeys(partitionKeys)
+ .options(tableOptions)
+ .definitionQuery(definitionQuery)
+ .freshness(intervalFreshness)
+ .logicalRefreshMode(logicalRefreshMode)
+ .refreshMode(refreshMode)
+ .refreshStatus(CatalogMaterializedTable.RefreshStatus.INITIALIZING)
+ .build());
+ }
+
+ protected final ObjectIdentifier getIdentifier(
+ SqlCreateMaterializedTable node, ConvertContext context) {
+ UnresolvedIdentifier unresolvedIdentifier = UnresolvedIdentifier.of(node.fullTableName());
+ return context.getCatalogManager().qualifyIdentifier(unresolvedIdentifier);
+ }
+
+ private void verifyPartitioningColumnsExist(
+ ResolvedSchema schema, List<String> partitionKeys, Map<String, String> tableOptions) {
+ final Set<String> partitionFieldOptions =
+ tableOptions.keySet().stream()
+ .filter(k -> k.startsWith(PARTITION_FIELDS))
+ .collect(Collectors.toSet());
+
+ for (String partitionKey : partitionKeys) {
+ if (schema.getColumn(partitionKey).isEmpty()) {
+ throw new ValidationException(
+ String.format(
+ "Partition column '%s' not defined in the query schema. Available columns: [%s].",
+ partitionKey,
+ schema.getColumnNames().stream()
+ .collect(Collectors.joining("', '", "'", "'"))));
+ }
+ }
+
+ // verify partition key used by materialized table partition option
+ // partition.fields.#.date-formatter whether exist
+ for (String partitionOption : partitionFieldOptions) {
+ String partitionKey =
+ partitionOption.substring(
+ PARTITION_FIELDS.length() + 1,
+ partitionOption.length() - (DATE_FORMATTER.length() + 1));
+ // partition key used in option partition.fields.#.date-formatter must be existed
+ if (!partitionKeys.contains(partitionKey)) {
+ throw new ValidationException(
+ String.format(
+ "Column '%s' referenced by materialized table option '%s' isn't a partition column. Available partition columns: [%s].",
+ partitionKey,
+ partitionOption,
+ partitionKeys.stream()
+ .collect(Collectors.joining("', '", "'", "'"))));
+ }
+
+ // partition key used in option partition.fields.#.date-formatter must be string type
+ LogicalType partitionKeyType =
+ schema.getColumn(partitionKey).get().getDataType().getLogicalType();
+ if (!partitionKeyType
+ .getTypeRoot()
+ .getFamilies()
+ .contains(LogicalTypeFamily.CHARACTER_STRING)) {
+ throw new ValidationException(
+ String.format(
+ "Materialized table option '%s' only supports referring to char, varchar and string type partition column. Column %s type is %s.",
+ partitionOption, partitionKey, partitionKeyType.asSummaryString()));
+ }
+ }
+ }
+}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlCreateMaterializedTableConverter.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlCreateMaterializedTableConverter.java
index 2107b209b27..3beb5596a90 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlCreateMaterializedTableConverter.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlCreateMaterializedTableConverter.java
@@ -18,234 +18,109 @@
package org.apache.flink.table.planner.operations.converters;
-import org.apache.flink.sql.parser.SqlConstraintValidator;
import org.apache.flink.sql.parser.ddl.SqlCreateMaterializedTable;
-import org.apache.flink.sql.parser.ddl.SqlRefreshMode;
-import org.apache.flink.sql.parser.ddl.constraint.SqlTableConstraint;
-import org.apache.flink.sql.parser.error.SqlValidateException;
+import org.apache.flink.sql.parser.ddl.SqlTableColumn.SqlRegularColumn;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.ValidationException;
-import org.apache.flink.table.catalog.CatalogMaterializedTable;
-import org.apache.flink.table.catalog.CatalogMaterializedTable.LogicalRefreshMode;
-import org.apache.flink.table.catalog.CatalogMaterializedTable.RefreshMode;
-import org.apache.flink.table.catalog.Column;
-import org.apache.flink.table.catalog.IntervalFreshness;
-import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.catalog.ResolvedSchema;
import org.apache.flink.table.catalog.TableDistribution;
-import org.apache.flink.table.catalog.UnresolvedIdentifier;
import org.apache.flink.table.operations.Operation;
import org.apache.flink.table.operations.materializedtable.CreateMaterializedTableOperation;
-import org.apache.flink.table.planner.operations.PlannerQueryOperation;
-import org.apache.flink.table.planner.utils.MaterializedTableUtils;
-import org.apache.flink.table.planner.utils.OperationConverterUtils;
-import org.apache.flink.table.types.logical.LogicalType;
-import org.apache.flink.table.types.logical.LogicalTypeFamily;
+import org.apache.flink.table.planner.operations.converters.table.MergeTableAsUtil;
import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlNodeList;
-import java.util.Arrays;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
-import java.util.stream.Collectors;
-import static org.apache.flink.table.api.config.MaterializedTableConfigOptions.DATE_FORMATTER;
-import static org.apache.flink.table.api.config.MaterializedTableConfigOptions.PARTITION_FIELDS;
-import static org.apache.flink.table.catalog.IntervalFreshness.validateFreshnessForCron;
-
-/** A converter for {@link SqlCreateMaterializedTable}. */
+/**
+ * A converter for {@link SqlCreateMaterializedTable} to {@link CreateMaterializedTableOperation}.
+ */
public class SqlCreateMaterializedTableConverter
- implements SqlNodeConverter<SqlCreateMaterializedTable> {
+ extends AbstractCreateMaterializedTableConverter<SqlCreateMaterializedTable> {
@Override
public Operation convertSqlNode(
SqlCreateMaterializedTable sqlCreateMaterializedTable, ConvertContext context) {
- UnresolvedIdentifier unresolvedIdentifier =
- UnresolvedIdentifier.of(sqlCreateMaterializedTable.fullTableName());
- ObjectIdentifier identifier =
- context.getCatalogManager().qualifyIdentifier(unresolvedIdentifier);
-
- // get comment
- String tableComment =
- OperationConverterUtils.getComment(sqlCreateMaterializedTable.getComment());
-
- // get options
- final Map<String, String> tableOptions =
- OperationConverterUtils.getProperties(sqlCreateMaterializedTable.getPropertyList());
-
- // get freshness
- IntervalFreshness intervalFreshness =
- Optional.ofNullable(sqlCreateMaterializedTable.getFreshness())
- .map(MaterializedTableUtils::getMaterializedTableFreshness)
- .orElse(null);
-
- // Get the logical refresh mode from SQL
- SqlRefreshMode sqlRefreshMode =
- Optional.ofNullable(sqlCreateMaterializedTable.getRefreshMode())
- .map(mode -> mode.getValueAs(SqlRefreshMode.class))
- .orElse(null);
-
- final LogicalRefreshMode logicalRefreshMode =
- MaterializedTableUtils.deriveLogicalRefreshMode(sqlRefreshMode);
-
- // get the physical refresh mode from SQL
- final RefreshMode refreshMode =
- sqlRefreshMode == null
- ? null
- : MaterializedTableUtils.fromSqltoRefreshMode(sqlRefreshMode);
-
- if (CatalogMaterializedTable.RefreshMode.FULL == refreshMode && intervalFreshness != null) {
- validateFreshnessForCron(intervalFreshness);
- }
-
- // get query schema and definition query
- SqlNode selectQuery = sqlCreateMaterializedTable.getAsQuery();
- SqlNode validatedQuery = context.getSqlValidator().validate(selectQuery);
-
- String definitionQuery = context.toQuotedSqlString(validatedQuery);
-
- PlannerQueryOperation queryOperation =
- new PlannerQueryOperation(
- context.toRelRoot(validatedQuery).project(), () -> definitionQuery);
-
- // get schema
- ResolvedSchema resolvedSchema = queryOperation.getResolvedSchema();
- Schema.Builder builder = Schema.newBuilder().fromResolvedSchema(resolvedSchema);
-
- // get and verify partition key
- List<String> partitionKeys =
- OperationConverterUtils.getColumnNames(
- sqlCreateMaterializedTable.getPartitionKeyList());
- verifyPartitioningColumnsExist(
- resolvedSchema,
- partitionKeys,
- tableOptions.keySet().stream()
- .filter(k -> k.startsWith(PARTITION_FIELDS))
- .collect(Collectors.toSet()));
-
- // verify and build primary key
- sqlCreateMaterializedTable
- .getTableConstraint()
- .ifPresent(
- sqlTableConstraint ->
- verifyAndBuildPrimaryKey(
- builder, resolvedSchema, sqlTableConstraint));
-
- Optional<TableDistribution> tableDistribution =
- Optional.ofNullable(sqlCreateMaterializedTable.getDistribution())
- .map(OperationConverterUtils::getDistributionFromSqlDistribution);
-
- CatalogMaterializedTable materializedTable =
- CatalogMaterializedTable.newBuilder()
- .schema(builder.build())
- .comment(tableComment)
- .distribution(tableDistribution.orElse(null))
- .partitionKeys(partitionKeys)
- .options(tableOptions)
- .definitionQuery(definitionQuery)
- .freshness(intervalFreshness)
- .logicalRefreshMode(logicalRefreshMode)
- .refreshMode(refreshMode)
- .refreshStatus(CatalogMaterializedTable.RefreshStatus.INITIALIZING)
- .build();
-
return new CreateMaterializedTableOperation(
- identifier,
- context.getCatalogManager().resolveCatalogMaterializedTable(materializedTable));
+ getIdentifier(sqlCreateMaterializedTable, context),
+ getResolvedCatalogMaterializedTable(sqlCreateMaterializedTable, context));
}
- private static void verifyPartitioningColumnsExist(
- ResolvedSchema resolvedSchema,
- List<String> partitionKeys,
- Set<String> partitionFieldOptions) {
- // verify partition key whether exists
- for (String partitionKey : partitionKeys) {
- if (!resolvedSchema.getColumn(partitionKey).isPresent()) {
- throw new ValidationException(
- String.format(
- "Partition column '%s' not defined in the query schema. Available columns: [%s].",
- partitionKey,
- resolvedSchema.getColumnNames().stream()
- .collect(Collectors.joining("', '", "'", "'"))));
+ @Override
+ protected MergeContext getMergeContext(
+ SqlCreateMaterializedTable sqlCreateMaterializedTable, ConvertContext context) {
+ return new MergeContext() {
+ private final MergeTableAsUtil mergeTableAsUtil = new MergeTableAsUtil(context);
+ private final String definitionQuery =
+ SqlCreateMaterializedTableConverter.this.getDerivedDefinitionQuery(
+ sqlCreateMaterializedTable, context);
+ private final ResolvedSchema querySchema =
+ SqlCreateMaterializedTableConverter.this.getQueryResolvedSchema(
+ sqlCreateMaterializedTable, context);
+
+ @Override
+ public Schema getMergedSchema() {
+ final Set<String> querySchemaColumnNames =
+ new HashSet<>(querySchema.getColumnNames());
+ final SqlNodeList sqlNodeList = sqlCreateMaterializedTable.getColumnList();
+ for (SqlNode column : sqlNodeList) {
+ if (!(column instanceof SqlRegularColumn)) {
+ continue;
+ }
+
+ SqlRegularColumn physicalColumn = (SqlRegularColumn) column;
+ if (!querySchemaColumnNames.contains(physicalColumn.getName().getSimple())) {
+ throw new ValidationException(
+ String.format(
+ "Invalid as physical column '%s' is defined in the DDL, but is not used in a query column.",
+ physicalColumn.getName().getSimple()));
+ }
+ }
+ if (sqlCreateMaterializedTable.isSchemaWithColumnsIdentifiersOnly()) {
+ // If only column identifiers are provided, then these are used to
+ // order the columns in the schema.
+ return mergeTableAsUtil.reorderSchema(sqlNodeList, querySchema);
+ } else {
+ return mergeTableAsUtil.mergeSchemas(
+ sqlNodeList,
+ sqlCreateMaterializedTable.getWatermark().orElse(null),
+ sqlCreateMaterializedTable.getFullConstraints(),
+ querySchema);
+ }
}
- }
- // verify partition key used by materialized table partition option
- // partition.fields.#.date-formatter whether exist
- for (String partitionOption : partitionFieldOptions) {
- String partitionKey =
- partitionOption.substring(
- PARTITION_FIELDS.length() + 1,
- partitionOption.length() - (DATE_FORMATTER.length() + 1));
- // partition key used in option partition.fields.#.date-formatter must be existed
- if (!partitionKeys.contains(partitionKey)) {
- throw new ValidationException(
- String.format(
- "Column '%s' referenced by materialized table option '%s' isn't a partition column. Available partition columns: [%s].",
- partitionKey,
- partitionOption,
- partitionKeys.stream()
- .collect(Collectors.joining("', '", "'", "'"))));
+ @Override
+ public Map<String, String> getMergedTableOptions() {
+ return SqlCreateMaterializedTableConverter.this.getDerivedTableOptions(
+ sqlCreateMaterializedTable);
}
- // partition key used in option partition.fields.#.date-formatter must be string type
- LogicalType partitionKeyType =
- resolvedSchema.getColumn(partitionKey).get().getDataType().getLogicalType();
- if (!partitionKeyType
- .getTypeRoot()
- .getFamilies()
- .contains(LogicalTypeFamily.CHARACTER_STRING)) {
- throw new ValidationException(
- String.format(
- "Materialized table option '%s' only supports referring to char, varchar and string type partition column. Column %s type is %s.",
- partitionOption, partitionKey, partitionKeyType.asSummaryString()));
+ @Override
+ public List<String> getMergedPartitionKeys() {
+ return SqlCreateMaterializedTableConverter.this.getDerivedPartitionKeys(
+ sqlCreateMaterializedTable);
}
- }
- }
- private static void verifyAndBuildPrimaryKey(
- Schema.Builder schemaBuilder,
- ResolvedSchema resolvedSchema,
- SqlTableConstraint sqlTableConstraint) {
- // check constraint type
- try {
- SqlConstraintValidator.validate(sqlTableConstraint);
- } catch (SqlValidateException e) {
- throw new ValidationException(
- String.format("Primary key validation failed: %s.", e.getMessage()), e);
- }
-
- List<String> primaryKeyColumns = Arrays.asList(sqlTableConstraint.getColumnNames());
- for (String columnName : primaryKeyColumns) {
- Optional<Column> columnOptional = resolvedSchema.getColumn(columnName);
- if (!columnOptional.isPresent()) {
- throw new ValidationException(
- String.format(
- "Primary key column '%s' not defined in the query schema. Available columns: [%s].",
- columnName,
- resolvedSchema.getColumnNames().stream()
- .collect(Collectors.joining("', '", "'", "'"))));
+ @Override
+ public Optional<TableDistribution> getMergedTableDistribution() {
+ return SqlCreateMaterializedTableConverter.this.getDerivedTableDistribution(
+ sqlCreateMaterializedTable);
}
- if (columnOptional.get().getDataType().getLogicalType().isNullable()) {
- throw new ValidationException(
- String.format(
- "Could not create a PRIMARY KEY with nullable column '%s'.\n"
- + "A PRIMARY KEY column must be declared on non-nullable physical columns.",
- columnName));
+ @Override
+ public String getMergedDefinitionQuery() {
+ return definitionQuery;
}
- }
- // build primary key
- String constraintName =
- sqlTableConstraint
- .getConstraintName()
- .orElseGet(
- () ->
- primaryKeyColumns.stream()
- .collect(Collectors.joining("_", "PK_", "")));
- schemaBuilder.primaryKeyNamed(constraintName, primaryKeyColumns);
+ @Override
+ public ResolvedSchema getMergedQuerySchema() {
+ return querySchema;
+ }
+ };
}
}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/table/AbstractCreateTableConverter.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/table/AbstractCreateTableConverter.java
index 3a3de75c9f2..c22c52b9457 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/table/AbstractCreateTableConverter.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/table/AbstractCreateTableConverter.java
@@ -45,7 +45,7 @@ import java.util.stream.Collectors;
public abstract class AbstractCreateTableConverter<T extends SqlCreateTable>
implements SqlNodeConverter<T> {
- /** Context of create table converters while mering source and derived items. */
+ /** Context of create table converters while merging source and derived items. */
protected interface MergeContext {
Schema getMergedSchema(ResolvedSchema schemaToMerge);
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/table/MergeTableAsUtil.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/table/MergeTableAsUtil.java
index a96c607693c..1af8044cc25 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/table/MergeTableAsUtil.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/table/MergeTableAsUtil.java
@@ -351,7 +351,7 @@ public class MergeTableAsUtil {
String name = ((SqlIdentifier) identifier).getSimple();
if (!sourceSchemaCols.containsKey(name)) {
throw new ValidationException(
- String.format("Column '%s' not found in the source schema. ", name));
+ String.format("Column '%s' not found in the source schema.", name));
}
sinkSchemaCols.put(name, sourceSchemaCols.get(name));
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/utils/MaterializedTableUtils.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/utils/MaterializedTableUtils.java
index ba765e40a69..2f32aac16f6 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/utils/MaterializedTableUtils.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/utils/MaterializedTableUtils.java
@@ -21,7 +21,7 @@ package org.apache.flink.table.planner.utils;
import org.apache.flink.annotation.Internal;
import org.apache.flink.sql.parser.ddl.SqlRefreshMode;
import org.apache.flink.table.api.ValidationException;
-import org.apache.flink.table.catalog.CatalogMaterializedTable;
+import org.apache.flink.table.catalog.CatalogMaterializedTable.LogicalRefreshMode;
import org.apache.flink.table.catalog.CatalogMaterializedTable.RefreshMode;
import org.apache.flink.table.catalog.IntervalFreshness;
@@ -61,24 +61,23 @@ public class MaterializedTableUtils {
}
}
- public static CatalogMaterializedTable.LogicalRefreshMode deriveLogicalRefreshMode(
- SqlRefreshMode sqlRefreshMode) {
+ public static LogicalRefreshMode deriveLogicalRefreshMode(SqlRefreshMode sqlRefreshMode) {
if (sqlRefreshMode == null) {
- return CatalogMaterializedTable.LogicalRefreshMode.AUTOMATIC;
+ return LogicalRefreshMode.AUTOMATIC;
}
switch (sqlRefreshMode) {
case FULL:
- return CatalogMaterializedTable.LogicalRefreshMode.FULL;
+ return LogicalRefreshMode.FULL;
case CONTINUOUS:
- return CatalogMaterializedTable.LogicalRefreshMode.CONTINUOUS;
+ return LogicalRefreshMode.CONTINUOUS;
default:
throw new ValidationException(
String.format("Unsupported logical refresh mode: %s.", sqlRefreshMode));
}
}
- public static RefreshMode fromSqltoRefreshMode(SqlRefreshMode sqlRefreshMode) {
+ public static RefreshMode fromSqlToRefreshMode(SqlRefreshMode sqlRefreshMode) {
switch (sqlRefreshMode) {
case FULL:
return RefreshMode.FULL;
@@ -88,4 +87,19 @@ public class MaterializedTableUtils {
throw new IllegalArgumentException("Unknown refresh mode: " + sqlRefreshMode);
}
}
+
+ public static RefreshMode fromLogicalRefreshModeToRefreshMode(
+ LogicalRefreshMode logicalRefreshMode) {
+ switch (logicalRefreshMode) {
+ case AUTOMATIC:
+ return null;
+ case FULL:
+ return RefreshMode.FULL;
+ case CONTINUOUS:
+ return RefreshMode.CONTINUOUS;
+ default:
+ throw new IllegalArgumentException(
+ "Unknown logical refresh mode: " + logicalRefreshMode);
+ }
+ }
}
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlCTASNodeToOperationTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlCTASNodeToOperationTest.java
index b4aa3ee7837..3add4e37f46 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlCTASNodeToOperationTest.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlCTASNodeToOperationTest.java
@@ -48,13 +48,13 @@ import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Test base for testing convert CREATE TABLE AS statement to operation. */
-public class SqlCTASNodeToOperationTest extends SqlNodeToOperationConversionTestBase {
+class SqlCTASNodeToOperationTest extends SqlNodeToOperationConversionTestBase {
private static final Map<String, String> TABLE_OPTIONS =
Map.of("connector", TestSimpleDynamicTableSourceFactory.IDENTIFIER());
@Test
- public void testCreateTableAsWithNotFoundColumnIdentifiers() {
+ void testCreateTableAsWithNotFoundColumnIdentifiers() {
CatalogTable catalogTable =
CatalogTable.newBuilder()
.schema(
@@ -76,7 +76,7 @@ public class SqlCTASNodeToOperationTest extends SqlNodeToOperationConversionTest
}
@Test
- public void testCreateTableAsWithMismatchIdentifiersLength() {
+ void testCreateTableAsWithMismatchIdentifiersLength() {
CatalogTable catalogTable =
CatalogTable.newBuilder()
.schema(
@@ -100,7 +100,7 @@ public class SqlCTASNodeToOperationTest extends SqlNodeToOperationConversionTest
}
@Test
- public void testCreateTableAsWithColumns() {
+ void testCreateTableAsWithColumns() {
CatalogTable catalogTable =
CatalogTable.newBuilder()
.schema(
@@ -142,7 +142,7 @@ public class SqlCTASNodeToOperationTest extends SqlNodeToOperationConversionTest
}
@Test
- public void testCreateTableAsWithColumnsOverridden() {
+ void testCreateTableAsWithColumnsOverridden() {
CatalogTable catalogTable =
CatalogTable.newBuilder()
.schema(
@@ -181,7 +181,7 @@ public class SqlCTASNodeToOperationTest extends SqlNodeToOperationConversionTest
}
@Test
- public void testCreateTableAsWithOverriddenVirtualMetadataColumnsNotAllowed() {
+ void testCreateTableAsWithOverriddenVirtualMetadataColumnsNotAllowed() {
CatalogTable catalogTable =
CatalogTable.newBuilder()
.schema(
@@ -207,7 +207,7 @@ public class SqlCTASNodeToOperationTest extends SqlNodeToOperationConversionTest
}
@Test
- public void testCreateTableAsWithOverriddenComputedColumnsNotAllowed() {
+ void testCreateTableAsWithOverriddenComputedColumnsNotAllowed() {
CatalogTable catalogTable =
CatalogTable.newBuilder()
.schema(
@@ -231,7 +231,7 @@ public class SqlCTASNodeToOperationTest extends SqlNodeToOperationConversionTest
}
@Test
- public void testCreateTableAsWithPrimaryAndPartitionKey() {
+ void testCreateTableAsWithPrimaryAndPartitionKey() {
CatalogTable catalogTable =
CatalogTable.newBuilder()
.schema(
@@ -266,7 +266,7 @@ public class SqlCTASNodeToOperationTest extends SqlNodeToOperationConversionTest
}
@Test
- public void testCreateTableAsWithWatermark() {
+ void testCreateTableAsWithWatermark() {
CatalogTable catalogTable =
CatalogTable.newBuilder()
.schema(
@@ -301,7 +301,7 @@ public class SqlCTASNodeToOperationTest extends SqlNodeToOperationConversionTest
}
@Test
- public void testCreateTableAsWithNotNullColumnsAreNotAllowed() {
+ void testCreateTableAsWithNotNullColumnsAreNotAllowed() {
CatalogTable catalogTable =
CatalogTable.newBuilder()
.schema(
@@ -323,7 +323,7 @@ public class SqlCTASNodeToOperationTest extends SqlNodeToOperationConversionTest
}
@Test
- public void testCreateTableAsWithIncompatibleImplicitCastTypes() {
+ void testCreateTableAsWithIncompatibleImplicitCastTypes() {
CatalogTable catalogTable =
CatalogTable.newBuilder()
.schema(
@@ -348,7 +348,7 @@ public class SqlCTASNodeToOperationTest extends SqlNodeToOperationConversionTest
}
@Test
- public void testMergingCreateTableAsWithDistribution() {
+ void testMergingCreateTableAsWithDistribution() {
CatalogTable catalogTable =
CatalogTable.newBuilder()
.schema(
@@ -388,7 +388,7 @@ public class SqlCTASNodeToOperationTest extends SqlNodeToOperationConversionTest
}
@Test
- public void testMergingCreateTableAsWitEmptyDistribution() {
+ void testMergingCreateTableAsWitEmptyDistribution() {
CatalogTable catalogTable =
CatalogTable.newBuilder()
.schema(
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlDdlToOperationConverterTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlDdlToOperationConverterTest.java
index 6a465210929..1ddd3a15860 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlDdlToOperationConverterTest.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlDdlToOperationConverterTest.java
@@ -1406,7 +1406,7 @@ class SqlDdlToOperationConverterTest extends SqlNodeToOperationConversionTestBas
checkAlterNonExistTable("alter table %s nonexistent drop watermark");
}
- @ParameterizedTest(name = "[{index}] {0}")
+ @ParameterizedTest(name = "{index}: {0}")
@MethodSource("provideCreateMaterializedTableTestCases")
void createMaterializedTableWithVariousOptions(
String testName,
@@ -1429,7 +1429,7 @@ class SqlDdlToOperationConverterTest extends SqlNodeToOperationConversionTestBas
return Stream.of(
Arguments.of(
"with refresh mode continuous",
- "CREATE MATERIALIZED TABLE users_shops ("
+ "CREATE MATERIALIZED TABLE users_shops (shop_id int, user_id int,"
+ " PRIMARY KEY (user_id) not enforced)"
+ " WITH(\n"
+ " 'format' = 'debezium-json'\n"
@@ -1439,14 +1439,14 @@ class SqlDdlToOperationConverterTest extends SqlNodeToOperationConversionTestBas
+ " AS SELECT 1 as shop_id, 2 as user_id ",
"CREATE MATERIALIZED TABLE: (materializedTable: "
+ "[ResolvedCatalogMaterializedTable{origin=DefaultCatalogMaterializedTable{schema=(\n"
- + " `shop_id` INT NOT NULL,\n"
+ + " `shop_id` INT,\n"
+ " `user_id` INT NOT NULL,\n"
+ " CONSTRAINT `PK_user_id` PRIMARY KEY (`user_id`) NOT ENFORCED\n"
+ "), comment='null', distribution=null, partitionKeys=[], "
+ "options={format=debezium-json}, snapshot=null, definitionQuery='SELECT 1 AS `shop_id`, 2 AS `user_id`', "
+ "freshness=INTERVAL '30' SECOND, logicalRefreshMode=CONTINUOUS, refreshMode=CONTINUOUS, "
+ "refreshStatus=INITIALIZING, refreshHandlerDescription='null', serializedRefreshHandler=null}, resolvedSchema=(\n"
- + " `shop_id` INT NOT NULL,\n"
+ + " `shop_id` INT,\n"
+ " `user_id` INT NOT NULL,\n"
+ " CONSTRAINT `PK_user_id` PRIMARY KEY (`user_id`) NOT ENFORCED\n"
+ ")}], identifier: [`builtin`.`default`.`users_shops`])",
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlDmlToOperationConverterTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlDmlToOperationConverterTest.java
index afe84800fd3..89d2ee48822 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlDmlToOperationConverterTest.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlDmlToOperationConverterTest.java
@@ -53,34 +53,34 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.assertj.core.api.InstanceOfAssertFactories.type;
/** Test cases for the DML statements for {@link SqlNodeToOperationConversion}. */
-public class SqlDmlToOperationConverterTest extends SqlNodeToOperationConversionTestBase {
+class SqlDmlToOperationConverterTest extends SqlNodeToOperationConversionTestBase {
@Test
- public void testExplainWithSelect() {
+ void testExplainWithSelect() {
final String sql = "explain select * from t1";
checkExplainSql(sql);
}
@Test
- public void testExplainWithInsert() {
+ void testExplainWithInsert() {
final String sql = "explain insert into t2 select * from t1";
checkExplainSql(sql);
}
@Test
- public void testExplainWithUnion() {
+ void testExplainWithUnion() {
final String sql = "explain select * from t1 union select * from t2";
checkExplainSql(sql);
}
@Test
- public void testExplainWithExplainDetails() {
+ void testExplainWithExplainDetails() {
String sql = "explain changelog_mode, estimated_cost, json_execution_plan select * from t1";
checkExplainSql(sql);
}
@Test
- public void testSqlInsertWithStaticPartition() {
+ void testSqlInsertWithStaticPartition() {
final String sql = "insert into t1 partition(a=1) select b, c, d from t2";
FlinkPlannerImpl planner = getPlannerBySqlDialect(SqlDialect.DEFAULT);
final CalciteParser parser = getParserBySqlDialect(SqlDialect.DEFAULT);
@@ -93,7 +93,7 @@ public class SqlDmlToOperationConverterTest extends SqlNodeToOperationConversion
}
@Test
- public void testSqlInsertWithDynamicTableOptions() {
+ void testSqlInsertWithDynamicTableOptions() {
final String sql =
"insert into t1 /*+ OPTIONS('k1'='v1', 'k2'='v2') */\n"
+ "select a, b, c, d from t2";
@@ -109,7 +109,7 @@ public class SqlDmlToOperationConverterTest extends SqlNodeToOperationConversion
}
@Test
- public void testDynamicTableWithInvalidOptions() {
+ void testDynamicTableWithInvalidOptions() {
final String sql = "select * from t1 /*+ OPTIONS('opt1', 'opt2') */";
FlinkPlannerImpl planner = getPlannerBySqlDialect(SqlDialect.DEFAULT);
final CalciteParser parser = getParserBySqlDialect(SqlDialect.DEFAULT);
@@ -120,7 +120,7 @@ public class SqlDmlToOperationConverterTest extends SqlNodeToOperationConversion
}
@Test
- public void testBeginStatementSet() {
+ void testBeginStatementSet() {
final String sql = "BEGIN STATEMENT SET";
Operation operation = parse(sql);
assertThat(operation).isInstanceOf(BeginStatementSetOperation.class);
@@ -131,7 +131,7 @@ public class SqlDmlToOperationConverterTest extends SqlNodeToOperationConversion
}
@Test
- public void testEnd() {
+ void testEnd() {
final String sql = "END";
Operation operation = parse(sql);
assertThat(operation).isInstanceOf(EndStatementSetOperation.class);
@@ -142,7 +142,7 @@ public class SqlDmlToOperationConverterTest extends SqlNodeToOperationConversion
}
@Test
- public void testSqlRichExplainWithSelect() {
+ void testSqlRichExplainWithSelect() {
final String sql = "explain plan for select a, b, c, d from t2";
FlinkPlannerImpl planner = getPlannerBySqlDialect(SqlDialect.DEFAULT);
final CalciteParser parser = getParserBySqlDialect(SqlDialect.DEFAULT);
@@ -151,7 +151,7 @@ public class SqlDmlToOperationConverterTest extends SqlNodeToOperationConversion
}
@Test
- public void testSqlRichExplainWithInsert() {
+ void testSqlRichExplainWithInsert() {
final String sql = "explain plan for insert into t1 select a, b, c, d from t2";
FlinkPlannerImpl planner = getPlannerBySqlDialect(SqlDialect.DEFAULT);
final CalciteParser parser = getParserBySqlDialect(SqlDialect.DEFAULT);
@@ -160,7 +160,7 @@ public class SqlDmlToOperationConverterTest extends SqlNodeToOperationConversion
}
@Test
- public void testSqlRichExplainWithStatementSet() {
+ void testSqlRichExplainWithStatementSet() {
final String sql =
"explain plan for statement set begin "
+ "insert into t1 select a, b, c, d from t2 where a > 1;"
@@ -173,7 +173,7 @@ public class SqlDmlToOperationConverterTest extends SqlNodeToOperationConversion
}
@Test
- public void testExplainDetailsWithSelect() {
+ void testExplainDetailsWithSelect() {
final String sql =
"explain estimated_cost, changelog_mode, plan_advice select a, b, c, d from t2";
FlinkPlannerImpl planner = getPlannerBySqlDialect(SqlDialect.DEFAULT);
@@ -182,7 +182,7 @@ public class SqlDmlToOperationConverterTest extends SqlNodeToOperationConversion
}
@Test
- public void testExplainDetailsWithInsert() {
+ void testExplainDetailsWithInsert() {
final String sql =
"explain estimated_cost, changelog_mode, plan_advice insert into t1 select a, b, c, d from t2";
FlinkPlannerImpl planner = getPlannerBySqlDialect(SqlDialect.DEFAULT);
@@ -191,7 +191,7 @@ public class SqlDmlToOperationConverterTest extends SqlNodeToOperationConversion
}
@Test
- public void testExplainDetailsWithStatementSet() {
+ void testExplainDetailsWithStatementSet() {
final String sql =
"explain estimated_cost, changelog_mode, plan_advice statement set begin "
+ "insert into t1 select a, b, c, d from t2 where a > 1;"
@@ -215,7 +215,7 @@ public class SqlDmlToOperationConverterTest extends SqlNodeToOperationConversion
}
@Test
- public void testSqlExecuteWithStatementSet() {
+ void testSqlExecuteWithStatementSet() {
final String sql =
"execute statement set begin "
+ "insert into t1 select a, b, c, d from t2 where a > 1;"
@@ -228,7 +228,7 @@ public class SqlDmlToOperationConverterTest extends SqlNodeToOperationConversion
}
@Test
- public void testSqlRichExplainWithExecuteStatementSet() {
+ void testSqlRichExplainWithExecuteStatementSet() {
final String sql =
"EXPLAIN EXECUTE STATEMENT SET BEGIN "
+ "INSERT INTO t1 SELECT a, b, c, d FROM t2 WHERE a > 1;"
@@ -241,7 +241,7 @@ public class SqlDmlToOperationConverterTest extends SqlNodeToOperationConversion
}
@Test
- public void testSqlExecuteWithInsert() {
+ void testSqlExecuteWithInsert() {
final String sql = "execute insert into t1 select a, b, c, d from t2 where a > 1";
FlinkPlannerImpl planner = getPlannerBySqlDialect(SqlDialect.DEFAULT);
final CalciteParser parser = getParserBySqlDialect(SqlDialect.DEFAULT);
@@ -250,7 +250,7 @@ public class SqlDmlToOperationConverterTest extends SqlNodeToOperationConversion
}
@Test
- public void testSqlRichExplainWithExecuteInsert() {
+ void testSqlRichExplainWithExecuteInsert() {
final String sql = "EXPLAIN EXECUTE INSERT INTO t1 SELECT a, b, c, d FROM t2";
FlinkPlannerImpl planner = getPlannerBySqlDialect(SqlDialect.DEFAULT);
final CalciteParser parser = getParserBySqlDialect(SqlDialect.DEFAULT);
@@ -259,7 +259,7 @@ public class SqlDmlToOperationConverterTest extends SqlNodeToOperationConversion
}
@Test
- public void testSqlExecuteWithSelect() {
+ void testSqlExecuteWithSelect() {
final String sql = "execute select a, b, c, d from t2 where a > 1";
FlinkPlannerImpl planner = getPlannerBySqlDialect(SqlDialect.DEFAULT);
final CalciteParser parser = getParserBySqlDialect(SqlDialect.DEFAULT);
@@ -268,7 +268,7 @@ public class SqlDmlToOperationConverterTest extends SqlNodeToOperationConversion
}
@Test
- public void testSqlRichExplainWithExecuteSelect() {
+ void testSqlRichExplainWithExecuteSelect() {
final String sql = "EXPLAIN EXECUTE SELECT a, b, c, d FROM t2";
FlinkPlannerImpl planner = getPlannerBySqlDialect(SqlDialect.DEFAULT);
final CalciteParser parser = getParserBySqlDialect(SqlDialect.DEFAULT);
@@ -277,7 +277,7 @@ public class SqlDmlToOperationConverterTest extends SqlNodeToOperationConversion
}
@Test
- public void testDelete() throws Exception {
+ void testDelete() throws Exception {
Map<String, String> options = new HashMap<>();
options.put("connector", TestUpdateDeleteTableFactory.IDENTIFIER);
CatalogTable catalogTable =
@@ -311,7 +311,7 @@ public class SqlDmlToOperationConverterTest extends SqlNodeToOperationConversion
}
@Test
- public void testUpdate() throws Exception {
+ void testUpdate() throws Exception {
Map<String, String> options = new HashMap<>();
options.put("connector", TestUpdateDeleteTableFactory.IDENTIFIER);
CatalogTable catalogTable =
@@ -340,7 +340,7 @@ public class SqlDmlToOperationConverterTest extends SqlNodeToOperationConversion
}
@Test
- public void testTruncateTable() {
+ void testTruncateTable() {
String sql = "TRUNCATE TABLE t1";
FlinkPlannerImpl planner = getPlannerBySqlDialect(SqlDialect.DEFAULT);
final CalciteParser parser = getParserBySqlDialect(SqlDialect.DEFAULT);
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlMaterializedTableNodeToOperationConverterTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlMaterializedTableNodeToOperationConverterTest.java
index 67a828d7fe0..460d8a91c91 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlMaterializedTableNodeToOperationConverterTest.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlMaterializedTableNodeToOperationConverterTest.java
@@ -18,6 +18,7 @@
package org.apache.flink.table.planner.operations;
+import org.apache.flink.sql.parser.error.SqlValidateException;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.FunctionDescriptor;
import org.apache.flink.table.api.Schema;
@@ -34,6 +35,7 @@ import org.apache.flink.table.catalog.ResolvedCatalogMaterializedTable;
import org.apache.flink.table.catalog.ResolvedSchema;
import org.apache.flink.table.catalog.TableChange;
import org.apache.flink.table.catalog.UnresolvedIdentifier;
+import org.apache.flink.table.catalog.WatermarkSpec;
import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException;
import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
import org.apache.flink.table.catalog.exceptions.TableNotExistException;
@@ -50,8 +52,13 @@ import org.apache.flink.shaded.guava33.com.google.common.collect.ImmutableMap;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
@@ -62,11 +69,11 @@ import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Test for the materialized table statements for {@link SqlNodeToOperationConversion}. */
-public class SqlMaterializedTableNodeToOperationConverterTest
+class SqlMaterializedTableNodeToOperationConverterTest
extends SqlNodeToOperationConversionTestBase {
@BeforeEach
- public void before() throws TableAlreadyExistException, DatabaseNotExistException {
+ void before() throws TableAlreadyExistException, DatabaseNotExistException {
super.before();
final ObjectPath path3 = new ObjectPath(catalogManager.getCurrentDatabase(), "t3");
final Schema tableSchema =
@@ -284,7 +291,7 @@ public class SqlMaterializedTableNodeToOperationConverterTest
+ ")\n"
+ "FRESHNESS = INTERVAL '30' SECOND\n"
+ "REFRESH_MODE = FULL\n"
- + "AS SELECT a, f1, f2 FROM t1,LATERAL TABLE(myFunc(b)) as T(f1, f2)";
+ + "AS SELECT a, f1, f2 FROM t1, LATERAL TABLE(myFunc(b)) as T(f1, f2)";
Operation operation = parse(sql);
assertThat(operation).isInstanceOf(CreateMaterializedTableOperation.class);
@@ -397,9 +404,8 @@ public class SqlMaterializedTableNodeToOperationConverterTest
+ "AS SELECT * FROM t1";
assertThatThrownBy(() -> parse(sql))
- .isInstanceOf(ValidationException.class)
- .hasMessageContaining(
- "Primary key validation failed: UNIQUE constraint is not supported yet.");
+ .isInstanceOf(SqlValidateException.class)
+ .hasMessageContaining("UNIQUE constraint is not supported yet");
// test primary key not defined in source table
final String sql2 =
@@ -412,7 +418,7 @@ public class SqlMaterializedTableNodeToOperationConverterTest
assertThatThrownBy(() -> parse(sql2))
.isInstanceOf(ValidationException.class)
.hasMessageContaining(
- "Primary key column 'e' not defined in the query schema. Available columns: ['a', 'b', 'c', 'd'].");
+ "Primary key column 'e' is not defined in the schema at line 2, column 31");
// test primary key with nullable source column
final String sql3 =
@@ -424,7 +430,7 @@ public class SqlMaterializedTableNodeToOperationConverterTest
assertThatThrownBy(() -> parse(sql3))
.isInstanceOf(ValidationException.class)
- .hasMessageContaining("Could not create a PRIMARY KEY with nullable column 'd'.");
+ .hasMessageContaining("Invalid primary key 'ct1'. Column 'd' is nullable.");
}
@Test
@@ -438,7 +444,7 @@ public class SqlMaterializedTableNodeToOperationConverterTest
assertThatThrownBy(() -> parse(sql))
.isInstanceOf(ValidationException.class)
.hasMessageContaining(
- "Partition column 'e' not defined in the query schema. Available columns: ['a', 'b', 'c', 'd'].");
+ "Partition column 'e' not defined in the query schema. Available columns: ['a', 'b', 'c', 'd']");
final String sql2 =
"CREATE MATERIALIZED TABLE mtbl1\n"
@@ -516,6 +522,65 @@ public class SqlMaterializedTableNodeToOperationConverterTest
"Materialized table freshness only support SECOND, MINUTE, HOUR, DAY as the time unit.");
}
+ @ParameterizedTest
+ @MethodSource("testDataForCreateMaterializedTableFailedCase")
+ void createMaterializedTableFailedCase(String sql, String expectedErrorMsg) {
+ assertThatThrownBy(() -> parse(sql))
+ .isInstanceOf(ValidationException.class)
+ .hasMessage(expectedErrorMsg);
+ }
+
+ @ParameterizedTest
+ @MethodSource("testDataWithDifferentSchemasSuccessCase")
+ void createMaterializedTableSuccessCase(String sql, ResolvedSchema expected) {
+ CreateMaterializedTableOperation operation = (CreateMaterializedTableOperation) parse(sql);
+ assertThat(operation.getCatalogMaterializedTable().getResolvedSchema()).isEqualTo(expected);
+ }
+
+ @Test
+ void createMaterializedTableWithWatermark() {
+ final String sql =
+ "CREATE MATERIALIZED TABLE users_shops (watermark for ts as ts - interval '2' second)"
+ + " FRESHNESS = INTERVAL '30' SECOND"
+ + " AS SELECT 1 as shop_id, 2 as user_id, cast(current_timestamp as timestamp(3)) ts";
+ CreateMaterializedTableOperation operation = (CreateMaterializedTableOperation) parse(sql);
+ ResolvedSchema schema = operation.getCatalogMaterializedTable().getResolvedSchema();
+ assertThat(schema.getColumns())
+ .containsExactly(
+ Column.physical("shop_id", DataTypes.INT().notNull()),
+ Column.physical("user_id", DataTypes.INT().notNull()),
+ Column.physical("ts", DataTypes.TIMESTAMP(3).notNull()));
+
+ assertThat(schema.getWatermarkSpecs()).hasSize(1);
+ WatermarkSpec watermarkSpec = schema.getWatermarkSpecs().get(0);
+ assertThat(watermarkSpec.getWatermarkExpression())
+ .hasToString("`ts` - INTERVAL '2' SECOND");
+ assertThat(schema.getPrimaryKey()).isEmpty();
+ assertThat(schema.getIndexes()).isEmpty();
+ }
+
+ @Test
+ void createMaterializedTableWithWatermarkUsingColumnFromCreatePart() {
+ final String sql =
+ "CREATE MATERIALIZED TABLE users_shops (ts TIMESTAMP(3), watermark for ts as ts - interval '2' second)"
+ + " FRESHNESS = INTERVAL '30' SECOND"
+ + " AS SELECT current_timestamp() as ts, 1 as shop_id, 2 as user_id";
+ CreateMaterializedTableOperation operation = (CreateMaterializedTableOperation) parse(sql);
+ ResolvedSchema schema = operation.getCatalogMaterializedTable().getResolvedSchema();
+ assertThat(schema.getColumns())
+ .containsExactly(
+ Column.physical("ts", DataTypes.TIMESTAMP(3)),
+ Column.physical("shop_id", DataTypes.INT().notNull()),
+ Column.physical("user_id", DataTypes.INT().notNull()));
+
+ assertThat(schema.getWatermarkSpecs()).hasSize(1);
+ WatermarkSpec watermarkSpec = schema.getWatermarkSpecs().get(0);
+ assertThat(watermarkSpec.getWatermarkExpression())
+ .hasToString("`ts` - INTERVAL '2' SECOND");
+ assertThat(schema.getPrimaryKey()).isEmpty();
+ assertThat(schema.getIndexes()).isEmpty();
+ }
+
@Test
void testAlterMaterializedTableRefreshOperationWithPartitionSpec() {
final String sql =
@@ -531,7 +596,7 @@ public class SqlMaterializedTableNodeToOperationConverterTest
}
@Test
- public void testAlterMaterializedTableRefreshOperationWithoutPartitionSpec() {
+ void testAlterMaterializedTableRefreshOperationWithoutPartitionSpec() {
final String sql = "ALTER MATERIALIZED TABLE mtbl1 REFRESH";
Operation operation = parse(sql);
@@ -644,49 +709,6 @@ public class SqlMaterializedTableNodeToOperationConverterTest
+ "FROM `builtin`.`default`.`t3` AS `t3`")));
}
- @Test
- void testAlterMaterializedTableAsQueryWithUnsupportedColumnChange() {
- // 1. delete existing column
- String sql1 = "ALTER MATERIALIZED TABLE base_mtbl AS SELECT a, b FROM t3";
- assertThatThrownBy(() -> parse(sql1))
- .isInstanceOf(ValidationException.class)
- .hasMessageContaining(
- "Failed to modify query because drop column is unsupported. When modifying a query, you can only append new columns at the end of original schema. The original schema has 4 columns, but the newly derived schema from the query has 2 columns.");
- // 2. swap column position
- String sql2 = "ALTER MATERIALIZED TABLE base_mtbl AS SELECT a, b, d, c FROM t3";
- assertThatThrownBy(() -> parse(sql2))
- .isInstanceOf(ValidationException.class)
- .hasMessageContaining(
- "When modifying the query of a materialized table, currently only support appending columns at the end of original schema, dropping, renaming, and reordering columns are not supported.\n"
- + "Column mismatch at position 2: Original column is [`c` INT], but new column is [`d` STRING].");
- // 3. change existing column type
- String sql3 =
- "ALTER MATERIALIZED TABLE base_mtbl AS SELECT a, b, c, cast(d as int) as d FROM t3";
- assertThatThrownBy(() -> parse(sql3))
- .isInstanceOf(ValidationException.class)
- .hasMessageContaining(
- "When modifying the query of a materialized table, currently only support appending columns at the end of original schema, dropping, renaming, and reordering columns are not supported.\n"
- + "Column mismatch at position 3: Original column is [`d` STRING], but new column is [`d` INT].");
- // 4. change existing column nullability
- String sql4 =
- "ALTER MATERIALIZED TABLE base_mtbl AS SELECT a, b, c, cast('d' as string) as d FROM t3";
- assertThatThrownBy(() -> parse(sql4))
- .isInstanceOf(ValidationException.class)
- .hasMessageContaining(
- "When modifying the query of a materialized table, currently only support appending columns at the end of original schema, dropping, renaming, and reordering columns are not supported.\n"
- + "Column mismatch at position 3: Original column is [`d` STRING], but new column is [`d` STRING NOT NULL].");
- }
-
- @Test
- void testAlterAlterMaterializedTableAsQueryWithCatalogTable() {
- // t1 is a CatalogTable not a Materialized Table
- final String sql = "ALTER MATERIALIZED TABLE t1 AS SELECT * FROM t1";
- assertThatThrownBy(() -> parse(sql))
- .isInstanceOf(ValidationException.class)
- .hasMessageContaining(
- "Only materialized tables support modifying the definition query.");
- }
-
@Test
void testDropMaterializedTable() {
final String sql = "DROP MATERIALIZED TABLE mtbl1";
@@ -706,4 +728,153 @@ public class SqlMaterializedTableNodeToOperationConverterTest
.isEqualTo(
"DROP MATERIALIZED TABLE: (identifier: [`builtin`.`default`.`mtbl1`], IfExists: [true])");
}
+
+ private static Collection<Arguments> testDataForCreateMaterializedTableFailedCase() {
+ final Collection<Arguments> list = new ArrayList<>();
+ list.addAll(create());
+ list.addAll(alter());
+ return list;
+ }
+
+ private static Collection<Arguments> alter() {
+ return List.of(
+ Arguments.of(
+ "ALTER MATERIALIZED TABLE base_mtbl AS SELECT a, b FROM t3",
+ "Failed to modify query because drop column is unsupported. When modifying "
+ + "a query, you can only append new columns at the end of original "
+ + "schema. The original schema has 4 columns, but the newly derived "
+ + "schema from the query has 2 columns."),
+ Arguments.of(
+ "ALTER MATERIALIZED TABLE base_mtbl AS SELECT a, b, d, c FROM t3",
+ "When modifying the query of a materialized table, currently only support "
+ + "appending columns at the end of original schema, dropping, "
+ + "renaming, and reordering columns are not supported.\n"
+ + "Column mismatch at position 2: Original column is [`c` INT], "
+ + "but new column is [`d` STRING]."),
+ Arguments.of(
+ "ALTER MATERIALIZED TABLE base_mtbl AS SELECT a, b, c, CAST(d AS INT) AS d FROM t3",
+ "When modifying the query of a materialized table, currently only support "
+ + "appending columns at the end of original schema, dropping, "
+ + "renaming, and reordering columns are not supported.\n"
+ + "Column mismatch at position 3: Original column is [`d` STRING], "
+ + "but new column is [`d` INT]."),
+ Arguments.of(
+ "ALTER MATERIALIZED TABLE base_mtbl AS SELECT a, b, c, CAST('d' AS STRING) AS d FROM t3",
+ "When modifying the query of a materialized table, currently only support "
+ + "appending columns at the end of original schema, dropping, "
+ + "renaming, and reordering columns are not supported.\n"
+ + "Column mismatch at position 3: Original column is [`d` STRING], "
+ + "but new column is [`d` STRING NOT NULL]."),
+ Arguments.of(
+ "ALTER MATERIALIZED TABLE t1 AS SELECT * FROM t1",
+ "Only materialized tables support modifying the definition query."));
+ }
+
+ private static Collection<Arguments> create() {
+ return List.of(
+ Arguments.of(
+ "CREATE MATERIALIZED TABLE users_shops (shop_id)"
+ + " FRESHNESS = INTERVAL '30' SECOND"
+ + " AS SELECT 1 AS shop_id, 2 AS user_id",
+ "The number of columns in the column list must match the number of columns in the source schema."),
+ Arguments.of(
+ "CREATE MATERIALIZED TABLE users_shops (id, name, address)"
+ + " FRESHNESS = INTERVAL '30' SECOND"
+ + " AS SELECT 1 AS shop_id, 2 AS user_id",
+ "The number of columns in the column list must match the number of columns in the source schema."),
+ Arguments.of(
+ "CREATE MATERIALIZED TABLE users_shops (id, name)"
+ + " FRESHNESS = INTERVAL '30' SECOND"
+ + " AS SELECT 1 AS shop_id, 2 AS user_id",
+ "Column 'id' not found in the source schema."),
+ Arguments.of(
+ "CREATE MATERIALIZED TABLE users_shops (shop_id STRING, user_id STRING)"
+ + " FRESHNESS = INTERVAL '30' SECOND"
+ + " AS SELECT 1 AS shop_id, 2 AS user_id",
+ "Incompatible types for sink column 'shop_id' at position 0. The source column has type 'INT NOT NULL', "
+ + "while the target column has type 'STRING'."),
+ Arguments.of(
+ "CREATE MATERIALIZED TABLE users_shops (shop_id INT, WATERMARK FOR ts AS `ts` - INTERVAL '5' SECOND)"
+ + " FRESHNESS = INTERVAL '30' SECOND"
+ + " AS SELECT 1 AS shop_id, 2 AS user_id",
+ "The rowtime attribute field 'ts' is not defined in the table schema, at line 1, column 67\n"
+ + "Available fields: ['shop_id', 'user_id']"),
+ Arguments.of(
+ "CREATE MATERIALIZED TABLE users_shops (shop_id INT, user_id INT, PRIMARY KEY(id) NOT ENFORCED)"
+ + " FRESHNESS = INTERVAL '30' SECOND"
+ + " AS SELECT 1 AS shop_id, 2 AS user_id",
+ "Primary key column 'id' is not defined in the schema at line 1, column 78"),
+ Arguments.of(
+ "CREATE MATERIALIZED TABLE users_shops (WATERMARK FOR ts AS ts - INTERVAL '2' SECOND)"
+ + " FRESHNESS = INTERVAL '30' SECOND"
+ + " AS SELECT 1 AS shop_id, 2 AS user_id",
+ "The rowtime attribute field 'ts' is not defined in the table schema, at line 1, column 54\n"
+ + "Available fields: ['shop_id', 'user_id']"),
+ Arguments.of(
+ "CREATE MATERIALIZED TABLE users_shops (a INT, b INT)"
+ + " FRESHNESS = INTERVAL '30' SECOND"
+ + " AS SELECT 1 AS shop_id, 2 AS user_id",
+ "Invalid as physical column 'a' is defined in the DDL, but is not used in a query column."),
+ Arguments.of(
+ "CREATE MATERIALIZED TABLE users_shops (shop_id INT, b INT)"
+ + " FRESHNESS = INTERVAL '30' SECOND"
+ + " AS SELECT 1 AS shop_id, 2 AS user_id",
+ "Invalid as physical column 'b' is defined in the DDL, but is not used in a query column."));
+ }
+
+ private static Collection<Arguments> testDataWithDifferentSchemasSuccessCase() {
+ return List.of(
+ Arguments.of(
+ "CREATE MATERIALIZED TABLE users_shops (shop_id, user_id)"
+ + " FRESHNESS = INTERVAL '30' SECOND"
+ + " AS SELECT 1 AS shop_id, 2 AS user_id",
+ ResolvedSchema.of(
+ Column.physical("shop_id", DataTypes.INT().notNull()),
+ Column.physical("user_id", DataTypes.INT().notNull()))),
+ Arguments.of(
+ "CREATE MATERIALIZED TABLE users_shops"
+ + " FRESHNESS = INTERVAL '30' SECOND"
+ + " AS SELECT CAST(1 AS DOUBLE) AS shop_id, CAST(2 AS STRING) AS user_id",
+ ResolvedSchema.of(
+ Column.physical("shop_id", DataTypes.DOUBLE().notNull()),
+ Column.physical("user_id", DataTypes.STRING().notNull()))),
+ Arguments.of(
+ "CREATE MATERIALIZED TABLE users_shops (user_id, shop_id)"
+ + " FRESHNESS = INTERVAL '30' SECOND"
+ + " AS SELECT 1 AS shop_id, 2 AS user_id",
+ ResolvedSchema.of(
+ Column.physical("user_id", DataTypes.INT().notNull()),
+ Column.physical("shop_id", DataTypes.INT().notNull()))),
+ Arguments.of(
+ "CREATE MATERIALIZED TABLE users_shops (user_id INT, shop_id BIGINT)"
+ + " FRESHNESS = INTERVAL '30' SECOND"
+ + " AS SELECT 1 AS shop_id, 2 AS user_id",
+ ResolvedSchema.of(
+ Column.physical("shop_id", DataTypes.BIGINT()),
+ Column.physical("user_id", DataTypes.INT()))),
+ Arguments.of(
+ "CREATE MATERIALIZED TABLE users_shops (user_id INT, shop_id BIGINT, PRIMARY KEY(user_id) NOT ENFORCED)"
+ + " FRESHNESS = INTERVAL '30' SECOND"
+ + " AS SELECT 1 AS shop_id, 2 AS user_id",
+ new ResolvedSchema(
+ List.of(
+ Column.physical("shop_id", DataTypes.BIGINT()),
+ Column.physical("user_id", DataTypes.INT().notNull())),
+ List.of(),
+ org.apache.flink.table.catalog.UniqueConstraint.primaryKey(
+ "PK_user_id", List.of("user_id")),
+ List.of())),
+ Arguments.of(
+ "CREATE MATERIALIZED TABLE users_shops (PRIMARY KEY(user_id) NOT ENFORCED)"
+ + " FRESHNESS = INTERVAL '30' SECOND"
+ + " AS SELECT 1 AS shop_id, 2 AS user_id",
+ new ResolvedSchema(
+ List.of(
+ Column.physical("shop_id", DataTypes.INT().notNull()),
+ Column.physical("user_id", DataTypes.INT().notNull())),
+ List.of(),
+ org.apache.flink.table.catalog.UniqueConstraint.primaryKey(
+ "PK_user_id", List.of("user_id")),
+ List.of())));
+ }
}
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlModelOperationConverterTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlModelOperationConverterTest.java
index 53f45f58d0c..87bc54a5af1 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlModelOperationConverterTest.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlModelOperationConverterTest.java
@@ -58,9 +58,9 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.junit.jupiter.api.Assertions.assertNotNull;
/** Test for testing convert model statement to operation. */
-public class SqlModelOperationConverterTest extends SqlNodeToOperationConversionTestBase {
+class SqlModelOperationConverterTest extends SqlNodeToOperationConversionTestBase {
@Test
- public void testCreateModel() {
+ void testCreateModel() {
final String sql =
"CREATE MODEL model1 \n"
+ "INPUT(a bigint comment 'column a', b varchar, c int, d varchar)\n"
@@ -102,7 +102,7 @@ public class SqlModelOperationConverterTest extends SqlNodeToOperationConversion
}
@Test
- public void testDropModel() throws Exception {
+ void testDropModel() throws Exception {
Catalog catalog = new GenericInMemoryCatalog("default", "default");
if (!catalogManager.getCatalog("cat1").isPresent()) {
catalogManager.registerCatalog("cat1", catalog);
@@ -135,7 +135,7 @@ public class SqlModelOperationConverterTest extends SqlNodeToOperationConversion
}
@Test
- public void testDescribeModel() {
+ void testDescribeModel() {
Operation operation = parse("DESCRIBE MODEL m1");
assertThat(operation).isInstanceOf(DescribeModelOperation.class);
DescribeModelOperation describeModelOperation = (DescribeModelOperation) operation;
@@ -202,7 +202,7 @@ public class SqlModelOperationConverterTest extends SqlNodeToOperationConversion
}
@Test
- public void testAlterModel() throws Exception {
+ void testAlterModel() throws Exception {
Catalog catalog = new GenericInMemoryCatalog("default", "default");
if (!catalogManager.getCatalog("cat1").isPresent()) {
catalogManager.registerCatalog("cat1", catalog);
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlNodeToCallOperationTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlNodeToCallOperationTest.java
index a15ac7fc71b..c47d2431080 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlNodeToCallOperationTest.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlNodeToCallOperationTest.java
@@ -42,10 +42,10 @@ import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Test cases for the call statements for {@link SqlNodeToOperationConversion}. */
-public class SqlNodeToCallOperationTest extends SqlNodeToOperationConversionTestBase {
+class SqlNodeToCallOperationTest extends SqlNodeToOperationConversionTestBase {
@BeforeEach
- public void before() {
+ void before() {
CatalogWithBuiltInProcedure procedureCatalog =
new CatalogWithBuiltInProcedure("procedure_catalog");
catalogManager.registerCatalog("p1", procedureCatalog);
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlNodeToOperationConversionTestBase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlNodeToOperationConversionTestBase.java
index ef251378a5a..9ed19411a7a 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlNodeToOperationConversionTestBase.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlNodeToOperationConversionTestBase.java
@@ -59,7 +59,7 @@ import java.util.function.Supplier;
import static org.apache.calcite.jdbc.CalciteSchemaBuilder.asRootSchema;
/** Test base for testing convert sql statement to operation. */
-public class SqlNodeToOperationConversionTestBase {
+class SqlNodeToOperationConversionTestBase {
private final boolean isStreamingMode = false;
private final TableConfig tableConfig = TableConfig.getDefault();
protected final Catalog catalog = new GenericInMemoryCatalog("MockCatalog", "default");
@@ -96,7 +96,7 @@ public class SqlNodeToOperationConversionTestBase {
plannerContext.getRexFactory());
@BeforeEach
- public void before() throws TableAlreadyExistException, DatabaseNotExistException {
+ void before() throws TableAlreadyExistException, DatabaseNotExistException {
catalogManager.initSchemaResolver(
isStreamingMode,
ExpressionResolverMocks.basicResolver(catalogManager, functionCatalog, parser),
@@ -122,7 +122,7 @@ public class SqlNodeToOperationConversionTestBase {
}
@AfterEach
- public void after() throws TableNotExistException {
+ void after() throws TableNotExistException {
final ObjectPath path1 = new ObjectPath(catalogManager.getCurrentDatabase(), "t1");
final ObjectPath path2 = new ObjectPath(catalogManager.getCurrentDatabase(), "t2");
catalog.dropTable(path1, true);
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlRTASNodeToOperationConverterTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlRTASNodeToOperationConverterTest.java
index 40114de579f..dba843b5eb7 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlRTASNodeToOperationConverterTest.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlRTASNodeToOperationConverterTest.java
@@ -46,10 +46,10 @@ import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Test base for testing convert [CREATE OR] REPLACE TABLE AS statement to operation. */
-public class SqlRTASNodeToOperationConverterTest extends SqlNodeToOperationConversionTestBase {
+class SqlRTASNodeToOperationConverterTest extends SqlNodeToOperationConversionTestBase {
@Test
- public void testReplaceTableAs() {
+ void testReplaceTableAs() {
String tableName = "replace_table";
String tableComment = "test table comment 表描述";
String sql =
@@ -62,7 +62,7 @@ public class SqlRTASNodeToOperationConverterTest extends SqlNodeToOperationConve
}
@Test
- public void testReplaceTableAsWithOrderingColumns() {
+ void testReplaceTableAsWithOrderingColumns() {
String tableName = "replace_table";
String sql =
"REPLACE TABLE "
@@ -78,7 +78,7 @@ public class SqlRTASNodeToOperationConverterTest extends SqlNodeToOperationConve
}
@Test
- public void testReplaceTableAsWithNotFoundColumnIdentifiers() {
+ void testReplaceTableAsWithNotFoundColumnIdentifiers() {
String tableName = "replace_table";
String sql =
"REPLACE TABLE "
@@ -91,7 +91,7 @@ public class SqlRTASNodeToOperationConverterTest extends SqlNodeToOperationConve
}
@Test
- public void testReplaceTableAsWithMismatchIdentifiersLength() {
+ void testReplaceTableAsWithMismatchIdentifiersLength() {
String tableName = "replace_table";
String sql =
"REPLACE TABLE "
@@ -106,7 +106,7 @@ public class SqlRTASNodeToOperationConverterTest extends SqlNodeToOperationConve
}
@Test
- public void testCreateOrReplaceTableAs() {
+ void testCreateOrReplaceTableAs() {
String tableName = "create_or_replace_table";
String sql =
"CREATE OR REPLACE TABLE "
@@ -116,7 +116,7 @@ public class SqlRTASNodeToOperationConverterTest extends SqlNodeToOperationConve
}
@Test
- public void testCreateOrReplaceTableAsWithColumns() {
+ void testCreateOrReplaceTableAsWithColumns() {
String tableName = "create_or_replace_table";
String sql =
"CREATE OR REPLACE TABLE "
@@ -136,7 +136,7 @@ public class SqlRTASNodeToOperationConverterTest extends SqlNodeToOperationConve
}
@Test
- public void testCreateOrReplaceTableAsWithColumnsOverridden() {
+ void testCreateOrReplaceTableAsWithColumnsOverridden() {
String tableName = "create_or_replace_table";
String sql =
"CREATE OR REPLACE TABLE "
@@ -157,7 +157,7 @@ public class SqlRTASNodeToOperationConverterTest extends SqlNodeToOperationConve
}
@Test
- public void testCreateOrReplaceTableAsWithNotNullColumnsAreNotAllowed() {
+ void testCreateOrReplaceTableAsWithNotNullColumnsAreNotAllowed() {
String tableName = "create_or_replace_table";
String sql =
"CREATE OR REPLACE TABLE "
@@ -171,7 +171,7 @@ public class SqlRTASNodeToOperationConverterTest extends SqlNodeToOperationConve
}
@Test
- public void testCreateOrReplaceTableAsWithOverriddenVirtualMetadataColumnsNotAllowed() {
+ void testCreateOrReplaceTableAsWithOverriddenVirtualMetadataColumnsNotAllowed() {
String tableName = "create_or_replace_table";
String sql =
"CREATE OR REPLACE TABLE "
@@ -188,7 +188,7 @@ public class SqlRTASNodeToOperationConverterTest extends SqlNodeToOperationConve
}
@Test
- public void testCreateOrReplaceTableAsWithOverriddenComputedColumnsNotAllowed() {
+ void testCreateOrReplaceTableAsWithOverriddenComputedColumnsNotAllowed() {
String tableName = "create_or_replace_table";
String sql =
"CREATE OR REPLACE TABLE "
@@ -204,7 +204,7 @@ public class SqlRTASNodeToOperationConverterTest extends SqlNodeToOperationConve
}
@Test
- public void testCreateOrReplaceTableAsWithIncompatibleImplicitCastTypes() {
+ void testCreateOrReplaceTableAsWithIncompatibleImplicitCastTypes() {
String tableName = "create_or_replace_table";
String sql =
"CREATE OR REPLACE TABLE "
@@ -221,7 +221,7 @@ public class SqlRTASNodeToOperationConverterTest extends SqlNodeToOperationConve
}
@Test
- public void testCreateOrReplaceTableAsWithDistribution() {
+ void testCreateOrReplaceTableAsWithDistribution() {
String tableName = "create_or_replace_table";
String sql =
"CREATE OR REPLACE TABLE "
@@ -240,7 +240,7 @@ public class SqlRTASNodeToOperationConverterTest extends SqlNodeToOperationConve
}
@Test
- public void testCreateOrReplaceTableAsWithPrimaryKey() {
+ void testCreateOrReplaceTableAsWithPrimaryKey() {
String tableName = "create_or_replace_table";
String sql =
"CREATE OR REPLACE TABLE "
@@ -260,7 +260,7 @@ public class SqlRTASNodeToOperationConverterTest extends SqlNodeToOperationConve
}
@Test
- public void testCreateOrReplaceTableAsWithWatermark() {
+ void testCreateOrReplaceTableAsWithWatermark() {
String tableName = "create_or_replace_table";
String sql =
"CREATE OR REPLACE TABLE "
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlShowToOperationConverterTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlShowToOperationConverterTest.java
index 60606aa9771..baab04a471e 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlShowToOperationConverterTest.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlShowToOperationConverterTest.java
@@ -27,17 +27,17 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
-public class SqlShowToOperationConverterTest extends SqlNodeToOperationConversionTestBase {
+class SqlShowToOperationConverterTest extends SqlNodeToOperationConversionTestBase {
@BeforeEach
- public void before() throws TableAlreadyExistException, DatabaseNotExistException {
+ void before() throws TableAlreadyExistException, DatabaseNotExistException {
// Do nothing
// No need to create schema, tables and etc. since the test executes for unset catalog and
// database
}
@AfterEach
- public void after() throws TableNotExistException {
+ void after() throws TableNotExistException {
// Do nothing
}