You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ja...@apache.org on 2021/01/15 14:40:40 UTC
[flink] branch master updated: [FLINK-20937][table-api] Fix
exception when DROP a table with illegal watermark definition
This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new de19a27 [FLINK-20937][table-api] Fix exception when DROP a table with illegal watermark definition
de19a27 is described below
commit de19a27fc58688712c86079333e3af3a6ccaa235
Author: zhuxiaoshang <31...@users.noreply.github.com>
AuthorDate: Fri Jan 15 22:40:24 2021 +0800
[FLINK-20937][table-api] Fix exception when DROP a table with illegal watermark definition
This closes #14640
Co-authored-by: zhuxiaoshang <zh...@qutoutiao.net>
---
.../apache/flink/table/catalog/CatalogManager.java | 14 ++++++++----
.../table/planner/catalog/CatalogTableITCase.scala | 26 ++++++++++++++++++++--
2 files changed, 34 insertions(+), 6 deletions(-)
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/CatalogManager.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/CatalogManager.java
index f475ba5..c7c71f2 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/CatalogManager.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/CatalogManager.java
@@ -375,7 +375,7 @@ public final class CatalogManager {
TableSchema resolvedSchema = resolveTableSchema(temporaryTable);
return Optional.of(TableLookupResult.temporary(temporaryTable, resolvedSchema));
} else {
- return getPermanentTable(objectIdentifier);
+ return getPermanentTable(objectIdentifier, true);
}
}
@@ -404,13 +404,17 @@ public final class CatalogManager {
return Optional.empty();
}
- private Optional<TableLookupResult> getPermanentTable(ObjectIdentifier objectIdentifier) {
+ private Optional<TableLookupResult> getPermanentTable(
+ ObjectIdentifier objectIdentifier, boolean needResolveSchema) {
Catalog currentCatalog = catalogs.get(objectIdentifier.getCatalogName());
ObjectPath objectPath = objectIdentifier.toObjectPath();
if (currentCatalog != null) {
try {
CatalogBaseTable catalogTable = currentCatalog.getTable(objectPath);
- TableSchema resolvedSchema = resolveTableSchema(catalogTable);
+ TableSchema resolvedSchema =
+ needResolveSchema
+ ? resolveTableSchema(catalogTable)
+ : catalogTable.getSchema();
return Optional.of(TableLookupResult.permanent(catalogTable, resolvedSchema));
} catch (TableNotExistException e) {
// Ignore.
@@ -755,7 +759,9 @@ public final class CatalogManager {
+ "Drop it first before removing the permanent %s.",
tableOrView, objectIdentifier, tableOrView));
}
- final Optional<TableLookupResult> resultOpt = getPermanentTable(objectIdentifier);
+ // we can't resolve the schema here, because the schema might be illegal and DROP TABLE
+ // doesn't care about the schema, so we skip schema resolve here, see FLINK-20937
+ final Optional<TableLookupResult> resultOpt = getPermanentTable(objectIdentifier, false);
if (resultOpt.isPresent() && filter.test(resultOpt.get().getTable())) {
execute(
(catalog, path) -> catalog.dropTable(path, ignoreIfNotExists),
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/catalog/CatalogTableITCase.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/catalog/CatalogTableITCase.scala
index 780db45..d9ecb0f 100644
--- a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/catalog/CatalogTableITCase.scala
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/catalog/CatalogTableITCase.scala
@@ -29,8 +29,7 @@ import org.apache.flink.table.planner.utils.DateTimeTestUtil.localDateTime
import org.apache.flink.test.util.AbstractTestBase
import org.apache.flink.types.Row
import org.apache.flink.util.FileUtils
-
-import org.junit.Assert.{assertEquals, fail}
+import org.junit.Assert.{assertEquals, assertTrue, fail}
import org.junit.rules.ExpectedException
import org.junit.runner.RunWith
import org.junit.runners.Parameterized
@@ -909,6 +908,29 @@ class CatalogTableITCase(isStreamingMode: Boolean) extends AbstractTestBase {
}
@Test
+ def testDropTableWithIllegalWatermark(): Unit = {
+ // for FLINK-20937
+ val illegalDDL =
+ """
+ |create table t1 (
+ | a bigint,
+ | b bigint,
+ | proctime as PROCTIME(),
+ | WATERMARK FOR proctime as proctime - INTERVAL '5' SECOND
+ |) with (
+ | 'connector' = 'COLLECTION'
+ |)
+ """.stripMargin
+
+ // create table doesn't check the validity for now
+ tableEnv.executeSql(illegalDDL)
+ assert(tableEnv.listTables().sameElements(Array[String]("t1")))
+ // should success
+ tableEnv.executeSql("DROP TABLE t1")
+ assertTrue(tableEnv.listTables().isEmpty)
+ }
+
+ @Test
def testDropViewSameNameWithTable(): Unit = {
val createTable1 =
"""