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 2023/03/19 07:33:28 UTC
[flink] branch master updated: [FLINK-31505][table] Move execution logic of DropOperation out from TableEnvironmentImpl (#22209)
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 06be368a3f1 [FLINK-31505][table] Move execution logic of DropOperation out from TableEnvironmentImpl (#22209)
06be368a3f1 is described below
commit 06be368a3f1e821bdb1063b828009748f2f4239c
Author: Sergey Nuyanzin <sn...@gmail.com>
AuthorDate: Sun Mar 19 08:33:20 2023 +0100
[FLINK-31505][table] Move execution logic of DropOperation out from TableEnvironmentImpl (#22209)
---
.../internal/ExecutableOperationContextImpl.java | 12 ++-
.../table/api/internal/TableEnvironmentImpl.java | 111 +--------------------
.../apache/flink/table/catalog/CatalogManager.java | 8 ++
.../table/operations/ExecutableOperation.java | 3 +
.../ddl/DropCatalogFunctionOperation.java | 30 ++++++
.../table/operations/ddl/DropCatalogOperation.java | 15 +++
.../operations/ddl/DropDatabaseOperation.java | 22 ++++
.../flink/table/operations/ddl/DropOperation.java | 3 +-
.../table/operations/ddl/DropTableOperation.java | 12 +++
.../ddl/DropTempSystemFunctionOperation.java | 16 +++
.../table/operations/ddl/DropViewOperation.java | 12 +++
11 files changed, 133 insertions(+), 111 deletions(-)
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/ExecutableOperationContextImpl.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/ExecutableOperationContextImpl.java
index 6cc41d7843c..bcc5695d27e 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/ExecutableOperationContextImpl.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/ExecutableOperationContextImpl.java
@@ -19,6 +19,7 @@
package org.apache.flink.table.api.internal;
import org.apache.flink.table.catalog.CatalogManager;
+import org.apache.flink.table.catalog.FunctionCatalog;
import org.apache.flink.table.module.ModuleManager;
import org.apache.flink.table.operations.ExecutableOperation;
@@ -26,11 +27,15 @@ import org.apache.flink.table.operations.ExecutableOperation;
public class ExecutableOperationContextImpl implements ExecutableOperation.Context {
private final CatalogManager catalogManager;
+ private final FunctionCatalog functionCatalog;
private final ModuleManager moduleManager;
public ExecutableOperationContextImpl(
- CatalogManager catalogManager, ModuleManager moduleManager) {
+ CatalogManager catalogManager,
+ FunctionCatalog functionCatalog,
+ ModuleManager moduleManager) {
this.catalogManager = catalogManager;
+ this.functionCatalog = functionCatalog;
this.moduleManager = moduleManager;
}
@@ -39,6 +44,11 @@ public class ExecutableOperationContextImpl implements ExecutableOperation.Conte
return catalogManager;
}
+ @Override
+ public FunctionCatalog getFunctionCatalog() {
+ return functionCatalog;
+ }
+
@Override
public ModuleManager getModuleManager() {
return moduleManager;
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java
index f5871f08dc3..62344fca162 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java
@@ -63,7 +63,6 @@ import org.apache.flink.table.catalog.UnresolvedIdentifier;
import org.apache.flink.table.catalog.WatermarkSpec;
import org.apache.flink.table.catalog.exceptions.CatalogException;
import org.apache.flink.table.catalog.exceptions.DatabaseAlreadyExistException;
-import org.apache.flink.table.catalog.exceptions.DatabaseNotEmptyException;
import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException;
import org.apache.flink.table.catalog.exceptions.FunctionAlreadyExistException;
import org.apache.flink.table.catalog.exceptions.FunctionNotExistException;
@@ -115,7 +114,6 @@ import org.apache.flink.table.operations.SourceQueryOperation;
import org.apache.flink.table.operations.StatementSetOperation;
import org.apache.flink.table.operations.TableSourceQueryOperation;
import org.apache.flink.table.operations.UnloadModuleOperation;
-import org.apache.flink.table.operations.UseModulesOperation;
import org.apache.flink.table.operations.command.AddJarOperation;
import org.apache.flink.table.operations.command.ExecutePlanOperation;
import org.apache.flink.table.operations.command.ShowJarsOperation;
@@ -140,13 +138,7 @@ import org.apache.flink.table.operations.ddl.CreateDatabaseOperation;
import org.apache.flink.table.operations.ddl.CreateTableOperation;
import org.apache.flink.table.operations.ddl.CreateTempSystemFunctionOperation;
import org.apache.flink.table.operations.ddl.CreateViewOperation;
-import org.apache.flink.table.operations.ddl.DropCatalogFunctionOperation;
-import org.apache.flink.table.operations.ddl.DropCatalogOperation;
-import org.apache.flink.table.operations.ddl.DropDatabaseOperation;
import org.apache.flink.table.operations.ddl.DropPartitionsOperation;
-import org.apache.flink.table.operations.ddl.DropTableOperation;
-import org.apache.flink.table.operations.ddl.DropTempSystemFunctionOperation;
-import org.apache.flink.table.operations.ddl.DropViewOperation;
import org.apache.flink.table.operations.utils.OperationTreeBuilder;
import org.apache.flink.table.resource.ResourceManager;
import org.apache.flink.table.resource.ResourceType;
@@ -262,7 +254,8 @@ public class TableEnvironmentImpl implements TableEnvironmentInternal {
isStreamingMode);
catalogManager.initSchemaResolver(
isStreamingMode, operationTreeBuilder.getResolverBuilder());
- this.operationCtx = new ExecutableOperationContextImpl(catalogManager, moduleManager);
+ this.operationCtx =
+ new ExecutableOperationContextImpl(catalogManager, functionCatalog, moduleManager);
}
public static TableEnvironmentImpl create(Configuration configuration) {
@@ -1010,16 +1003,6 @@ public class TableEnvironmentImpl implements TableEnvironmentInternal {
createTableOperation.isIgnoreIfExists());
}
return TableResultImpl.TABLE_RESULT_OK;
- } else if (operation instanceof DropTableOperation) {
- DropTableOperation dropTableOperation = (DropTableOperation) operation;
- if (dropTableOperation.isTemporary()) {
- catalogManager.dropTemporaryTable(
- dropTableOperation.getTableIdentifier(), dropTableOperation.isIfExists());
- } else {
- catalogManager.dropTable(
- dropTableOperation.getTableIdentifier(), dropTableOperation.isIfExists());
- }
- return TableResultImpl.TABLE_RESULT_OK;
} else if (operation instanceof AlterTableOperation) {
AlterTableOperation alterTableOperation = (AlterTableOperation) operation;
Catalog catalog =
@@ -1111,16 +1094,6 @@ public class TableEnvironmentImpl implements TableEnvironmentInternal {
createViewOperation.isIgnoreIfExists());
}
return TableResultImpl.TABLE_RESULT_OK;
- } else if (operation instanceof DropViewOperation) {
- DropViewOperation dropViewOperation = (DropViewOperation) operation;
- if (dropViewOperation.isTemporary()) {
- catalogManager.dropTemporaryView(
- dropViewOperation.getViewIdentifier(), dropViewOperation.isIfExists());
- } else {
- catalogManager.dropView(
- dropViewOperation.getViewIdentifier(), dropViewOperation.isIfExists());
- }
- return TableResultImpl.TABLE_RESULT_OK;
} else if (operation instanceof AlterViewOperation) {
AlterViewOperation alterViewOperation = (AlterViewOperation) operation;
Catalog catalog =
@@ -1171,21 +1144,6 @@ public class TableEnvironmentImpl implements TableEnvironmentInternal {
} catch (Exception e) {
throw new TableException(exMsg, e);
}
- } else if (operation instanceof DropDatabaseOperation) {
- DropDatabaseOperation dropDatabaseOperation = (DropDatabaseOperation) operation;
- Catalog catalog = getCatalogOrThrowException(dropDatabaseOperation.getCatalogName());
- String exMsg = getDDLOpExecuteErrorMsg(dropDatabaseOperation.asSummaryString());
- try {
- catalog.dropDatabase(
- dropDatabaseOperation.getDatabaseName(),
- dropDatabaseOperation.isIfExists(),
- dropDatabaseOperation.isCascade());
- return TableResultImpl.TABLE_RESULT_OK;
- } catch (DatabaseNotExistException | DatabaseNotEmptyException e) {
- throw new ValidationException(exMsg, e);
- } catch (Exception e) {
- throw new TableException(exMsg, e);
- }
} else if (operation instanceof AlterDatabaseOperation) {
AlterDatabaseOperation alterDatabaseOperation = (AlterDatabaseOperation) operation;
Catalog catalog = getCatalogOrThrowException(alterDatabaseOperation.getCatalogName());
@@ -1205,10 +1163,6 @@ public class TableEnvironmentImpl implements TableEnvironmentInternal {
return createCatalogFunction((CreateCatalogFunctionOperation) operation);
} else if (operation instanceof CreateTempSystemFunctionOperation) {
return createSystemFunction((CreateTempSystemFunctionOperation) operation);
- } else if (operation instanceof DropCatalogFunctionOperation) {
- return dropCatalogFunction((DropCatalogFunctionOperation) operation);
- } else if (operation instanceof DropTempSystemFunctionOperation) {
- return dropSystemFunction((DropTempSystemFunctionOperation) operation);
} else if (operation instanceof AlterCatalogFunctionOperation) {
return alterCatalogFunction((AlterCatalogFunctionOperation) operation);
} else if (operation instanceof AddJarOperation) {
@@ -1217,16 +1171,6 @@ public class TableEnvironmentImpl implements TableEnvironmentInternal {
return buildShowResult("jars", listJars());
} else if (operation instanceof CreateCatalogOperation) {
return createCatalog((CreateCatalogOperation) operation);
- } else if (operation instanceof DropCatalogOperation) {
- DropCatalogOperation dropCatalogOperation = (DropCatalogOperation) operation;
- String exMsg = getDDLOpExecuteErrorMsg(dropCatalogOperation.asSummaryString());
- try {
- catalogManager.unregisterCatalog(
- dropCatalogOperation.getCatalogName(), dropCatalogOperation.isIfExists());
- return TableResultImpl.TABLE_RESULT_OK;
- } catch (CatalogException e) {
- throw new ValidationException(exMsg, e);
- }
} else if (operation instanceof LoadModuleOperation) {
return loadModule((LoadModuleOperation) operation);
} else if (operation instanceof UnloadModuleOperation) {
@@ -1500,16 +1444,6 @@ public class TableEnvironmentImpl implements TableEnvironmentInternal {
}
}
- private TableResultInternal useModules(UseModulesOperation operation) {
- String exMsg = getDDLOpExecuteErrorMsg(operation.asSummaryString());
- try {
- moduleManager.useModules(operation.getModuleNames().toArray(new String[0]));
- return TableResultImpl.TABLE_RESULT_OK;
- } catch (ValidationException e) {
- throw new ValidationException(String.format("%s. %s", exMsg, e.getMessage()), e);
- }
- }
-
private TableResultInternal buildShowResult(String columnName, String[] objects) {
return buildResult(
new String[] {columnName},
@@ -1941,35 +1875,6 @@ public class TableEnvironmentImpl implements TableEnvironmentInternal {
}
}
- private TableResultInternal dropCatalogFunction(
- DropCatalogFunctionOperation dropCatalogFunctionOperation) {
- String exMsg = getDDLOpExecuteErrorMsg(dropCatalogFunctionOperation.asSummaryString());
- try {
- if (dropCatalogFunctionOperation.isTemporary()) {
- functionCatalog.dropTempCatalogFunction(
- dropCatalogFunctionOperation.getFunctionIdentifier(),
- dropCatalogFunctionOperation.isIfExists());
- } else {
- Catalog catalog =
- getCatalogOrThrowException(
- dropCatalogFunctionOperation
- .getFunctionIdentifier()
- .getCatalogName());
-
- catalog.dropFunction(
- dropCatalogFunctionOperation.getFunctionIdentifier().toObjectPath(),
- dropCatalogFunctionOperation.isIfExists());
- }
- return TableResultImpl.TABLE_RESULT_OK;
- } catch (ValidationException e) {
- throw e;
- } catch (FunctionNotExistException e) {
- throw new ValidationException(e.getMessage(), e);
- } catch (Exception e) {
- throw new TableException(exMsg, e);
- }
- }
-
private TableResultInternal createSystemFunction(CreateTempSystemFunctionOperation operation) {
String exMsg = getDDLOpExecuteErrorMsg(operation.asSummaryString());
@@ -1986,18 +1891,6 @@ public class TableEnvironmentImpl implements TableEnvironmentInternal {
}
}
- private TableResultInternal dropSystemFunction(DropTempSystemFunctionOperation operation) {
- try {
- functionCatalog.dropTemporarySystemFunction(
- operation.getFunctionName(), operation.isIfExists());
- return TableResultImpl.TABLE_RESULT_OK;
- } catch (ValidationException e) {
- throw e;
- } catch (Exception e) {
- throw new TableException(getDDLOpExecuteErrorMsg(operation.asSummaryString()), e);
- }
- }
-
@VisibleForTesting
public TableImpl createTable(QueryOperation tableOperation) {
return TableImpl.createTable(
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 09fbb880e85..4b65dcd8aa8 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
@@ -246,6 +246,14 @@ public final class CatalogManager {
return Optional.ofNullable(catalogs.get(catalogName));
}
+ public Catalog getCatalogOrThrowException(String catalogName) {
+ return getCatalog(catalogName)
+ .orElseThrow(
+ () ->
+ new ValidationException(
+ String.format("Catalog %s does not exist", catalogName)));
+ }
+
/**
* Gets the current catalog that will be used when resolving table path.
*
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ExecutableOperation.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ExecutableOperation.java
index ae542de7ad4..acfcf5347bb 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ExecutableOperation.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ExecutableOperation.java
@@ -22,6 +22,7 @@ import org.apache.flink.annotation.Internal;
import org.apache.flink.table.api.internal.TableEnvironmentImpl;
import org.apache.flink.table.api.internal.TableResultInternal;
import org.apache.flink.table.catalog.CatalogManager;
+import org.apache.flink.table.catalog.FunctionCatalog;
import org.apache.flink.table.module.ModuleManager;
/**
@@ -53,6 +54,8 @@ public interface ExecutableOperation extends Operation {
CatalogManager getCatalogManager();
+ FunctionCatalog getFunctionCatalog();
+
ModuleManager getModuleManager();
}
}
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/DropCatalogFunctionOperation.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/DropCatalogFunctionOperation.java
index 8bff640a5e8..3b37d25eda4 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/DropCatalogFunctionOperation.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/DropCatalogFunctionOperation.java
@@ -18,7 +18,13 @@
package org.apache.flink.table.operations.ddl;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.api.internal.TableResultImpl;
+import org.apache.flink.table.api.internal.TableResultInternal;
+import org.apache.flink.table.catalog.Catalog;
import org.apache.flink.table.catalog.ObjectIdentifier;
+import org.apache.flink.table.catalog.exceptions.FunctionNotExistException;
import org.apache.flink.table.operations.Operation;
import org.apache.flink.table.operations.OperationUtils;
@@ -68,4 +74,28 @@ public class DropCatalogFunctionOperation implements DropOperation {
public String getFunctionName() {
return this.functionIdentifier.getObjectName();
}
+
+ @Override
+ public TableResultInternal execute(Context ctx) {
+ try {
+ if (isTemporary()) {
+ ctx.getFunctionCatalog()
+ .dropTempCatalogFunction(getFunctionIdentifier(), isIfExists());
+ } else {
+ Catalog catalog =
+ ctx.getCatalogManager()
+ .getCatalogOrThrowException(
+ getFunctionIdentifier().getCatalogName());
+
+ catalog.dropFunction(getFunctionIdentifier().toObjectPath(), isIfExists());
+ }
+ return TableResultImpl.TABLE_RESULT_OK;
+ } catch (ValidationException e) {
+ throw e;
+ } catch (FunctionNotExistException e) {
+ throw new ValidationException(e.getMessage(), e);
+ } catch (Exception e) {
+ throw new TableException(String.format("Could not execute %s", asSummaryString()), e);
+ }
+ }
}
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/DropCatalogOperation.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/DropCatalogOperation.java
index 077d2676038..bf203fe2c60 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/DropCatalogOperation.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/DropCatalogOperation.java
@@ -18,6 +18,10 @@
package org.apache.flink.table.operations.ddl;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.api.internal.TableResultImpl;
+import org.apache.flink.table.api.internal.TableResultInternal;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
import org.apache.flink.table.operations.Operation;
import org.apache.flink.table.operations.OperationUtils;
@@ -51,4 +55,15 @@ public class DropCatalogOperation implements DropOperation {
return OperationUtils.formatWithChildren(
"DROP CATALOG", params, Collections.emptyList(), Operation::asSummaryString);
}
+
+ @Override
+ public TableResultInternal execute(Context ctx) {
+ try {
+ ctx.getCatalogManager().unregisterCatalog(getCatalogName(), isIfExists());
+ return TableResultImpl.TABLE_RESULT_OK;
+ } catch (CatalogException e) {
+ throw new ValidationException(
+ String.format("Could not execute %s", asSummaryString()), e);
+ }
+ }
}
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/DropDatabaseOperation.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/DropDatabaseOperation.java
index 41d89543e16..6e93144e00a 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/DropDatabaseOperation.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/DropDatabaseOperation.java
@@ -18,6 +18,14 @@
package org.apache.flink.table.operations.ddl;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.api.internal.TableResultImpl;
+import org.apache.flink.table.api.internal.TableResultInternal;
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.catalog.exceptions.DatabaseNotEmptyException;
+import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException;
+
/** Operation to describe a DROP DATABASE statement. */
public class DropDatabaseOperation implements DropOperation {
private final String catalogName;
@@ -57,4 +65,18 @@ public class DropDatabaseOperation implements DropOperation {
summaryString.append(cascade ? " CASCADE" : " RESTRICT");
return summaryString.toString();
}
+
+ @Override
+ public TableResultInternal execute(Context ctx) {
+ Catalog catalog = ctx.getCatalogManager().getCatalogOrThrowException(getCatalogName());
+ try {
+ catalog.dropDatabase(getDatabaseName(), isIfExists(), isCascade());
+ return TableResultImpl.TABLE_RESULT_OK;
+ } catch (DatabaseNotExistException | DatabaseNotEmptyException e) {
+ throw new ValidationException(
+ String.format("Could not execute %s", asSummaryString()), e);
+ } catch (Exception e) {
+ throw new TableException(String.format("Could not execute %s", asSummaryString()), e);
+ }
+ }
}
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/DropOperation.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/DropOperation.java
index a1d3704ac45..bf685ddec8f 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/DropOperation.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/DropOperation.java
@@ -18,6 +18,7 @@
package org.apache.flink.table.operations.ddl;
+import org.apache.flink.table.operations.ExecutableOperation;
import org.apache.flink.table.operations.Operation;
/**
@@ -26,4 +27,4 @@ import org.apache.flink.table.operations.Operation;
* <p>Different sub operations can have their special target name. For example, a drop table
* operation may have a target table name and a flag to describe if is exists.
*/
-public interface DropOperation extends Operation {}
+public interface DropOperation extends Operation, ExecutableOperation {}
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/DropTableOperation.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/DropTableOperation.java
index c7e0359a5ed..3fc3c76107a 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/DropTableOperation.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/DropTableOperation.java
@@ -18,6 +18,8 @@
package org.apache.flink.table.operations.ddl;
+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;
@@ -61,4 +63,14 @@ public class DropTableOperation implements DropOperation {
return OperationUtils.formatWithChildren(
"DROP TABLE", params, Collections.emptyList(), Operation::asSummaryString);
}
+
+ @Override
+ public TableResultInternal execute(Context ctx) {
+ if (isTemporary()) {
+ ctx.getCatalogManager().dropTemporaryTable(getTableIdentifier(), isIfExists());
+ } else {
+ ctx.getCatalogManager().dropTable(getTableIdentifier(), isIfExists());
+ }
+ return TableResultImpl.TABLE_RESULT_OK;
+ }
}
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/DropTempSystemFunctionOperation.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/DropTempSystemFunctionOperation.java
index 3887909fce4..70e02df4d25 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/DropTempSystemFunctionOperation.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/DropTempSystemFunctionOperation.java
@@ -18,6 +18,10 @@
package org.apache.flink.table.operations.ddl;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.api.internal.TableResultImpl;
+import org.apache.flink.table.api.internal.TableResultInternal;
import org.apache.flink.table.operations.Operation;
import org.apache.flink.table.operations.OperationUtils;
@@ -55,4 +59,16 @@ public class DropTempSystemFunctionOperation implements DropOperation {
Collections.emptyList(),
Operation::asSummaryString);
}
+
+ @Override
+ public TableResultInternal execute(Context ctx) {
+ try {
+ ctx.getFunctionCatalog().dropTemporarySystemFunction(getFunctionName(), isIfExists());
+ return TableResultImpl.TABLE_RESULT_OK;
+ } catch (ValidationException e) {
+ throw e;
+ } catch (Exception e) {
+ throw new TableException(String.format("Could not execute %s", asSummaryString()), e);
+ }
+ }
}
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/DropViewOperation.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/DropViewOperation.java
index 0f515d2b15b..dd20001f85a 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/DropViewOperation.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/DropViewOperation.java
@@ -18,6 +18,8 @@
package org.apache.flink.table.operations.ddl;
+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;
@@ -62,4 +64,14 @@ public class DropViewOperation implements DropOperation {
return OperationUtils.formatWithChildren(
"DROP VIEW", params, Collections.emptyList(), Operation::asSummaryString);
}
+
+ @Override
+ public TableResultInternal execute(Context ctx) {
+ if (isTemporary()) {
+ ctx.getCatalogManager().dropTemporaryView(getViewIdentifier(), isIfExists());
+ } else {
+ ctx.getCatalogManager().dropView(getViewIdentifier(), isIfExists());
+ }
+ return TableResultImpl.TABLE_RESULT_OK;
+ }
}