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 2023/03/19 19:14:20 UTC
[flink] branch master updated: [FLINK-31504][table] Move execution logic of CreateOperation out from TableEnvironmentImpl (#22214)
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
The following commit(s) were added to refs/heads/master by this push:
new 3e6a831eb30 [FLINK-31504][table] Move execution logic of CreateOperation out from TableEnvironmentImpl (#22214)
3e6a831eb30 is described below
commit 3e6a831eb30883790aaf37c2f49dfcb5e1b5e8ae
Author: Jark Wu <ja...@apache.org>
AuthorDate: Mon Mar 20 03:14:10 2023 +0800
[FLINK-31504][table] Move execution logic of CreateOperation out from TableEnvironmentImpl (#22214)
---
.../internal/ExecutableOperationContextImpl.java | 20 ++++-
.../table/api/internal/TableEnvironmentImpl.java | 93 ++--------------------
.../table/operations/ExecutableOperation.java | 6 ++
.../ddl/CreateCatalogFunctionOperation.java | 39 ++++++++-
.../operations/ddl/CreateCatalogOperation.java | 29 ++++++-
.../operations/ddl/CreateDatabaseOperation.java | 20 +++++
.../table/operations/ddl/CreateOperation.java | 3 +-
.../table/operations/ddl/CreateTableOperation.java | 19 ++++-
.../ddl/CreateTempSystemFunctionOperation.java | 14 ++++
.../table/operations/ddl/CreateViewOperation.java | 19 ++++-
10 files changed, 162 insertions(+), 100 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 bcc5695d27e..17503b4b060 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
@@ -18,10 +18,12 @@
package org.apache.flink.table.api.internal;
+import org.apache.flink.table.api.TableConfig;
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;
+import org.apache.flink.table.resource.ResourceManager;
/** A simple implementation of {@link ExecutableOperation.Context}. */
public class ExecutableOperationContextImpl implements ExecutableOperation.Context {
@@ -29,14 +31,20 @@ public class ExecutableOperationContextImpl implements ExecutableOperation.Conte
private final CatalogManager catalogManager;
private final FunctionCatalog functionCatalog;
private final ModuleManager moduleManager;
+ private final ResourceManager resourceManager;
+ private final TableConfig tableConfig;
public ExecutableOperationContextImpl(
CatalogManager catalogManager,
FunctionCatalog functionCatalog,
- ModuleManager moduleManager) {
+ ModuleManager moduleManager,
+ ResourceManager resourceManager,
+ TableConfig tableConfig) {
this.catalogManager = catalogManager;
this.functionCatalog = functionCatalog;
this.moduleManager = moduleManager;
+ this.resourceManager = resourceManager;
+ this.tableConfig = tableConfig;
}
@Override
@@ -53,4 +61,14 @@ public class ExecutableOperationContextImpl implements ExecutableOperation.Conte
public ModuleManager getModuleManager() {
return moduleManager;
}
+
+ @Override
+ public ResourceManager getResourceManager() {
+ return resourceManager;
+ }
+
+ @Override
+ public TableConfig getTableConfig() {
+ return tableConfig;
+ }
}
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 62344fca162..82e0315f883 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
@@ -62,9 +62,7 @@ import org.apache.flink.table.catalog.ResolvedSchema;
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.DatabaseNotExistException;
-import org.apache.flink.table.catalog.exceptions.FunctionAlreadyExistException;
import org.apache.flink.table.catalog.exceptions.FunctionNotExistException;
import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
import org.apache.flink.table.catalog.exceptions.TableNotExistException;
@@ -132,12 +130,8 @@ import org.apache.flink.table.operations.ddl.AlterViewPropertiesOperation;
import org.apache.flink.table.operations.ddl.AlterViewRenameOperation;
import org.apache.flink.table.operations.ddl.AnalyzeTableOperation;
import org.apache.flink.table.operations.ddl.CompilePlanOperation;
-import org.apache.flink.table.operations.ddl.CreateCatalogFunctionOperation;
import org.apache.flink.table.operations.ddl.CreateCatalogOperation;
-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.DropPartitionsOperation;
import org.apache.flink.table.operations.utils.OperationTreeBuilder;
import org.apache.flink.table.resource.ResourceManager;
@@ -255,7 +249,12 @@ public class TableEnvironmentImpl implements TableEnvironmentInternal {
catalogManager.initSchemaResolver(
isStreamingMode, operationTreeBuilder.getResolverBuilder());
this.operationCtx =
- new ExecutableOperationContextImpl(catalogManager, functionCatalog, moduleManager);
+ new ExecutableOperationContextImpl(
+ catalogManager,
+ functionCatalog,
+ moduleManager,
+ resourceManager,
+ tableConfig);
}
public static TableEnvironmentImpl create(Configuration configuration) {
@@ -989,20 +988,6 @@ public class TableEnvironmentImpl implements TableEnvironmentInternal {
return executeInternal(Collections.singletonList((ModifyOperation) operation));
} else if (operation instanceof StatementSetOperation) {
return executeInternal(((StatementSetOperation) operation).getOperations());
- } else if (operation instanceof CreateTableOperation) {
- CreateTableOperation createTableOperation = (CreateTableOperation) operation;
- if (createTableOperation.isTemporary()) {
- catalogManager.createTemporaryTable(
- createTableOperation.getCatalogTable(),
- createTableOperation.getTableIdentifier(),
- createTableOperation.isIgnoreIfExists());
- } else {
- catalogManager.createTable(
- createTableOperation.getCatalogTable(),
- createTableOperation.getTableIdentifier(),
- createTableOperation.isIgnoreIfExists());
- }
- return TableResultImpl.TABLE_RESULT_OK;
} else if (operation instanceof AlterTableOperation) {
AlterTableOperation alterTableOperation = (AlterTableOperation) operation;
Catalog catalog =
@@ -1080,20 +1065,6 @@ public class TableEnvironmentImpl implements TableEnvironmentInternal {
} catch (Exception e) {
throw new TableException(exMsg, e);
}
- } else if (operation instanceof CreateViewOperation) {
- CreateViewOperation createViewOperation = (CreateViewOperation) operation;
- if (createViewOperation.isTemporary()) {
- catalogManager.createTemporaryTable(
- createViewOperation.getCatalogView(),
- createViewOperation.getViewIdentifier(),
- createViewOperation.isIgnoreIfExists());
- } else {
- catalogManager.createTable(
- createViewOperation.getCatalogView(),
- createViewOperation.getViewIdentifier(),
- createViewOperation.isIgnoreIfExists());
- }
- return TableResultImpl.TABLE_RESULT_OK;
} else if (operation instanceof AlterViewOperation) {
AlterViewOperation alterViewOperation = (AlterViewOperation) operation;
Catalog catalog =
@@ -1129,21 +1100,6 @@ public class TableEnvironmentImpl implements TableEnvironmentInternal {
} catch (Exception e) {
throw new TableException(exMsg, e);
}
- } else if (operation instanceof CreateDatabaseOperation) {
- CreateDatabaseOperation createDatabaseOperation = (CreateDatabaseOperation) operation;
- Catalog catalog = getCatalogOrThrowException(createDatabaseOperation.getCatalogName());
- String exMsg = getDDLOpExecuteErrorMsg(createDatabaseOperation.asSummaryString());
- try {
- catalog.createDatabase(
- createDatabaseOperation.getDatabaseName(),
- createDatabaseOperation.getCatalogDatabase(),
- createDatabaseOperation.isIgnoreIfExists());
- return TableResultImpl.TABLE_RESULT_OK;
- } catch (DatabaseAlreadyExistException 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());
@@ -1159,18 +1115,12 @@ public class TableEnvironmentImpl implements TableEnvironmentInternal {
} catch (Exception e) {
throw new TableException(exMsg, e);
}
- } else if (operation instanceof CreateCatalogFunctionOperation) {
- return createCatalogFunction((CreateCatalogFunctionOperation) operation);
- } else if (operation instanceof CreateTempSystemFunctionOperation) {
- return createSystemFunction((CreateTempSystemFunctionOperation) operation);
} else if (operation instanceof AlterCatalogFunctionOperation) {
return alterCatalogFunction((AlterCatalogFunctionOperation) operation);
} else if (operation instanceof AddJarOperation) {
return addJar((AddJarOperation) operation);
} else if (operation instanceof ShowJarsOperation) {
return buildShowResult("jars", listJars());
- } else if (operation instanceof CreateCatalogOperation) {
- return createCatalog((CreateCatalogOperation) operation);
} else if (operation instanceof LoadModuleOperation) {
return loadModule((LoadModuleOperation) operation);
} else if (operation instanceof UnloadModuleOperation) {
@@ -1816,37 +1766,6 @@ public class TableEnvironmentImpl implements TableEnvironmentInternal {
.map(ContextResolvedTable::getResolvedTable);
}
- private TableResultInternal createCatalogFunction(
- CreateCatalogFunctionOperation createCatalogFunctionOperation) {
- String exMsg = getDDLOpExecuteErrorMsg(createCatalogFunctionOperation.asSummaryString());
- try {
- if (createCatalogFunctionOperation.isTemporary()) {
- functionCatalog.registerTemporaryCatalogFunction(
- UnresolvedIdentifier.of(
- createCatalogFunctionOperation.getFunctionIdentifier().toList()),
- createCatalogFunctionOperation.getCatalogFunction(),
- createCatalogFunctionOperation.isIgnoreIfExists());
- } else {
- Catalog catalog =
- getCatalogOrThrowException(
- createCatalogFunctionOperation
- .getFunctionIdentifier()
- .getCatalogName());
- catalog.createFunction(
- createCatalogFunctionOperation.getFunctionIdentifier().toObjectPath(),
- createCatalogFunctionOperation.getCatalogFunction(),
- createCatalogFunctionOperation.isIgnoreIfExists());
- }
- return TableResultImpl.TABLE_RESULT_OK;
- } catch (ValidationException e) {
- throw e;
- } catch (FunctionAlreadyExistException e) {
- throw new ValidationException(e.getMessage(), e);
- } catch (Exception e) {
- throw new TableException(exMsg, e);
- }
- }
-
private TableResultInternal alterCatalogFunction(
AlterCatalogFunctionOperation alterCatalogFunctionOperation) {
String exMsg = getDDLOpExecuteErrorMsg(alterCatalogFunctionOperation.asSummaryString());
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 acfcf5347bb..5109268a5a9 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
@@ -19,11 +19,13 @@
package org.apache.flink.table.operations;
import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.TableConfig;
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;
+import org.apache.flink.table.resource.ResourceManager;
/**
* An {@link ExecutableOperation} represents an operation that is executed for its side effects.
@@ -57,5 +59,9 @@ public interface ExecutableOperation extends Operation {
FunctionCatalog getFunctionCatalog();
ModuleManager getModuleManager();
+
+ ResourceManager getResourceManager();
+
+ TableConfig getTableConfig();
}
}
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/CreateCatalogFunctionOperation.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/CreateCatalogFunctionOperation.java
index 0f4e5c19b2a..4516cb61063 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/CreateCatalogFunctionOperation.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/CreateCatalogFunctionOperation.java
@@ -18,8 +18,15 @@
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.CatalogFunction;
import org.apache.flink.table.catalog.ObjectIdentifier;
+import org.apache.flink.table.catalog.UnresolvedIdentifier;
+import org.apache.flink.table.catalog.exceptions.FunctionAlreadyExistException;
import org.apache.flink.table.operations.Operation;
import org.apache.flink.table.operations.OperationUtils;
@@ -30,9 +37,9 @@ import java.util.Map;
/** Operation to describe a CREATE FUNCTION statement for catalog functions. */
public class CreateCatalogFunctionOperation implements CreateOperation {
private final ObjectIdentifier functionIdentifier;
- private CatalogFunction catalogFunction;
- private boolean ignoreIfExists;
- private boolean isTemporary;
+ private final CatalogFunction catalogFunction;
+ private final boolean ignoreIfExists;
+ private final boolean isTemporary;
public CreateCatalogFunctionOperation(
ObjectIdentifier functionIdentifier,
@@ -79,4 +86,30 @@ public class CreateCatalogFunctionOperation implements CreateOperation {
public String getFunctionName() {
return this.functionIdentifier.getObjectName();
}
+
+ @Override
+ public TableResultInternal execute(Context ctx) {
+ try {
+ if (isTemporary) {
+ ctx.getFunctionCatalog()
+ .registerTemporaryCatalogFunction(
+ UnresolvedIdentifier.of(functionIdentifier.toList()),
+ catalogFunction,
+ ignoreIfExists);
+ } else {
+ Catalog catalog =
+ ctx.getCatalogManager()
+ .getCatalogOrThrowException(functionIdentifier.getCatalogName());
+ catalog.createFunction(
+ functionIdentifier.toObjectPath(), catalogFunction, ignoreIfExists);
+ }
+ return TableResultImpl.TABLE_RESULT_OK;
+ } catch (ValidationException e) {
+ throw e;
+ } catch (FunctionAlreadyExistException 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/CreateCatalogOperation.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/CreateCatalogOperation.java
index 92e3bf6ebd4..3a6a72ceb49 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/CreateCatalogOperation.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/CreateCatalogOperation.java
@@ -18,6 +18,12 @@
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.Catalog;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.flink.table.factories.FactoryUtil;
import org.apache.flink.table.operations.Operation;
import org.apache.flink.table.operations.OperationUtils;
@@ -34,7 +40,7 @@ public class CreateCatalogOperation implements CreateOperation {
public CreateCatalogOperation(String catalogName, Map<String, String> properties) {
this.catalogName = checkNotNull(catalogName);
- this.properties = checkNotNull(properties);
+ this.properties = Collections.unmodifiableMap(checkNotNull(properties));
}
public String getCatalogName() {
@@ -42,7 +48,7 @@ public class CreateCatalogOperation implements CreateOperation {
}
public Map<String, String> getProperties() {
- return Collections.unmodifiableMap(properties);
+ return properties;
}
@Override
@@ -54,4 +60,23 @@ public class CreateCatalogOperation implements CreateOperation {
return OperationUtils.formatWithChildren(
"CREATE CATALOG", params, Collections.emptyList(), Operation::asSummaryString);
}
+
+ @Override
+ public TableResultInternal execute(Context ctx) {
+ try {
+
+ Catalog catalog =
+ FactoryUtil.createCatalog(
+ catalogName,
+ properties,
+ ctx.getTableConfig(),
+ ctx.getResourceManager().getUserClassLoader());
+ ctx.getCatalogManager().registerCatalog(catalogName, catalog);
+
+ 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/CreateDatabaseOperation.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/CreateDatabaseOperation.java
index 4cbc13818fc..ea938ffa108 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/CreateDatabaseOperation.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/CreateDatabaseOperation.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.CatalogDatabase;
+import org.apache.flink.table.catalog.exceptions.DatabaseAlreadyExistException;
import org.apache.flink.table.operations.Operation;
import org.apache.flink.table.operations.OperationUtils;
@@ -71,4 +77,18 @@ public class CreateDatabaseOperation implements CreateOperation {
return OperationUtils.formatWithChildren(
"CREATE DATABASE", params, Collections.emptyList(), Operation::asSummaryString);
}
+
+ @Override
+ public TableResultInternal execute(Context ctx) {
+ Catalog catalog = ctx.getCatalogManager().getCatalogOrThrowException(catalogName);
+ try {
+ catalog.createDatabase(databaseName, catalogDatabase, ignoreIfExists);
+ return TableResultImpl.TABLE_RESULT_OK;
+ } catch (DatabaseAlreadyExistException 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/CreateOperation.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/CreateOperation.java
index 67d94f0322a..1a5422c7fbf 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/CreateOperation.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/CreateOperation.java
@@ -19,6 +19,7 @@
package org.apache.flink.table.operations.ddl;
import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.operations.ExecutableOperation;
import org.apache.flink.table.operations.Operation;
/**
@@ -30,4 +31,4 @@ import org.apache.flink.table.operations.Operation;
* instance.
*/
@Internal
-public interface CreateOperation extends Operation {}
+public interface CreateOperation extends Operation, ExecutableOperation {}
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/CreateTableOperation.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/CreateTableOperation.java
index 14709d2931b..3a2c19cf372 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/CreateTableOperation.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/CreateTableOperation.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.CatalogTable;
import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.operations.Operation;
@@ -30,9 +32,9 @@ import java.util.Map;
/** Operation to describe a CREATE TABLE statement. */
public class CreateTableOperation implements CreateOperation {
private final ObjectIdentifier tableIdentifier;
- private CatalogTable catalogTable;
- private boolean ignoreIfExists;
- private boolean isTemporary;
+ private final CatalogTable catalogTable;
+ private final boolean ignoreIfExists;
+ private final boolean isTemporary;
public CreateTableOperation(
ObjectIdentifier tableIdentifier,
@@ -72,4 +74,15 @@ public class CreateTableOperation implements CreateOperation {
return OperationUtils.formatWithChildren(
"CREATE TABLE", params, Collections.emptyList(), Operation::asSummaryString);
}
+
+ @Override
+ public TableResultInternal execute(Context ctx) {
+ if (isTemporary) {
+ ctx.getCatalogManager()
+ .createTemporaryTable(catalogTable, tableIdentifier, ignoreIfExists);
+ } else {
+ ctx.getCatalogManager().createTable(catalogTable, tableIdentifier, ignoreIfExists);
+ }
+ return TableResultImpl.TABLE_RESULT_OK;
+ }
}
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/CreateTempSystemFunctionOperation.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/CreateTempSystemFunctionOperation.java
index 281bafeee5d..279df355075 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/CreateTempSystemFunctionOperation.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/CreateTempSystemFunctionOperation.java
@@ -18,6 +18,9 @@
package org.apache.flink.table.operations.ddl;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.api.internal.TableResultImpl;
+import org.apache.flink.table.api.internal.TableResultInternal;
import org.apache.flink.table.catalog.CatalogFunction;
import org.apache.flink.table.catalog.CatalogFunctionImpl;
import org.apache.flink.table.catalog.FunctionCatalog;
@@ -83,4 +86,15 @@ public class CreateTempSystemFunctionOperation implements CreateOperation {
Collections.emptyList(),
Operation::asSummaryString);
}
+
+ @Override
+ public TableResultInternal execute(Context ctx) {
+ try {
+ ctx.getFunctionCatalog()
+ .registerTemporarySystemFunction(functionName, catalogFunction, ignoreIfExists);
+ return TableResultImpl.TABLE_RESULT_OK;
+ } 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/CreateViewOperation.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/CreateViewOperation.java
index 902e3dd4021..780e8ff9a8b 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/CreateViewOperation.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/CreateViewOperation.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.CatalogView;
import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.operations.Operation;
@@ -30,9 +32,9 @@ import java.util.Map;
/** Operation to describe a CREATE VIEW statement. */
public class CreateViewOperation implements CreateOperation {
private final ObjectIdentifier viewIdentifier;
- private CatalogView catalogView;
- private boolean ignoreIfExists;
- private boolean isTemporary;
+ private final CatalogView catalogView;
+ private final boolean ignoreIfExists;
+ private final boolean isTemporary;
public CreateViewOperation(
ObjectIdentifier viewIdentifier,
@@ -72,4 +74,15 @@ public class CreateViewOperation implements CreateOperation {
return OperationUtils.formatWithChildren(
"CREATE VIEW", params, Collections.emptyList(), Operation::asSummaryString);
}
+
+ @Override
+ public TableResultInternal execute(Context ctx) {
+ if (isTemporary) {
+ ctx.getCatalogManager()
+ .createTemporaryTable(catalogView, viewIdentifier, ignoreIfExists);
+ } else {
+ ctx.getCatalogManager().createTable(catalogView, viewIdentifier, ignoreIfExists);
+ }
+ return TableResultImpl.TABLE_RESULT_OK;
+ }
}