You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by sh...@apache.org on 2023/12/19 12:08:22 UTC
(flink) branch pull/23760/head created (now d68e775022d)
This is an automated email from the ASF dual-hosted git repository.
shengkai pushed a change to branch pull/23760/head
in repository https://gitbox.apache.org/repos/asf/flink.git
at d68e775022d [FLINK-33490][table-planner] Validate the name conflicts when creating view
This branch includes the following new commits:
new d68e775022d [FLINK-33490][table-planner] Validate the name conflicts when creating view
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(flink) 01/01: [FLINK-33490][table-planner] Validate the name conflicts when creating view
Posted by sh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
shengkai pushed a commit to branch pull/23760/head
in repository https://gitbox.apache.org/repos/asf/flink.git
commit d68e775022d9c8f52e8c61b0b7c34bbf2ba30baa
Author: xuyang <xy...@163.com>
AuthorDate: Mon Nov 20 19:43:39 2023 +0800
[FLINK-33490][table-planner] Validate the name conflicts when creating view
---
.../operations/converters/SqlNodeConvertUtils.java | 69 ++++++++++++++++++++++
.../operations/SqlDdlToOperationConverterTest.java | 52 ++++++++++++++++
2 files changed, 121 insertions(+)
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlNodeConvertUtils.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlNodeConvertUtils.java
index 28d030af707..727b136123f 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlNodeConvertUtils.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlNodeConvertUtils.java
@@ -19,6 +19,7 @@
package org.apache.flink.table.planner.operations.converters;
import org.apache.flink.sql.parser.ddl.SqlAlterView;
+import org.apache.flink.sql.parser.error.SqlValidateException;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.ValidationException;
import org.apache.flink.table.catalog.CatalogBaseTable;
@@ -32,10 +33,20 @@ import org.apache.flink.table.planner.operations.PlannerQueryOperation;
import org.apache.flink.table.planner.operations.converters.SqlNodeConverter.ConvertContext;
import org.apache.calcite.rel.RelRoot;
+import org.apache.calcite.sql.SqlBasicCall;
import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlSelect;
+import org.apache.calcite.sql.SqlSetOperator;
+import org.apache.calcite.sql.SqlWith;
+import org.apache.calcite.sql.parser.SqlParserPos;
+import org.apache.calcite.sql.validate.SqlValidatorNamespace;
+import javax.annotation.Nullable;
+
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.Optional;
import java.util.stream.Collectors;
@@ -63,6 +74,14 @@ class SqlNodeConvertUtils {
// This bug is fixed in CALCITE-3877 of Calcite 1.23.0.
String originalQuery = context.toQuotedSqlString(query);
SqlNode validateQuery = context.getSqlValidator().validate(query);
+
+ // Check name is unique.
+ // Don't rely on the calcite because if the field names are duplicate, calcite will add
+ // index to identify the duplicate names.
+ SqlValidatorNamespace validatedNamespace =
+ context.getSqlValidator().getNamespace(validateQuery);
+ validateDuplicatedColumnNames(query, validatedNamespace);
+
// The LATERAL operator was eliminated during sql validation, thus the unparsed SQL
// does not contain LATERAL which is problematic,
// the issue was resolved in CALCITE-4077
@@ -120,4 +139,54 @@ class SqlNodeConvertUtils {
}
return (CatalogView) baseTable;
}
+
+ private static void validateDuplicatedColumnNames(
+ SqlNode query, SqlValidatorNamespace namespace) {
+ Map<String, Integer> nameToPos = new HashMap<>();
+ for (int i = 0;
+ i < Objects.requireNonNull(namespace).getType().getFieldList().size();
+ i++) {
+ String columnName = namespace.getType().getFieldList().get(i).getName();
+ if (nameToPos.containsKey(columnName)) {
+ SqlSelect select = extractSelect(query);
+ // Can not get the origin schema.
+ if (select == null) {
+ throw new ValidationException(
+ String.format(
+ "SQL validation failed. Column `%s` has been specified.",
+ columnName));
+ }
+ SqlParserPos errorPos = select.getSelectList().get(i).getParserPosition();
+ String msg =
+ String.format(
+ "A column with the same name `%s` has been defined at %s.",
+ columnName,
+ select.getSelectList()
+ .get(nameToPos.get(columnName))
+ .getParserPosition());
+ throw new ValidationException(
+ "SQL validation failed. " + msg, new SqlValidateException(errorPos, msg));
+ }
+ nameToPos.put(columnName, i);
+ }
+ }
+
+ private static @Nullable SqlSelect extractSelect(SqlNode query) {
+ if (query instanceof SqlSelect) {
+ return (SqlSelect) query;
+ } else if (query instanceof SqlBasicCall) {
+ SqlBasicCall call = (SqlBasicCall) query;
+ if (call.getOperator() instanceof SqlSetOperator) {
+ // UNION/INTERSECT/EXCEPT/...
+ return extractSelect(call.getOperandList().get(0));
+ } else {
+ return null;
+ }
+ } else if (query instanceof SqlWith) {
+ SqlWith with = (SqlWith) query;
+ return extractSelect(with.body);
+ } else {
+ return null;
+ }
+ }
}
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 289804fd2c2..1de2d72583a 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
@@ -18,6 +18,7 @@
package org.apache.flink.table.planner.operations;
+import org.apache.flink.core.testutils.FlinkAssertions;
import org.apache.flink.sql.parser.ddl.SqlCreateTable;
import org.apache.flink.sql.parser.error.SqlValidateException;
import org.apache.flink.table.api.DataTypes;
@@ -2210,6 +2211,57 @@ public class SqlDdlToOperationConverterTest extends SqlNodeToOperationConversion
"ALTER TABLE cat1.db1.tb1 DROP IF EXISTS PARTITION (b=1, c=2) PARTITION (b=2)");
}
+ @Test
+ public void testCreateViewWithDuplicateFieldName() {
+ Map<String, String> prop = new HashMap<>();
+ prop.put("connector", "values");
+ prop.put("bounded", "true");
+ CatalogTable catalogTable =
+ CatalogTable.of(
+ Schema.newBuilder()
+ .column("id", DataTypes.BIGINT().notNull())
+ .column("uid", DataTypes.BIGINT().notNull())
+ .build(),
+ null,
+ Collections.emptyList(),
+ prop);
+
+ catalogManager.createTable(
+ catalogTable, ObjectIdentifier.of("builtin", "default", "id_table"), false);
+
+ assertThatThrownBy(
+ () -> parse("CREATE VIEW id_view AS\nSELECT id, uid AS id FROM id_table"))
+ .satisfies(
+ FlinkAssertions.anyCauseMatches(
+ SqlValidateException.class,
+ "A column with the same name `id` has been defined at line 2, column 8."));
+
+ assertThatThrownBy(
+ () ->
+ parse(
+ "CREATE VIEW union_view AS\n"
+ + " SELECT id, uid AS id FROM id_table\n"
+ + " UNION\n"
+ + " SELECT uid, id AS uid FROM id_table"))
+ .satisfies(
+ FlinkAssertions.anyCauseMatches(
+ SqlValidateException.class,
+ "A column with the same name `id` has been defined at line 2, column 10."));
+ assertThatThrownBy(
+ () ->
+ parse(
+ "CREATE VIEW cte_view AS\n"
+ + "WITH id_num AS (\n"
+ + " select id from id_table\n"
+ + ")\n"
+ + "SELECT id, uid as id\n"
+ + "FROM id_table\n"))
+ .satisfies(
+ FlinkAssertions.anyCauseMatches(
+ SqlValidateException.class,
+ "A column with the same name `id` has been defined at line 5, column 8."));
+ }
+
// ~ Tool Methods ----------------------------------------------------------
private static TestItem createTestItem(Object... args) {