This is an automated email from the ASF dual-hosted git repository.
ron pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 132d2fe24f18f27d3f752e96ea43ee7c6d196418
Author: fengli <ldliu...@163.com>
AuthorDate: Wed Jun 26 13:31:18 2024 +0800
[FLINK-35691][table] Fix drop table statement can drop materialized table
---
.../service/MaterializedTableStatementITCase.java | 46 ++++++++++++++++++++++
.../apache/flink/table/catalog/CatalogManager.java | 30 ++++++++++----
.../DropMaterializedTableOperation.java | 27 +++++++++++--
3 files changed, 92 insertions(+), 11 deletions(-)
diff --git
a/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/service/MaterializedTableStatementITCase.java
b/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/service/MaterializedTableStatementITCase.java
index 98ee00d48ee..6ed91b954af 100644
---
a/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/service/MaterializedTableStatementITCase.java
+++
b/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/service/MaterializedTableStatementITCase.java
@@ -991,6 +991,29 @@ public class MaterializedTableStatementITCase extends
AbstractMaterializedTableS
List<RowData> jobResults = fetchAllResults(service, sessionHandle,
describeJobHandle);
assertThat(jobResults.get(0).getString(2).toString()).isEqualTo("RUNNING");
+ // Drop materialized table using drop table statement
+ String dropTableUsingMaterializedTableDDL = "DROP TABLE users_shops";
+ OperationHandle dropTableUsingMaterializedTableHandle =
+ service.executeStatement(
+ sessionHandle, dropTableUsingMaterializedTableDDL, -1,
new Configuration());
+
+ assertThatThrownBy(
+ () ->
+ awaitOperationTermination(
+ service,
+ sessionHandle,
+ dropTableUsingMaterializedTableHandle))
+ .rootCause()
+ .isInstanceOf(ValidationException.class)
+ .hasMessage(
+ String.format(
+ "Table with identifier '%s' does not exist.",
+ ObjectIdentifier.of(
+ fileSystemCatalogName,
+ TEST_DEFAULT_DATABASE,
+ "users_shops")
+ .asSummaryString()));
+
// drop materialized table
String dropMaterializedTableDDL = "DROP MATERIALIZED TABLE IF EXISTS
users_shops";
OperationHandle dropMaterializedTableHandle =
@@ -1091,6 +1114,29 @@ public class MaterializedTableStatementITCase extends
AbstractMaterializedTableS
// verify refresh workflow is created
assertThat(embeddedWorkflowScheduler.getQuartzScheduler().checkExists(jobKey)).isTrue();
+ // Drop materialized table using drop table statement
+ String dropTableUsingMaterializedTableDDL = "DROP TABLE users_shops";
+ OperationHandle dropTableUsingMaterializedTableHandle =
+ service.executeStatement(
+ sessionHandle, dropTableUsingMaterializedTableDDL, -1,
new Configuration());
+
+ assertThatThrownBy(
+ () ->
+ awaitOperationTermination(
+ service,
+ sessionHandle,
+ dropTableUsingMaterializedTableHandle))
+ .rootCause()
+ .isInstanceOf(ValidationException.class)
+ .hasMessage(
+ String.format(
+ "Table with identifier '%s' does not exist.",
+ ObjectIdentifier.of(
+ fileSystemCatalogName,
+ TEST_DEFAULT_DATABASE,
+ "users_shops")
+ .asSummaryString()));
+
// drop materialized table
String dropMaterializedTableDDL = "DROP MATERIALIZED TABLE IF EXISTS
users_shops";
OperationHandle dropMaterializedTableHandle =
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 5491df3f79e..bba3673c435 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
@@ -1257,7 +1257,19 @@ public final class CatalogManager implements
CatalogRegistry, AutoCloseable {
* exist.
*/
public void dropTable(ObjectIdentifier objectIdentifier, boolean
ignoreIfNotExists) {
- dropTableInternal(objectIdentifier, ignoreIfNotExists, true);
+ dropTableInternal(objectIdentifier, ignoreIfNotExists, true, false);
+ }
+
+ /**
+ * Drops a materialized table in a given fully qualified path.
+ *
+ * @param objectIdentifier The fully qualified path of the materialized
table to drop.
+ * @param ignoreIfNotExists If false exception will be thrown if the table
to drop does not
+ * exist.
+ */
+ public void dropMaterializedTable(
+ ObjectIdentifier objectIdentifier, boolean ignoreIfNotExists) {
+ dropTableInternal(objectIdentifier, ignoreIfNotExists, true, true);
}
/**
@@ -1268,16 +1280,19 @@ public final class CatalogManager implements
CatalogRegistry, AutoCloseable {
* exist.
*/
public void dropView(ObjectIdentifier objectIdentifier, boolean
ignoreIfNotExists) {
- dropTableInternal(objectIdentifier, ignoreIfNotExists, false);
+ dropTableInternal(objectIdentifier, ignoreIfNotExists, false, false);
}
private void dropTableInternal(
- ObjectIdentifier objectIdentifier, boolean ignoreIfNotExists,
boolean isDropTable) {
+ ObjectIdentifier objectIdentifier,
+ boolean ignoreIfNotExists,
+ boolean isDropTable,
+ boolean isDropMaterializedTable) {
Predicate<CatalogBaseTable> filter =
isDropTable
- ? table ->
- table instanceof CatalogTable
- || table instanceof
CatalogMaterializedTable
+ ? isDropMaterializedTable
+ ? table -> table instanceof
CatalogMaterializedTable
+ : table -> table instanceof CatalogTable
: table -> table instanceof CatalogView;
// Same name temporary table or view exists.
if (filter.test(temporaryTables.get(objectIdentifier))) {
@@ -1317,7 +1332,8 @@ public final class CatalogManager implements
CatalogRegistry, AutoCloseable {
ignoreIfNotExists,
"DropTable");
} else if (!ignoreIfNotExists) {
- String tableOrView = isDropTable ? "Table" : "View";
+ String tableOrView =
+ isDropTable ? isDropMaterializedTable ? "Materialized
Table" : "Table" : "View";
throw new ValidationException(
String.format(
"%s with identifier '%s' does not exist.",
diff --git
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/materializedtable/DropMaterializedTableOperation.java
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/materializedtable/DropMaterializedTableOperation.java
index 46dd86ad96b..b499a3e0b60 100644
---
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/materializedtable/DropMaterializedTableOperation.java
+++
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/materializedtable/DropMaterializedTableOperation.java
@@ -19,10 +19,12 @@
package org.apache.flink.table.operations.materializedtable;
import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.internal.TableResultImpl;
+import org.apache.flink.table.api.internal.TableResultInternal;
import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.operations.Operation;
import org.apache.flink.table.operations.OperationUtils;
-import org.apache.flink.table.operations.ddl.DropTableOperation;
+import org.apache.flink.table.operations.ddl.DropOperation;
import java.util.Collections;
import java.util.LinkedHashMap;
@@ -30,11 +32,22 @@ import java.util.Map;
/** Operation to describe a DROP MATERIALIZED TABLE statement. */
@Internal
-public class DropMaterializedTableOperation extends DropTableOperation
- implements MaterializedTableOperation {
+public class DropMaterializedTableOperation implements DropOperation,
MaterializedTableOperation {
+
+ private final ObjectIdentifier tableIdentifier;
+ private final boolean ifExists;
public DropMaterializedTableOperation(ObjectIdentifier tableIdentifier,
boolean ifExists) {
- super(tableIdentifier, ifExists, false);
+ this.tableIdentifier = tableIdentifier;
+ this.ifExists = ifExists;
+ }
+
+ public ObjectIdentifier getTableIdentifier() {
+ return tableIdentifier;
+ }
+
+ public boolean isIfExists() {
+ return ifExists;
}
@Override
@@ -49,4 +62,10 @@ public class DropMaterializedTableOperation extends
DropTableOperation
Collections.emptyList(),
Operation::asSummaryString);
}
+
+ @Override
+ public TableResultInternal execute(Context ctx) {
+ ctx.getCatalogManager().dropMaterializedTable(getTableIdentifier(),
isIfExists());
+ return TableResultImpl.TABLE_RESULT_OK;
+ }
}
