listViews(String catalogName, String databaseName) {
.collect(Collectors.toUnmodifiableSet());
}
- public ResultFetcher callStopJobOperation(
+ private ResultFetcher callStopJobOperation(
TableEnvironmentInternal tableEnv,
OperationHandle handle,
StopJobOperation stopJobOperation)
@@ -879,7 +888,7 @@ public ResultFetcher callShowJobsOperation(
resultRows);
}
- public ResultFetcher callDescribeJobOperation(
+ private ResultFetcher callDescribeJobOperation(
TableEnvironmentInternal tableEnv,
OperationHandle operationHandle,
DescribeJobOperation describeJobOperation)
diff --git a/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/utils/Constants.java b/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/utils/Constants.java
index be72df7a94f6bb..c2ceb435fe261e 100644
--- a/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/utils/Constants.java
+++ b/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/utils/Constants.java
@@ -29,5 +29,4 @@ public class Constants {
public static final String SET_VALUE = "value";
public static final String COMPLETION_CANDIDATES = "candidates";
public static final String SAVEPOINT_PATH = "savepoint path";
- public static final String CLUSTER_INFO = "cluster info";
}
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 f71337cfd11e4c..65e35f87b660ae 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
@@ -33,6 +33,7 @@
import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointingStatisticsHeaders;
import org.apache.flink.runtime.rest.util.RestMapperUtils;
import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.TableException;
import org.apache.flink.table.api.ValidationException;
import org.apache.flink.table.catalog.CatalogMaterializedTable.LogicalRefreshMode;
import org.apache.flink.table.catalog.CatalogMaterializedTable.RefreshMode;
@@ -819,7 +820,7 @@ void testAlterMaterializedTableWithRepeatedSuspendAndResumeInContinuousMode(
sessionHandle,
repeatedAlterMaterializedTableSuspendHandle))
.rootCause()
- .isInstanceOf(SqlExecutionException.class)
+ .isInstanceOf(TableException.class)
.hasMessageContaining(
String.format(
"Materialized table %s continuous refresh job has been suspended",
@@ -841,7 +842,7 @@ void testAlterMaterializedTableWithRepeatedSuspendAndResumeInContinuousMode(
sessionHandle,
repeatedAlterMaterializedTableResumeHandle))
.rootCause()
- .isInstanceOf(SqlExecutionException.class)
+ .isInstanceOf(TableException.class)
.hasMessageContaining(
String.format(
"Materialized table %s continuous refresh job has been resumed",
@@ -950,7 +951,7 @@ void testAlterMaterializedTableWithRepeatedSuspendAndResumeInFullMode() throws E
sessionHandle,
repeatedAlterMaterializedTableSuspendHandle))
.rootCause()
- .isInstanceOf(SqlExecutionException.class)
+ .isInstanceOf(TableException.class)
.hasMessageContaining(
String.format(
"Materialized table %s refresh workflow has been suspended.",
@@ -973,7 +974,7 @@ void testAlterMaterializedTableWithRepeatedSuspendAndResumeInFullMode() throws E
sessionHandle,
repeatedAlterMaterializedTableResumeHandle))
.rootCause()
- .isInstanceOf(SqlExecutionException.class)
+ .isInstanceOf(TableException.class)
.hasMessageContaining(
String.format(
"Materialized table %s refresh workflow has been resumed.",
@@ -1870,7 +1871,7 @@ void testRefreshMaterializedTableWithInvalidParameterInContinuousMode() throws E
awaitOperationTermination(
service, sessionHandle, invalidRefreshTableHandle2))
.rootCause()
- .isInstanceOf(SqlExecutionException.class)
+ .isInstanceOf(TableException.class)
.hasMessage(
String.format(
"Failed to parse a valid partition value for the field 'ds' in materialized table %s using the scheduler time '20240103 00:00:00.000' based on the date format 'yyyy-MM-dd HH:mm:ss'.",
diff --git a/flink-table/flink-table-api-bridge-base/src/main/java/org/apache/flink/table/api/bridge/internal/AbstractStreamTableEnvironmentImpl.java b/flink-table/flink-table-api-bridge-base/src/main/java/org/apache/flink/table/api/bridge/internal/AbstractStreamTableEnvironmentImpl.java
index 2e0bd471e5492e..15d32433d318ba 100644
--- a/flink-table/flink-table-api-bridge-base/src/main/java/org/apache/flink/table/api/bridge/internal/AbstractStreamTableEnvironmentImpl.java
+++ b/flink-table/flink-table-api-bridge-base/src/main/java/org/apache/flink/table/api/bridge/internal/AbstractStreamTableEnvironmentImpl.java
@@ -46,6 +46,8 @@
import org.apache.flink.table.delegation.ExecutorFactory;
import org.apache.flink.table.delegation.Planner;
import org.apache.flink.table.delegation.StreamExecutorFactory;
+import org.apache.flink.table.delegation.materializedtable.MaterializedTableExecutorFactory;
+import org.apache.flink.table.delegation.materializedtable.MaterializedTableJobSubmitter;
import org.apache.flink.table.expressions.ApiExpressionUtils;
import org.apache.flink.table.expressions.Expression;
import org.apache.flink.table.factories.FactoryUtil;
@@ -86,6 +88,32 @@ public AbstractStreamTableEnvironmentImpl(
Planner planner,
boolean isStreamingMode,
StreamExecutionEnvironment executionEnvironment) {
+ this(
+ catalogManager,
+ moduleManager,
+ resourceManager,
+ tableConfig,
+ executor,
+ functionCatalog,
+ planner,
+ isStreamingMode,
+ executionEnvironment,
+ lookupMaterializedTableExecutorFactory(resourceManager.getUserClassLoader()),
+ null);
+ }
+
+ public AbstractStreamTableEnvironmentImpl(
+ CatalogManager catalogManager,
+ ModuleManager moduleManager,
+ ResourceManager resourceManager,
+ TableConfig tableConfig,
+ Executor executor,
+ FunctionCatalog functionCatalog,
+ Planner planner,
+ boolean isStreamingMode,
+ StreamExecutionEnvironment executionEnvironment,
+ MaterializedTableExecutorFactory materializedTableExecutorFactory,
+ @Nullable MaterializedTableJobSubmitter materializedTableJobSubmitter) {
super(
catalogManager,
moduleManager,
@@ -94,10 +122,20 @@ public AbstractStreamTableEnvironmentImpl(
executor,
functionCatalog,
planner,
- isStreamingMode);
+ isStreamingMode,
+ materializedTableExecutorFactory,
+ materializedTableJobSubmitter);
this.executionEnvironment = executionEnvironment;
}
+ public static MaterializedTableExecutorFactory lookupMaterializedTableExecutorFactory(
+ ClassLoader classLoader) {
+ return FactoryUtil.discoverFactory(
+ classLoader,
+ MaterializedTableExecutorFactory.class,
+ MaterializedTableExecutorFactory.DEFAULT_IDENTIFIER);
+ }
+
public static Executor lookupExecutor(
ClassLoader classLoader, StreamExecutionEnvironment executionEnvironment) {
final ExecutorFactory executorFactory;
diff --git a/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/api/bridge/java/internal/StreamTableEnvironmentImpl.java b/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/api/bridge/java/internal/StreamTableEnvironmentImpl.java
index 810546ed6a6a5e..52cdd12df7bc2b 100644
--- a/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/api/bridge/java/internal/StreamTableEnvironmentImpl.java
+++ b/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/api/bridge/java/internal/StreamTableEnvironmentImpl.java
@@ -45,6 +45,8 @@
import org.apache.flink.table.connector.ChangelogMode;
import org.apache.flink.table.delegation.Executor;
import org.apache.flink.table.delegation.Planner;
+import org.apache.flink.table.delegation.materializedtable.MaterializedTableExecutorFactory;
+import org.apache.flink.table.delegation.materializedtable.MaterializedTableJobSubmitter;
import org.apache.flink.table.expressions.Expression;
import org.apache.flink.table.factories.ApiFactoryUtil;
import org.apache.flink.table.factories.CatalogStoreFactory;
@@ -62,6 +64,8 @@
import org.apache.flink.util.MutableURLClassLoader;
import org.apache.flink.util.Preconditions;
+import javax.annotation.Nullable;
+
import java.net.URL;
import java.util.Arrays;
import java.util.Optional;
@@ -85,7 +89,9 @@ public StreamTableEnvironmentImpl(
StreamExecutionEnvironment executionEnvironment,
Planner planner,
Executor executor,
- boolean isStreamingMode) {
+ boolean isStreamingMode,
+ MaterializedTableExecutorFactory materializedTableExecutorFactory,
+ @Nullable MaterializedTableJobSubmitter materializedTableJobSubmitter) {
super(
catalogManager,
moduleManager,
@@ -95,7 +101,9 @@ public StreamTableEnvironmentImpl(
functionCatalog,
planner,
isStreamingMode,
- executionEnvironment);
+ executionEnvironment,
+ materializedTableExecutorFactory,
+ materializedTableJobSubmitter);
}
public static StreamTableEnvironment create(
@@ -153,6 +161,9 @@ public static StreamTableEnvironment create(
catalogManager,
functionCatalog);
+ final MaterializedTableExecutorFactory materializedTableExecutorFactory =
+ lookupMaterializedTableExecutorFactory(userClassLoader);
+
return new StreamTableEnvironmentImpl(
catalogManager,
moduleManager,
@@ -162,7 +173,9 @@ public static StreamTableEnvironment create(
executionEnvironment,
planner,
executor,
- settings.isStreamingMode());
+ settings.isStreamingMode(),
+ materializedTableExecutorFactory,
+ null);
}
@Override
diff --git a/flink-table/flink-table-api-java-bridge/src/test/java/org/apache/flink/table/api/bridge/java/internal/StreamTableEnvironmentImplTest.java b/flink-table/flink-table-api-java-bridge/src/test/java/org/apache/flink/table/api/bridge/java/internal/StreamTableEnvironmentImplTest.java
index 63302f7f7889f9..e268c6187803c1 100644
--- a/flink-table/flink-table-api-java-bridge/src/test/java/org/apache/flink/table/api/bridge/java/internal/StreamTableEnvironmentImplTest.java
+++ b/flink-table/flink-table-api-java-bridge/src/test/java/org/apache/flink/table/api/bridge/java/internal/StreamTableEnvironmentImplTest.java
@@ -30,6 +30,7 @@
import org.apache.flink.table.resource.ResourceManager;
import org.apache.flink.table.utils.CatalogManagerMocks;
import org.apache.flink.table.utils.ExecutorMock;
+import org.apache.flink.table.utils.MaterializedTableExecutorFactoryMock;
import org.apache.flink.table.utils.PlannerMock;
import org.apache.flink.types.Row;
@@ -94,7 +95,9 @@ private StreamTableEnvironmentImpl getStreamTableEnvironment(
env,
new TestPlanner(elements.getTransformation()),
new ExecutorMock(),
- true);
+ true,
+ new MaterializedTableExecutorFactoryMock(),
+ null);
}
private static class TestPlanner extends PlannerMock {
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 496369e78f269d..c574a460a4fdb4 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
@@ -47,6 +47,7 @@
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.ValidationException;
import org.apache.flink.table.api.config.TableConfigOptions;
+import org.apache.flink.table.api.internal.materializedtable.DefaultMaterializedTableJobSubmitter;
import org.apache.flink.table.catalog.Catalog;
import org.apache.flink.table.catalog.CatalogBaseTable;
import org.apache.flink.table.catalog.CatalogDescriptor;
@@ -76,6 +77,9 @@
import org.apache.flink.table.delegation.InternalPlan;
import org.apache.flink.table.delegation.Parser;
import org.apache.flink.table.delegation.Planner;
+import org.apache.flink.table.delegation.materializedtable.MaterializedTableExecutor;
+import org.apache.flink.table.delegation.materializedtable.MaterializedTableExecutorFactory;
+import org.apache.flink.table.delegation.materializedtable.MaterializedTableJobSubmitter;
import org.apache.flink.table.execution.StagingSinkJobStatusHook;
import org.apache.flink.table.expressions.ApiExpressionUtils;
import org.apache.flink.table.expressions.DefaultSqlFactory;
@@ -115,6 +119,7 @@
import org.apache.flink.table.operations.ddl.AnalyzeTableOperation;
import org.apache.flink.table.operations.ddl.CompilePlanOperation;
import org.apache.flink.table.operations.ddl.CreateTableOperation;
+import org.apache.flink.table.operations.materializedtable.MaterializedTableOperation;
import org.apache.flink.table.operations.utils.ExecutableOperationUtils;
import org.apache.flink.table.operations.utils.OperationTreeBuilder;
import org.apache.flink.table.resource.ResourceManager;
@@ -132,6 +137,8 @@
import org.apache.flink.util.MutableURLClassLoader;
import org.apache.flink.util.Preconditions;
+import javax.annotation.Nullable;
+
import java.io.IOException;
import java.net.URL;
import java.util.ArrayList;
@@ -173,6 +180,10 @@ public class TableEnvironmentImpl implements TableEnvironmentInternal {
private final boolean isStreamingMode;
private final ExecutableOperation.Context operationCtx;
+ private final MaterializedTableExecutorFactory materializedTableExecutorFactory;
+ private final @Nullable MaterializedTableJobSubmitter materializedTableJobSubmitter;
+ private @Nullable MaterializedTableExecutor materializedTableExecutor;
+
private static final String UNSUPPORTED_QUERY_IN_EXECUTE_SQL_MSG =
"Unsupported SQL query! executeSql() only accepts a single SQL statement of type "
+ "CREATE TABLE, DROP TABLE, ALTER TABLE, CREATE DATABASE, DROP DATABASE, ALTER DATABASE, "
@@ -191,7 +202,9 @@ protected TableEnvironmentImpl(
Executor executor,
FunctionCatalog functionCatalog,
Planner planner,
- boolean isStreamingMode) {
+ boolean isStreamingMode,
+ MaterializedTableExecutorFactory materializedTableExecutorFactory,
+ @Nullable MaterializedTableJobSubmitter materializedTableJobSubmitter) {
this.catalogManager = catalogManager;
this.moduleManager = moduleManager;
this.resourceManager = resourceManager;
@@ -239,6 +252,8 @@ protected TableEnvironmentImpl(
resourceManager,
tableConfig,
isStreamingMode);
+ this.materializedTableExecutorFactory = materializedTableExecutorFactory;
+ this.materializedTableJobSubmitter = materializedTableJobSubmitter;
}
public static TableEnvironmentImpl create(Configuration configuration) {
@@ -318,6 +333,12 @@ public static TableEnvironmentImpl create(EnvironmentSettings settings) {
catalogManager,
functionCatalog);
+ final MaterializedTableExecutorFactory materializedTableExecutorFactory =
+ FactoryUtil.discoverFactory(
+ userClassLoader,
+ MaterializedTableExecutorFactory.class,
+ MaterializedTableExecutorFactory.DEFAULT_IDENTIFIER);
+
return new TableEnvironmentImpl(
catalogManager,
moduleManager,
@@ -326,7 +347,9 @@ public static TableEnvironmentImpl create(EnvironmentSettings settings) {
executor,
functionCatalog,
planner,
- settings.isStreamingMode());
+ settings.isStreamingMode(),
+ materializedTableExecutorFactory,
+ null);
}
@Override
@@ -1312,8 +1335,25 @@ private TableResultInternal executeQueryOperation(
}
}
+ private MaterializedTableExecutor getMaterializedTableExecutor() {
+ if (materializedTableExecutor == null) {
+ final MaterializedTableJobSubmitter jobSubmitter =
+ materializedTableJobSubmitter != null
+ ? materializedTableJobSubmitter
+ : new DefaultMaterializedTableJobSubmitter(
+ operationCtx, planner, execEnv);
+ materializedTableExecutor =
+ materializedTableExecutorFactory.createExecutor(operationCtx, jobSubmitter);
+ }
+ return materializedTableExecutor;
+ }
+
@Override
public TableResultInternal executeInternal(Operation operation) {
+ if (operation instanceof MaterializedTableOperation) {
+ return getMaterializedTableExecutor().execute((MaterializedTableOperation) operation);
+ }
+
// delegate execution to Operation if it implements ExecutableOperation
if (operation instanceof ExecutableOperation) {
return ((ExecutableOperation) operation).execute(operationCtx);
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/materializedtable/DefaultMaterializedTableJobSubmitter.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/materializedtable/DefaultMaterializedTableJobSubmitter.java
new file mode 100644
index 00000000000000..eed6b4c4190eb8
--- /dev/null
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/materializedtable/DefaultMaterializedTableJobSubmitter.java
@@ -0,0 +1,185 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.api.internal.materializedtable;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.dag.Pipeline;
+import org.apache.flink.api.dag.Transformation;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.delegation.Executor;
+import org.apache.flink.table.delegation.ExecutorFactory;
+import org.apache.flink.table.delegation.Planner;
+import org.apache.flink.table.delegation.materializedtable.MaterializedTableJobSubmitter;
+import org.apache.flink.table.delegation.materializedtable.RefreshJobResult;
+import org.apache.flink.table.delegation.materializedtable.RefreshJobTarget;
+import org.apache.flink.table.delegation.materializedtable.RefreshWorkflowContext;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.table.factories.PlannerFactoryUtil;
+import org.apache.flink.table.operations.ExecutableOperation;
+import org.apache.flink.table.operations.ModifyOperation;
+import org.apache.flink.table.operations.Operation;
+import org.apache.flink.table.workflow.WorkflowScheduler;
+
+import java.util.List;
+import java.util.Optional;
+
+import static org.apache.flink.api.common.RuntimeExecutionMode.STREAMING;
+import static org.apache.flink.configuration.ExecutionOptions.RUNTIME_MODE;
+
+/**
+ * The default implementation of a {@link MaterializedTableJobSubmitter}.
+ *
+ * It does not support application targets (see {@link #supportsApplicationTargets()}) and does
+ * not expose a {@link WorkflowScheduler}, so full-mode materialized tables cannot be created,
+ * suspended, resumed or dropped here, while a one-time REFRESH still works.
+ */
+@Internal
+public class DefaultMaterializedTableJobSubmitter implements MaterializedTableJobSubmitter {
+
+ private static final String DEFAULT_REFRESH_JOB_NAME = "materialized-table-refresh";
+
+ private final ExecutableOperation.Context operationCtx;
+ private final Planner planner;
+ private final Executor sessionExecutor;
+
+ public DefaultMaterializedTableJobSubmitter(
+ ExecutableOperation.Context operationCtx, Planner planner, Executor sessionExecutor) {
+ this.operationCtx = operationCtx;
+ this.planner = planner;
+ this.sessionExecutor = sessionExecutor;
+ }
+
+ @Override
+ public RefreshJobResult submitRefreshJob(
+ Configuration executionConfig, String insertStatement) {
+ final RefreshJobTarget target = RefreshJobTarget.forNewJob(operationCtx.getTableConfig());
+ final boolean isStreamingRefresh = STREAMING == executionConfig.get(RUNTIME_MODE);
+ final JobClient jobClient =
+ isStreamingRefresh == operationCtx.isStreamingMode()
+ ? submitWithSessionPlanner(target, executionConfig, insertStatement)
+ : submitWithThrowawayPlanner(target, executionConfig, insertStatement);
+
+ return new RefreshJobResult(target, jobClient.getJobID().toString(), jobClient);
+ }
+
+ @Override
+ public boolean supportsApplicationTargets() {
+ return false;
+ }
+
+ @Override
+ public Optional getRefreshWorkflowContext() {
+ return Optional.empty();
+ }
+
+ private static Configuration refreshConfiguration(
+ Configuration tableConfiguration,
+ Configuration executionConfig,
+ RefreshJobTarget target) {
+ final Configuration refreshConfig = new Configuration(tableConfiguration);
+ refreshConfig.addAll(executionConfig);
+ target.applyTo(refreshConfig);
+ return refreshConfig;
+ }
+
+ private JobClient submitWithSessionPlanner(
+ RefreshJobTarget target, Configuration executionConfig, String insertStatement) {
+ final ModifyOperation modifyOperation =
+ parseSingleInsertOperation(planner, insertStatement);
+ operationCtx.getResourceManager().addJarConfiguration(operationCtx.getTableConfig());
+
+ final Configuration pipelineConfig =
+ refreshConfiguration(
+ operationCtx.getTableConfig().getConfiguration(), executionConfig, target);
+ return submitPipeline(planner, modifyOperation, createRefreshExecutor(), pipelineConfig);
+ }
+
+ private JobClient submitWithThrowawayPlanner(
+ RefreshJobTarget target, Configuration executionConfig, String insertStatement) {
+ final Configuration refreshConfig =
+ refreshConfiguration(
+ operationCtx.getTableConfig().getConfiguration(), executionConfig, target);
+ final TableConfig refreshTableConfig = TableConfig.getDefault();
+ refreshTableConfig.setRootConfiguration(
+ operationCtx.getTableConfig().getRootConfiguration());
+ refreshTableConfig.addConfiguration(refreshConfig);
+
+ final Executor refreshExecutor = createRefreshExecutor();
+ final Planner refreshPlanner =
+ PlannerFactoryUtil.createPlanner(
+ refreshExecutor,
+ refreshTableConfig,
+ operationCtx.getResourceManager().getUserClassLoader(),
+ operationCtx.getModuleManager(),
+ operationCtx.getCatalogManager(),
+ operationCtx.getFunctionCatalog());
+
+ final ModifyOperation modifyOperation =
+ parseSingleInsertOperation(refreshPlanner, insertStatement);
+ operationCtx.getResourceManager().addJarConfiguration(refreshTableConfig);
+
+ return submitPipeline(
+ refreshPlanner,
+ modifyOperation,
+ refreshExecutor,
+ refreshTableConfig.getConfiguration());
+ }
+
+ private Executor createRefreshExecutor() {
+ final ExecutorFactory executorFactory =
+ FactoryUtil.discoverFactory(
+ operationCtx.getResourceManager().getUserClassLoader(),
+ ExecutorFactory.class,
+ ExecutorFactory.DEFAULT_IDENTIFIER);
+ return executorFactory.create(
+ Configuration.fromMap(sessionExecutor.getConfiguration().toMap()));
+ }
+
+ private static JobClient submitPipeline(
+ Planner planner,
+ ModifyOperation modifyOperation,
+ Executor refreshExecutor,
+ Configuration pipelineConfig) {
+ final List> transformations = planner.translate(List.of(modifyOperation));
+ final Pipeline pipeline =
+ refreshExecutor.createPipeline(
+ transformations, pipelineConfig, DEFAULT_REFRESH_JOB_NAME);
+ try {
+ return refreshExecutor.executeAsync(pipeline);
+ } catch (Exception e) {
+ throw new TableException("Failed to submit the materialized table refresh job.", e);
+ }
+ }
+
+ private static ModifyOperation parseSingleInsertOperation(
+ Planner planner, String insertStatement) {
+ final List operations = planner.getParser().parse(insertStatement);
+ if (operations.size() != 1 || !(operations.get(0) instanceof ModifyOperation)) {
+ throw new TableException(
+ String.format(
+ "The materialized table refresh statement must translate to a single"
+ + " INSERT operation, but was: %s",
+ insertStatement));
+ }
+ return (ModifyOperation) operations.get(0);
+ }
+}
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/delegation/materializedtable/MaterializedTableExecutor.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/delegation/materializedtable/MaterializedTableExecutor.java
new file mode 100644
index 00000000000000..27ea7ec20d6fe7
--- /dev/null
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/delegation/materializedtable/MaterializedTableExecutor.java
@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.delegation.materializedtable;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.internal.TableResultInternal;
+import org.apache.flink.table.operations.materializedtable.MaterializedTableOperation;
+
+/** Executes {@link MaterializedTableOperation}s through a single execution path. */
+@Internal
+public interface MaterializedTableExecutor {
+
+ /**
+ * Executes the given materialized table operation.
+ *
+ * @param op the materialized table operation to execute.
+ * @return the execution result.
+ */
+ TableResultInternal execute(MaterializedTableOperation op);
+}
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/delegation/materializedtable/MaterializedTableExecutorFactory.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/delegation/materializedtable/MaterializedTableExecutorFactory.java
new file mode 100644
index 00000000000000..1af9a2711175e1
--- /dev/null
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/delegation/materializedtable/MaterializedTableExecutorFactory.java
@@ -0,0 +1,37 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.delegation.materializedtable;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.factories.Factory;
+import org.apache.flink.table.operations.ExecutableOperation;
+
+/**
+ * A {@link Factory} for discovering and creating {@link MaterializedTableExecutor} instances via
+ * Java's Service Provider Interface (SPI).
+ */
+@Internal
+public interface MaterializedTableExecutorFactory extends Factory {
+
+ String DEFAULT_IDENTIFIER = "default";
+
+ /** Creates a corresponding {@link MaterializedTableExecutor}. */
+ MaterializedTableExecutor createExecutor(
+ ExecutableOperation.Context operationCtx, MaterializedTableJobSubmitter jobSubmitter);
+}
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/delegation/materializedtable/MaterializedTableJobSubmitter.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/delegation/materializedtable/MaterializedTableJobSubmitter.java
new file mode 100644
index 00000000000000..b7fcb5e977ea5a
--- /dev/null
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/delegation/materializedtable/MaterializedTableJobSubmitter.java
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.delegation.materializedtable;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.configuration.Configuration;
+
+import java.util.Optional;
+
+/**
+ * Provides the caller-specific refresh-job submission and scheduling capabilities a {@link
+ * MaterializedTableExecutor} needs to run materialized table operations.
+ */
+@Internal
+public interface MaterializedTableJobSubmitter {
+
+ /**
+ * Resolves the target of a refresh job and submits it there with the given execution
+ * configuration and INSERT statement.
+ *
+ * @param executionConfig the configuration used to submit the refresh job.
+ * @param insertStatement the INSERT statement executed to refresh the materialized table.
+ * @return the result identifying the submitted refresh job and the target to record for it.
+ */
+ RefreshJobResult submitRefreshJob(Configuration executionConfig, String insertStatement);
+
+ /**
+ * Whether refresh jobs can be deployed to application targets, each in its own cluster.
+ *
+ * Returning {@code true} also switches the executor to the SQL Gateway's behaviour: it skips
+ * the target check before any catalog change and the check that a restart stays on the recorded
+ * target and cluster.
+ */
+ boolean supportsApplicationTargets();
+
+ /**
+ * Returns the periodic-refresh scheduling capability for full-mode materialized tables, or
+ * {@link Optional#empty()} when this job submitter cannot schedule refresh workflows.
+ */
+ Optional getRefreshWorkflowContext();
+}
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/delegation/materializedtable/RefreshJobResult.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/delegation/materializedtable/RefreshJobResult.java
new file mode 100644
index 00000000000000..44c62a963ae8ff
--- /dev/null
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/delegation/materializedtable/RefreshJobResult.java
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.delegation.materializedtable;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.core.execution.JobClient;
+
+import javax.annotation.Nullable;
+
+import java.util.Optional;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * The result of submitting a materialized table refresh job, describing where the job was deployed
+ * and how to identify it. It optionally carries the client returned at submission.
+ */
+@Internal
+public final class RefreshJobResult {
+
+ private final RefreshJobTarget target;
+ private final String jobId;
+ @Nullable private final JobClient jobClient;
+
+ public RefreshJobResult(RefreshJobTarget target, String jobId, @Nullable JobClient jobClient) {
+ this.target = checkNotNull(target);
+ this.jobId = checkNotNull(jobId);
+ this.jobClient = jobClient;
+ checkArgument(
+ jobClient == null || jobClient.getJobID().toHexString().equals(jobId),
+ "The job client controls job %s, not refresh job %s.",
+ jobClient == null ? null : jobClient.getJobID(),
+ jobId);
+ }
+
+ public RefreshJobTarget getTarget() {
+ return target;
+ }
+
+ public String getJobId() {
+ return jobId;
+ }
+
+ public Optional getJobClient() {
+ return Optional.ofNullable(jobClient);
+ }
+}
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/delegation/materializedtable/RefreshJobTarget.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/delegation/materializedtable/RefreshJobTarget.java
new file mode 100644
index 00000000000000..81d6dfe6cae9d5
--- /dev/null
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/delegation/materializedtable/RefreshJobTarget.java
@@ -0,0 +1,196 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.delegation.materializedtable;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.DeploymentOptions;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.catalog.ObjectIdentifier;
+import org.apache.flink.table.refresh.ContinuousRefreshHandler;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/** The execution target and cluster id a materialized table refresh job is submitted to. */
+@Internal
+public final class RefreshJobTarget {
+
+ private static final String YARN_APPLICATION_ID_KEY = "yarn.application.id";
+ private static final String KUBERNETES_CLUSTER_ID_KEY = "kubernetes.cluster-id";
+
+ private static final String LOCAL_TARGET = "local";
+ private static final String EMBEDDED_TARGET = "embedded";
+ private static final String REMOTE_TARGET = "remote";
+ private static final String STANDALONE_CLUSTER_ID = "StandaloneClusterId";
+
+ private final String executionTarget;
+ private final String clusterId;
+
+ public RefreshJobTarget(String executionTarget, String clusterId) {
+ this.executionTarget = checkNotNull(executionTarget);
+ this.clusterId = checkNotNull(clusterId);
+ checkArgument(!executionTarget.isEmpty(), "The execution target must not be empty.");
+ }
+
+ /** The target a new refresh job of a program is submitted to. */
+ public static RefreshJobTarget forNewJob(ReadableConfig config) {
+ final String executionTarget = config.getOptional(DeploymentOptions.TARGET).orElse(null);
+ if (executionTarget == null
+ || executionTarget.isEmpty()
+ || LOCAL_TARGET.equals(executionTarget)
+ || EMBEDDED_TARGET.equals(executionTarget)) {
+ throw new ValidationException(
+ String.format(
+ "Unsupported execution target detected: %s. Materialized table refresh "
+ + "jobs need a session cluster: set 'execution.target' to "
+ + "'remote', 'yarn-session' or 'kubernetes-session' (with "
+ + "'yarn.application.id' or 'kubernetes.cluster-id').",
+ executionTarget));
+ }
+ final RefreshJobTarget unresolvedTarget = new RefreshJobTarget(executionTarget, "");
+ if (unresolvedTarget.isApplicationTarget()) {
+ throw new ValidationException(
+ String.format(
+ "Application-mode materialized table refresh is not supported in this "
+ + "environment: execution target '%s' requires the SQL Gateway. Use a "
+ + "session cluster target such as 'remote', 'yarn-session' or "
+ + "'kubernetes-session'.",
+ executionTarget));
+ }
+ return new RefreshJobTarget(executionTarget, unresolvedTarget.resolveClusterId(config));
+ }
+
+ /** The target for restarting a recorded refresh job. */
+ public static RefreshJobTarget forRestart(
+ ReadableConfig config,
+ ObjectIdentifier tableIdentifier,
+ ContinuousRefreshHandler refreshHandler) {
+ final RefreshJobTarget recordedTarget = recordedIn(refreshHandler);
+ if (recordedTarget.isApplicationTarget()) {
+ throw new ValidationException(
+ String.format(
+ "The continuous refresh job of materialized table %s runs in its own "
+ + "application cluster (%s). Restarting it deploys a new "
+ + "application cluster, which only the SQL Gateway can do, so "
+ + "resume, alter or redeploy the table through the SQL Gateway. "
+ + "Suspending, refreshing and dropping it work here.",
+ tableIdentifier, recordedTarget.describe()));
+ }
+
+ final RefreshJobTarget programTarget = forNewJob(config);
+ final boolean sameTarget =
+ recordedTarget.executionTarget.equals(programTarget.executionTarget);
+ final boolean sameCluster =
+ !recordedTarget.identifiesClusterById()
+ || recordedTarget.clusterId.equals(programTarget.clusterId);
+ if (!sameTarget || !sameCluster) {
+ throw new ValidationException(
+ String.format(
+ "The continuous refresh job of materialized table %s runs on %s, but "
+ + "this program submits refresh jobs to %s. RESUME, ALTER and a "
+ + "redeploy never move a refresh job to another cluster. Set "
+ + "'execution.target' and its cluster id to the recorded values "
+ + "to restart it here.",
+ tableIdentifier, recordedTarget.describe(), programTarget.describe()));
+ }
+ return programTarget;
+ }
+
+ /** The target a refresh handler records. */
+ public static RefreshJobTarget recordedIn(ContinuousRefreshHandler refreshHandler) {
+ final String clusterId = refreshHandler.getClusterId();
+ return new RefreshJobTarget(
+ refreshHandler.getExecutionTarget(), clusterId == null ? "" : clusterId);
+ }
+
+ public String getExecutionTarget() {
+ return executionTarget;
+ }
+
+ /** The cluster id, or an empty string when it is not known. */
+ public String getClusterId() {
+ return clusterId;
+ }
+
+ /** Whether the target deploys each job in its own application cluster. */
+ public boolean isApplicationTarget() {
+ return executionTarget.endsWith("application");
+ }
+
+ public void applyTo(Configuration configuration) {
+ configuration.set(DeploymentOptions.TARGET, executionTarget);
+ if (identifiesClusterById()) {
+ configuration.setString(clusterIdKey().get(), clusterId);
+ }
+ }
+
+ public Map toClusterInfo() {
+ final Map clusterInfo = new HashMap<>();
+ clusterInfo.put(DeploymentOptions.TARGET.key(), executionTarget);
+ clusterIdKey().ifPresent(key -> clusterInfo.put(key, clusterId));
+ return Collections.unmodifiableMap(clusterInfo);
+ }
+
+ private Optional clusterIdKey() {
+ if (executionTarget.startsWith("yarn")) {
+ return Optional.of(YARN_APPLICATION_ID_KEY);
+ } else if (executionTarget.startsWith("kubernetes")) {
+ return Optional.of(KUBERNETES_CLUSTER_ID_KEY);
+ }
+ return Optional.empty();
+ }
+
+ private boolean identifiesClusterById() {
+ return !clusterId.isEmpty() && clusterIdKey().isPresent();
+ }
+
+ private String describe() {
+ if (identifiesClusterById()) {
+ return String.format(
+ "execution.target '%s', %s '%s'",
+ executionTarget, clusterIdKey().get(), clusterId);
+ }
+ return String.format("execution.target '%s'", executionTarget);
+ }
+
+ private String resolveClusterId(ReadableConfig config) {
+ final Optional clusterIdKey = clusterIdKey();
+ if (clusterIdKey.isEmpty()) {
+ // record StandaloneClusterId#toString for 'remote'
+ return REMOTE_TARGET.equals(executionTarget) ? STANDALONE_CLUSTER_ID : "";
+ }
+ return config.getOptional(
+ ConfigOptions.key(clusterIdKey.get()).stringType().noDefaultValue())
+ .filter(configuredClusterId -> !configuredClusterId.isBlank())
+ .orElseThrow(
+ () ->
+ new ValidationException(
+ String.format(
+ "Execution target '%s' requires '%s' to be configured to submit or control a materialized table refresh job.",
+ executionTarget, clusterIdKey.get())));
+ }
+}
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/delegation/materializedtable/RefreshWorkflowContext.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/delegation/materializedtable/RefreshWorkflowContext.java
new file mode 100644
index 00000000000000..2a815c68cb5fec
--- /dev/null
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/delegation/materializedtable/RefreshWorkflowContext.java
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.delegation.materializedtable;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.refresh.RefreshHandler;
+import org.apache.flink.table.workflow.WorkflowScheduler;
+
+import java.util.Map;
+
+/**
+ * The capability to schedule periodic refresh workflows for full-mode materialized tables, exposed
+ * by a {@link MaterializedTableJobSubmitter} only when the deployment can provide it.
+ */
+@Internal
+public interface RefreshWorkflowContext {
+
+ /** The scheduler that creates, modifies and deletes periodic refresh workflows. */
+ WorkflowScheduler extends RefreshHandler> scheduler();
+
+ /** The REST endpoint URL a scheduled workflow uses to reach the cluster managing refreshes. */
+ String restEndpointUrl();
+
+ /** The session configuration a scheduled refresh job is initialized with. */
+ Map sessionInitializationConf();
+}
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/materializedtable/AlterMaterializedTableAsQueryOperation.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/materializedtable/AlterMaterializedTableAsQueryOperation.java
index 0dcbfd72926658..21354a5ab0c89c 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/materializedtable/AlterMaterializedTableAsQueryOperation.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/materializedtable/AlterMaterializedTableAsQueryOperation.java
@@ -32,7 +32,7 @@
* Operation to describe an ALTER MATERIALIZED TABLE AS query operation. The operation is not
* executable and only serves as a wrapper.
*
- * As an example see {@link MaterializedTableManager} where it is transformed to {@link
+ *
As an example see {@code DefaultMaterializedTableExecutor} where it is transformed to {@link
* AlterMaterializedTableChangeOperation} before execution.
*/
@Internal
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/materializedtable/AlterMaterializedTableRefreshOperation.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/materializedtable/AlterMaterializedTableRefreshOperation.java
index 5212cf58c7ff48..c98f112f076e1b 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/materializedtable/AlterMaterializedTableRefreshOperation.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/materializedtable/AlterMaterializedTableRefreshOperation.java
@@ -23,6 +23,8 @@
import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.operations.OperationUtils;
+import javax.annotation.Nullable;
+
import java.util.Map;
/**
@@ -33,17 +35,48 @@
public class AlterMaterializedTableRefreshOperation extends AlterMaterializedTableOperation {
private final Map partitionSpec;
+ private final boolean isPeriodic;
+ private final @Nullable String scheduleTime;
+ private final Map dynamicOptions;
public AlterMaterializedTableRefreshOperation(
ObjectIdentifier tableIdentifier, Map partitionSpec) {
+ this(tableIdentifier, partitionSpec, false, null, Map.of());
+ }
+
+ private AlterMaterializedTableRefreshOperation(
+ ObjectIdentifier tableIdentifier,
+ Map partitionSpec,
+ boolean isPeriodic,
+ @Nullable String scheduleTime,
+ Map dynamicOptions) {
super(tableIdentifier);
this.partitionSpec = partitionSpec;
+ this.isPeriodic = isPeriodic;
+ this.scheduleTime = scheduleTime;
+ this.dynamicOptions = dynamicOptions;
+ }
+
+ public static AlterMaterializedTableRefreshOperation oneTime(
+ ObjectIdentifier tableIdentifier,
+ Map partitionSpec,
+ Map dynamicOptions) {
+ return new AlterMaterializedTableRefreshOperation(
+ tableIdentifier, partitionSpec, false, null, dynamicOptions);
+ }
+
+ public static AlterMaterializedTableRefreshOperation periodic(
+ ObjectIdentifier tableIdentifier,
+ @Nullable String scheduleTime,
+ Map dynamicOptions) {
+ return new AlterMaterializedTableRefreshOperation(
+ tableIdentifier, Map.of(), true, scheduleTime, dynamicOptions);
}
@Override
public TableResultInternal execute(Context ctx) {
- // execute AlterMaterializedTableRefreshOperation in SqlGateway OperationExecutor.
- // see more at MaterializedTableManager#callAlterMaterializedTableRefreshOperation
+ // TableEnvironmentImpl#executeInternal routes every MaterializedTableOperation
+ // to the MaterializedTableExecutor before ExecutableOperation dispatch.
throw new UnsupportedOperationException(
"AlterMaterializedTableRefreshOperation does not support ExecutableOperation yet.");
}
@@ -52,6 +85,18 @@ public Map getPartitionSpec() {
return partitionSpec;
}
+ public boolean isPeriodic() {
+ return isPeriodic;
+ }
+
+ public @Nullable String getScheduleTime() {
+ return scheduleTime;
+ }
+
+ public Map getDynamicOptions() {
+ return dynamicOptions;
+ }
+
@Override
public String asSummaryString() {
StringBuilder sb =
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/planner/loader/PlannerModule.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/planner/loader/PlannerModule.java
index 1a8375b5039efe..1e457a571533cc 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/planner/loader/PlannerModule.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/planner/loader/PlannerModule.java
@@ -27,6 +27,7 @@
import org.apache.flink.table.api.TableException;
import org.apache.flink.table.delegation.ExecutorFactory;
import org.apache.flink.table.delegation.PlannerFactory;
+import org.apache.flink.table.delegation.materializedtable.MaterializedTableExecutorFactory;
import org.apache.flink.table.factories.FactoryUtil;
import org.apache.flink.util.FileUtils;
import org.apache.flink.util.IOUtils;
@@ -169,6 +170,13 @@ public PlannerFactory loadPlannerFactory() {
this.submoduleClassLoader, PlannerFactory.class, PlannerFactory.DEFAULT_IDENTIFIER);
}
+ public MaterializedTableExecutorFactory loadMaterializedTableExecutorFactory() {
+ return FactoryUtil.discoverFactory(
+ this.submoduleClassLoader,
+ MaterializedTableExecutorFactory.class,
+ MaterializedTableExecutorFactory.DEFAULT_IDENTIFIER);
+ }
+
/**
* A class loader extending {@link ComponentClassLoader} which overwrites method{@link #addURL}
* to enable it can add url to component classloader.
diff --git a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/utils/MaterializedTableExecutorFactoryMock.java b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/utils/MaterializedTableExecutorFactoryMock.java
new file mode 100644
index 00000000000000..36fe04ac2b500a
--- /dev/null
+++ b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/utils/MaterializedTableExecutorFactoryMock.java
@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.utils;
+
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.table.delegation.materializedtable.MaterializedTableExecutor;
+import org.apache.flink.table.delegation.materializedtable.MaterializedTableExecutorFactory;
+import org.apache.flink.table.delegation.materializedtable.MaterializedTableJobSubmitter;
+import org.apache.flink.table.operations.ExecutableOperation;
+
+import java.util.Set;
+
+/** Mocking {@link MaterializedTableExecutorFactory} for tests without the planner. */
+public class MaterializedTableExecutorFactoryMock implements MaterializedTableExecutorFactory {
+
+ @Override
+ public MaterializedTableExecutor createExecutor(
+ ExecutableOperation.Context operationCtx, MaterializedTableJobSubmitter jobSubmitter) {
+ return operation -> {
+ throw new UnsupportedOperationException(
+ "Materialized table execution is not supported by this test mock.");
+ };
+ }
+
+ @Override
+ public String factoryIdentifier() {
+ return DEFAULT_IDENTIFIER;
+ }
+
+ @Override
+ public Set> requiredOptions() {
+ return Set.of();
+ }
+
+ @Override
+ public Set> optionalOptions() {
+ return Set.of();
+ }
+}
diff --git a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/utils/TableEnvironmentMock.java b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/utils/TableEnvironmentMock.java
index ff046ed48581f8..08930c5f888e4c 100644
--- a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/utils/TableEnvironmentMock.java
+++ b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/utils/TableEnvironmentMock.java
@@ -57,7 +57,9 @@ protected TableEnvironmentMock(
executor,
functionCatalog,
planner,
- isStreamingMode);
+ isStreamingMode,
+ new MaterializedTableExecutorFactoryMock(),
+ null);
this.catalogManager = catalogManager;
this.executor = executor;
this.functionCatalog = functionCatalog;
diff --git a/flink-table/flink-table-api-scala-bridge/src/test/resources/META-INF/services/org.apache.flink.table.factories.Factory b/flink-table/flink-table-api-scala-bridge/src/test/resources/META-INF/services/org.apache.flink.table.factories.Factory
new file mode 100644
index 00000000000000..bf08ae4ef0730e
--- /dev/null
+++ b/flink-table/flink-table-api-scala-bridge/src/test/resources/META-INF/services/org.apache.flink.table.factories.Factory
@@ -0,0 +1,18 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# The Scala StreamTableEnvironmentImpl looks this factory up when it is created, and this module's
+# tests run without the planner.
+org.apache.flink.table.utils.MaterializedTableExecutorFactoryMock
diff --git a/flink-table/flink-table-planner-loader/src/main/java/org/apache/flink/table/planner/loader/DelegateMaterializedTableExecutorFactory.java b/flink-table/flink-table-planner-loader/src/main/java/org/apache/flink/table/planner/loader/DelegateMaterializedTableExecutorFactory.java
new file mode 100644
index 00000000000000..10583e967bbc38
--- /dev/null
+++ b/flink-table/flink-table-planner-loader/src/main/java/org/apache/flink/table/planner/loader/DelegateMaterializedTableExecutorFactory.java
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.loader;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.delegation.materializedtable.MaterializedTableExecutor;
+import org.apache.flink.table.delegation.materializedtable.MaterializedTableExecutorFactory;
+import org.apache.flink.table.delegation.materializedtable.MaterializedTableJobSubmitter;
+import org.apache.flink.table.operations.ExecutableOperation;
+
+/** Delegate of {@link MaterializedTableExecutorFactory}. */
+@Internal
+public class DelegateMaterializedTableExecutorFactory
+ extends BaseDelegateFactory
+ implements MaterializedTableExecutorFactory {
+
+ public DelegateMaterializedTableExecutorFactory() {
+ super(PlannerModule.getInstance().loadMaterializedTableExecutorFactory());
+ }
+
+ @Override
+ public MaterializedTableExecutor createExecutor(
+ ExecutableOperation.Context operationCtx, MaterializedTableJobSubmitter jobSubmitter) {
+ return delegate.createExecutor(operationCtx, jobSubmitter);
+ }
+}
diff --git a/flink-table/flink-table-planner-loader/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory b/flink-table/flink-table-planner-loader/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory
index a932494d960ed8..415099ac45d2d7 100644
--- a/flink-table/flink-table-planner-loader/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory
+++ b/flink-table/flink-table-planner-loader/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory
@@ -15,3 +15,4 @@
org.apache.flink.table.planner.loader.DelegateExecutorFactory
org.apache.flink.table.planner.loader.DelegatePlannerFactory
+org.apache.flink.table.planner.loader.DelegateMaterializedTableExecutorFactory
diff --git a/flink-table/flink-table-planner-loader/src/test/java/org/apache/flink/table/planner/loader/LoaderITCase.java b/flink-table/flink-table-planner-loader/src/test/java/org/apache/flink/table/planner/loader/LoaderITCase.java
index f245fcf4cf2850..325aaa972066fb 100644
--- a/flink-table/flink-table-planner-loader/src/test/java/org/apache/flink/table/planner/loader/LoaderITCase.java
+++ b/flink-table/flink-table-planner-loader/src/test/java/org/apache/flink/table/planner/loader/LoaderITCase.java
@@ -22,6 +22,7 @@
import org.apache.flink.configuration.ConfigurationUtils;
import org.apache.flink.table.delegation.ExecutorFactory;
import org.apache.flink.table.delegation.PlannerFactory;
+import org.apache.flink.table.delegation.materializedtable.MaterializedTableExecutorFactory;
import org.apache.flink.table.factories.FactoryUtil;
import org.apache.flink.util.FileUtils;
import org.apache.flink.util.TestLogger;
@@ -66,6 +67,21 @@ public void testExecutorFactory() {
.isEqualTo(ExecutorFactory.DEFAULT_IDENTIFIER);
}
+ @Test
+ public void testMaterializedTableExecutorFactory() {
+ MaterializedTableExecutorFactory materializedTableExecutorFactory =
+ FactoryUtil.discoverFactory(
+ LoaderITCase.class.getClassLoader(),
+ MaterializedTableExecutorFactory.class,
+ MaterializedTableExecutorFactory.DEFAULT_IDENTIFIER);
+
+ assertThat(materializedTableExecutorFactory)
+ .isNotNull()
+ .isInstanceOf(DelegateMaterializedTableExecutorFactory.class);
+ assertThat(materializedTableExecutorFactory.factoryIdentifier())
+ .isEqualTo(MaterializedTableExecutorFactory.DEFAULT_IDENTIFIER);
+ }
+
@Test
public void testPlannerFactory() {
assertThat(
diff --git a/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/materializedtable/MaterializedTableManager.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/delegation/materializedtable/DefaultMaterializedTableExecutor.java
similarity index 51%
rename from flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/materializedtable/MaterializedTableManager.java
rename to flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/delegation/materializedtable/DefaultMaterializedTableExecutor.java
index cc01f96e7ece27..bb9eb9fc49acb1 100644
--- a/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/materializedtable/MaterializedTableManager.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/delegation/materializedtable/DefaultMaterializedTableExecutor.java
@@ -16,21 +16,20 @@
* limitations under the License.
*/
-package org.apache.flink.table.gateway.service.materializedtable;
+package org.apache.flink.table.planner.delegation.materializedtable;
import org.apache.flink.annotation.Internal;
import org.apache.flink.annotation.VisibleForTesting;
-import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.JobStatus;
-import org.apache.flink.client.deployment.DefaultClusterClientServiceLoader;
-import org.apache.flink.client.deployment.application.ApplicationConfiguration;
-import org.apache.flink.client.deployment.application.cli.ApplicationClusterDeployer;
import org.apache.flink.configuration.CheckpointingOptions;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.ResultKind;
+import org.apache.flink.table.api.TableException;
import org.apache.flink.table.api.ValidationException;
-import org.apache.flink.table.api.config.TableConfigOptions;
-import org.apache.flink.table.api.internal.TableEnvironmentInternal;
+import org.apache.flink.table.api.internal.StaticResultProvider;
+import org.apache.flink.table.api.internal.TableResultImpl;
+import org.apache.flink.table.api.internal.TableResultInternal;
import org.apache.flink.table.catalog.CatalogMaterializedTable;
import org.apache.flink.table.catalog.CatalogMaterializedTable.RefreshMode;
import org.apache.flink.table.catalog.CatalogMaterializedTable.RefreshStatus;
@@ -45,17 +44,12 @@
import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.StringData;
-import org.apache.flink.table.factories.WorkflowSchedulerFactoryUtil;
-import org.apache.flink.table.gateway.api.operation.OperationHandle;
-import org.apache.flink.table.gateway.api.results.ResultSet;
-import org.apache.flink.table.gateway.api.utils.SqlGatewayException;
-import org.apache.flink.table.gateway.rest.SqlGatewayRestEndpointFactory;
-import org.apache.flink.table.gateway.rest.util.SqlGatewayRestOptions;
-import org.apache.flink.table.gateway.service.operation.OperationExecutor;
-import org.apache.flink.table.gateway.service.result.ResultFetcher;
-import org.apache.flink.table.gateway.service.utils.SqlExecutionException;
-import org.apache.flink.table.operations.command.DescribeJobOperation;
-import org.apache.flink.table.operations.command.StopJobOperation;
+import org.apache.flink.table.delegation.materializedtable.MaterializedTableExecutor;
+import org.apache.flink.table.delegation.materializedtable.MaterializedTableJobSubmitter;
+import org.apache.flink.table.delegation.materializedtable.RefreshJobResult;
+import org.apache.flink.table.delegation.materializedtable.RefreshJobTarget;
+import org.apache.flink.table.delegation.materializedtable.RefreshWorkflowContext;
+import org.apache.flink.table.operations.ExecutableOperation;
import org.apache.flink.table.operations.materializedtable.AlterMaterializedTableChangeOperation;
import org.apache.flink.table.operations.materializedtable.AlterMaterializedTableRefreshOperation;
import org.apache.flink.table.operations.materializedtable.AlterMaterializedTableResumeOperation;
@@ -68,7 +62,6 @@
import org.apache.flink.table.refresh.ContinuousRefreshHandlerSerializer;
import org.apache.flink.table.refresh.RefreshHandler;
import org.apache.flink.table.refresh.RefreshHandlerSerializer;
-import org.apache.flink.table.runtime.application.SqlDriver;
import org.apache.flink.table.types.logical.LogicalTypeFamily;
import org.apache.flink.table.workflow.CreatePeriodicRefreshWorkflow;
import org.apache.flink.table.workflow.CreateRefreshWorkflow;
@@ -77,18 +70,15 @@
import org.apache.flink.table.workflow.ResumeRefreshWorkflow;
import org.apache.flink.table.workflow.SuspendRefreshWorkflow;
import org.apache.flink.table.workflow.WorkflowScheduler;
+import org.apache.flink.types.Row;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import javax.annotation.Nullable;
-
import java.io.IOException;
-import java.net.URLClassLoader;
import java.time.Duration;
import java.time.ZoneId;
import java.util.ArrayList;
-import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -100,11 +90,8 @@
import static org.apache.flink.api.common.RuntimeExecutionMode.BATCH;
import static org.apache.flink.api.common.RuntimeExecutionMode.STREAMING;
-import static org.apache.flink.configuration.CheckpointingOptions.SAVEPOINT_DIRECTORY;
-import static org.apache.flink.configuration.DeploymentOptions.TARGET;
import static org.apache.flink.configuration.ExecutionOptions.RUNTIME_MODE;
import static org.apache.flink.configuration.PipelineOptions.NAME;
-import static org.apache.flink.configuration.PipelineOptionsInternal.PIPELINE_FIXED_JOB_ID;
import static org.apache.flink.configuration.StateRecoveryOptions.SAVEPOINT_PATH;
import static org.apache.flink.table.api.config.MaterializedTableConfigOptions.DATE_FORMATTER;
import static org.apache.flink.table.api.config.MaterializedTableConfigOptions.PARTITION_FIELDS;
@@ -112,115 +99,87 @@
import static org.apache.flink.table.api.internal.TableResultInternal.TABLE_RESULT_OK;
import static org.apache.flink.table.catalog.CatalogBaseTable.TableKind.MATERIALIZED_TABLE;
import static org.apache.flink.table.catalog.IntervalFreshness.convertFreshnessToCron;
-import static org.apache.flink.table.factories.WorkflowSchedulerFactoryUtil.WORKFLOW_SCHEDULER_PREFIX;
-import static org.apache.flink.table.gateway.api.endpoint.SqlGatewayEndpointFactoryUtils.getEndpointConfig;
-import static org.apache.flink.table.gateway.service.utils.Constants.CLUSTER_INFO;
-import static org.apache.flink.table.gateway.service.utils.Constants.JOB_ID;
import static org.apache.flink.table.utils.DateTimeUtils.formatTimestampStringWithOffset;
-/** Manager is responsible for execute the {@link MaterializedTableOperation}. */
+/** Default executor for {@link MaterializedTableOperation}s. */
@Internal
-public class MaterializedTableManager {
-
- private static final Logger LOG = LoggerFactory.getLogger(MaterializedTableManager.class);
-
- private final URLClassLoader userCodeClassLoader;
-
- private final @Nullable WorkflowScheduler extends RefreshHandler> workflowScheduler;
-
- private final String restEndpointUrl;
-
- public MaterializedTableManager(
- Configuration configuration, URLClassLoader userCodeClassLoader) {
- this.userCodeClassLoader = userCodeClassLoader;
- this.restEndpointUrl = buildRestEndpointUrl(configuration);
- this.workflowScheduler = buildWorkflowScheduler(configuration, userCodeClassLoader);
- }
-
- private String buildRestEndpointUrl(Configuration configuration) {
- Configuration restEndpointConfig =
- Configuration.fromMap(
- getEndpointConfig(configuration, SqlGatewayRestEndpointFactory.IDENTIFIER));
- String address = restEndpointConfig.get(SqlGatewayRestOptions.ADDRESS);
- int port = restEndpointConfig.get(SqlGatewayRestOptions.PORT);
-
- return String.format("http://%s:%s", address, port);
- }
-
- private WorkflowScheduler extends RefreshHandler> buildWorkflowScheduler(
- Configuration configuration, URLClassLoader userCodeClassLoader) {
- return WorkflowSchedulerFactoryUtil.createWorkflowScheduler(
- configuration, userCodeClassLoader);
- }
-
- public void open() throws Exception {
- if (workflowScheduler != null) {
- workflowScheduler.open();
- }
+public class DefaultMaterializedTableExecutor implements MaterializedTableExecutor {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(DefaultMaterializedTableExecutor.class);
+
+ private static final String LEFT_IN_INITIALIZING_STATUS =
+ " The table was left in INITIALIZING status. No materialized table statement can drop a"
+ + " table in this status or move it out of it; remove it from the catalog directly.";
+
+ private final ExecutableOperation.Context operationCtx;
+ private final MaterializedTableJobSubmitter jobSubmitter;
+ private final RefreshJobController refreshJobController;
+
+ public DefaultMaterializedTableExecutor(
+ ExecutableOperation.Context operationCtx, MaterializedTableJobSubmitter jobSubmitter) {
+ this.operationCtx = operationCtx;
+ this.jobSubmitter = jobSubmitter;
+ this.refreshJobController =
+ new RefreshJobController(
+ operationCtx.getTableConfig(),
+ operationCtx.getResourceManager().getUserClassLoader());
}
- public void close() throws Exception {
- if (workflowScheduler != null) {
- workflowScheduler.close();
- }
- }
+ @Override
+ public TableResultInternal execute(MaterializedTableOperation op) {
+ try {
+ if (op instanceof CreateMaterializedTableOperation) {
+ return callCreateMaterializedTableOperation((CreateMaterializedTableOperation) op);
+ } else if (op instanceof AlterMaterializedTableRefreshOperation) {
+ return callAlterMaterializedTableRefreshOperation(
+ (AlterMaterializedTableRefreshOperation) op);
+ } else if (op instanceof AlterMaterializedTableSuspendOperation) {
+ return callAlterMaterializedTableSuspend(
+ (AlterMaterializedTableSuspendOperation) op);
+ } else if (op instanceof AlterMaterializedTableResumeOperation) {
+ return callAlterMaterializedTableResume((AlterMaterializedTableResumeOperation) op);
+ } else if (op instanceof DropMaterializedTableOperation) {
+ return callDropMaterializedTableOperation((DropMaterializedTableOperation) op);
+ } else if (op instanceof AlterMaterializedTableChangeOperation) {
+ return callAlterMaterializedTableChangeOperation(
+ (AlterMaterializedTableChangeOperation) op);
+ } else if (op instanceof ConvertTableToMaterializedTableOperation) {
+ return callConvertTableToMaterializedTableOperation(
+ (ConvertTableToMaterializedTableOperation) op);
+ }
- public ResultFetcher callMaterializedTableOperation(
- OperationExecutor operationExecutor,
- OperationHandle handle,
- MaterializedTableOperation op) {
- if (op instanceof CreateMaterializedTableOperation) {
- return callCreateMaterializedTableOperation(
- operationExecutor, handle, (CreateMaterializedTableOperation) op);
- } else if (op instanceof AlterMaterializedTableRefreshOperation) {
- return callAlterMaterializedTableRefreshOperation(
- operationExecutor, handle, (AlterMaterializedTableRefreshOperation) op);
- } else if (op instanceof AlterMaterializedTableSuspendOperation) {
- return callAlterMaterializedTableSuspend(
- operationExecutor, handle, (AlterMaterializedTableSuspendOperation) op);
- } else if (op instanceof AlterMaterializedTableResumeOperation) {
- return callAlterMaterializedTableResume(
- operationExecutor, handle, (AlterMaterializedTableResumeOperation) op);
- } else if (op instanceof DropMaterializedTableOperation) {
- return callDropMaterializedTableOperation(
- operationExecutor, handle, (DropMaterializedTableOperation) op);
- } else if (op instanceof AlterMaterializedTableChangeOperation) {
- return callAlterMaterializedTableChangeOperation(
- operationExecutor, handle, (AlterMaterializedTableChangeOperation) op);
- } else if (op instanceof ConvertTableToMaterializedTableOperation) {
- return callConvertTableToMaterializedTableOperation(
- operationExecutor, handle, (ConvertTableToMaterializedTableOperation) op);
+ throw new TableException(
+ String.format(
+ "Unsupported Operation %s for materialized table.",
+ op.asSummaryString()));
+ } finally {
+ refreshJobController.releaseRetrievedJobClients();
}
-
- throw new SqlExecutionException(
- String.format(
- "Unsupported Operation %s for materialized table.", op.asSummaryString()));
}
- private ResultFetcher callCreateMaterializedTableOperation(
- OperationExecutor operationExecutor,
- OperationHandle handle,
+ private TableResultInternal callCreateMaterializedTableOperation(
CreateMaterializedTableOperation createMaterializedTableOperation) {
ResolvedCatalogMaterializedTable materializedTable =
createMaterializedTableOperation.getCatalogMaterializedTable();
if (RefreshMode.CONTINUOUS == materializedTable.getRefreshMode()) {
- createMaterializedTableInContinuousMode(
- operationExecutor, handle, createMaterializedTableOperation);
+ checkSubmissionTargetBeforeChanges();
+ createMaterializedTableInContinuousMode(createMaterializedTableOperation);
} else {
createMaterializedTableInFullMode(
- operationExecutor, handle, createMaterializedTableOperation);
+ createMaterializedTableOperation,
+ requireRefreshWorkflowContext(
+ "The workflow scheduler must be configured when creating materialized table in full refresh mode."));
}
// Just return ok to unify different refresh job info of continuous and full mode, user
// should get the refresh job info via desc table.
- return ResultFetcher.fromTableResult(handle, TABLE_RESULT_OK, false);
+ return TABLE_RESULT_OK;
}
private void createMaterializedTableInContinuousMode(
- OperationExecutor operationExecutor,
- OperationHandle handle,
CreateMaterializedTableOperation createMaterializedTableOperation) {
// create materialized table first
- operationExecutor.callExecutableOperation(handle, createMaterializedTableOperation);
+ createMaterializedTableOperation.execute(operationCtx);
ObjectIdentifier materializedTableIdentifier =
createMaterializedTableOperation.getTableIdentifier();
@@ -228,20 +187,21 @@ private void createMaterializedTableInContinuousMode(
createMaterializedTableOperation.getCatalogMaterializedTable();
try {
- executeContinuousRefreshJob(
- operationExecutor,
- handle,
- catalogMaterializedTable,
- materializedTableIdentifier,
- Map.of(),
- Optional.empty());
+ executeContinuousRefreshJobOrCancelIt(
+ catalogMaterializedTable, materializedTableIdentifier);
} catch (Exception e) {
// drop materialized table if submitting the Flink streaming job encounters an
- // exception. Thus, weak
- // atomicity is guaranteed
- operationExecutor.callExecutableOperation(
- handle, new DropMaterializedTableOperation(materializedTableIdentifier, true));
- throw new SqlExecutionException(
+ // exception. Thus, weak atomicity is guaranteed
+ cleanUpAfterFailure(
+ () ->
+ new DropMaterializedTableOperation(materializedTableIdentifier, true)
+ .execute(operationCtx),
+ e,
+ String.format(
+ "Failed to submit continuous refresh job for materialized table %s, and dropping the table failed as well."
+ + LEFT_IN_INITIALIZING_STATUS,
+ materializedTableIdentifier));
+ throw new TableException(
String.format(
"Failed to submit continuous refresh job for materialized table %s.",
materializedTableIdentifier),
@@ -250,15 +210,10 @@ private void createMaterializedTableInContinuousMode(
}
private void createMaterializedTableInFullMode(
- OperationExecutor operationExecutor,
- OperationHandle handle,
- CreateMaterializedTableOperation createMaterializedTableOperation) {
- if (workflowScheduler == null) {
- throw new SqlExecutionException(
- "The workflow scheduler must be configured when creating materialized table in full refresh mode.");
- }
+ CreateMaterializedTableOperation createMaterializedTableOperation,
+ RefreshWorkflowContext refreshWorkflowContext) {
// create materialized table first
- operationExecutor.callExecutableOperation(handle, createMaterializedTableOperation);
+ createMaterializedTableOperation.execute(operationCtx);
ObjectIdentifier materializedTableIdentifier =
createMaterializedTableOperation.getTableIdentifier();
@@ -267,16 +222,20 @@ private void createMaterializedTableInFullMode(
try {
createPeriodicRefreshWorkflow(
- operationExecutor,
- handle,
- materializedTableIdentifier,
- catalogMaterializedTable);
+ materializedTableIdentifier, catalogMaterializedTable, refreshWorkflowContext);
} catch (Exception e) {
// drop materialized table if creating the refresh workflow encounters an exception, so
// weak atomicity is guaranteed
- operationExecutor.callExecutableOperation(
- handle, new DropMaterializedTableOperation(materializedTableIdentifier, true));
- throw new SqlExecutionException(
+ cleanUpAfterFailure(
+ () ->
+ new DropMaterializedTableOperation(materializedTableIdentifier, true)
+ .execute(operationCtx),
+ e,
+ String.format(
+ "Failed to create refresh workflow for materialized table %s, and dropping the table failed as well."
+ + LEFT_IN_INITIALIZING_STATUS,
+ materializedTableIdentifier));
+ throw new TableException(
String.format(
"Failed to create refresh workflow for materialized table %s.",
materializedTableIdentifier),
@@ -284,29 +243,28 @@ private void createMaterializedTableInFullMode(
}
}
- private ResultFetcher callConvertTableToMaterializedTableOperation(
- OperationExecutor operationExecutor,
- OperationHandle handle,
+ private TableResultInternal callConvertTableToMaterializedTableOperation(
ConvertTableToMaterializedTableOperation convertOperation) {
ResolvedCatalogMaterializedTable materializedTable =
convertOperation.getMaterializedTable();
if (RefreshMode.CONTINUOUS == materializedTable.getRefreshMode()) {
- convertTableToMaterializedTableInContinuousMode(
- operationExecutor, handle, convertOperation);
+ convertTableToMaterializedTableInContinuousMode(convertOperation);
} else {
- convertTableToMaterializedTableInFullMode(operationExecutor, handle, convertOperation);
+ convertTableToMaterializedTableInFullMode(
+ convertOperation,
+ requireRefreshWorkflowContext(
+ "The workflow scheduler must be configured when converting a table to a materialized table in full refresh mode."));
}
// Just return ok for unify different refresh job info of continuous and full mode, user
// should get the refresh job info via desc table.
- return ResultFetcher.fromTableResult(handle, TABLE_RESULT_OK, false);
+ return TABLE_RESULT_OK;
}
private void convertTableToMaterializedTableInContinuousMode(
- OperationExecutor operationExecutor,
- OperationHandle handle,
ConvertTableToMaterializedTableOperation convertOperation) {
+ checkSubmissionTargetBeforeChanges();
// swap the catalog entry from a regular table to a materialized table first
- operationExecutor.callExecutableOperation(handle, convertOperation);
+ convertOperation.execute(operationCtx);
ObjectIdentifier materializedTableIdentifier = convertOperation.getTableIdentifier();
ResolvedCatalogMaterializedTable catalogMaterializedTable =
@@ -314,19 +272,13 @@ private void convertTableToMaterializedTableInContinuousMode(
try {
executeContinuousRefreshJob(
- operationExecutor,
- handle,
catalogMaterializedTable,
materializedTableIdentifier,
Map.of(),
Optional.empty());
} catch (Exception e) {
- suspendMaterializedTable(
- operationExecutor,
- handle,
- materializedTableIdentifier,
- catalogMaterializedTable);
- throw new SqlExecutionException(
+ suspendMaterializedTable(materializedTableIdentifier, catalogMaterializedTable);
+ throw new TableException(
String.format(
"Failed to start the continuous refresh job when converting table %s to a materialized table. "
+ "The table was converted and left in SUSPENDED status; resume it once the issue is resolved.",
@@ -336,15 +288,10 @@ private void convertTableToMaterializedTableInContinuousMode(
}
private void convertTableToMaterializedTableInFullMode(
- OperationExecutor operationExecutor,
- OperationHandle handle,
- ConvertTableToMaterializedTableOperation convertOperation) {
- if (workflowScheduler == null) {
- throw new SqlExecutionException(
- "The workflow scheduler must be configured when converting a table to a materialized table in full refresh mode.");
- }
+ ConvertTableToMaterializedTableOperation convertOperation,
+ RefreshWorkflowContext refreshWorkflowContext) {
// swap the catalog entry from a regular table to a materialized table first
- operationExecutor.callExecutableOperation(handle, convertOperation);
+ convertOperation.execute(operationCtx);
ObjectIdentifier materializedTableIdentifier = convertOperation.getTableIdentifier();
ResolvedCatalogMaterializedTable catalogMaterializedTable =
@@ -352,17 +299,10 @@ private void convertTableToMaterializedTableInFullMode(
try {
createPeriodicRefreshWorkflow(
- operationExecutor,
- handle,
- materializedTableIdentifier,
- catalogMaterializedTable);
+ materializedTableIdentifier, catalogMaterializedTable, refreshWorkflowContext);
} catch (Exception e) {
- suspendMaterializedTable(
- operationExecutor,
- handle,
- materializedTableIdentifier,
- catalogMaterializedTable);
- throw new SqlExecutionException(
+ suspendMaterializedTable(materializedTableIdentifier, catalogMaterializedTable);
+ throw new TableException(
String.format(
"Failed to create the refresh workflow when converting table %s to a materialized table. "
+ "The table was converted and left in SUSPENDED status; resume it once the issue is resolved.",
@@ -376,11 +316,12 @@ private void convertTableToMaterializedTableInFullMode(
* resulting handler with {@code ACTIVATED} status.
*/
private void createPeriodicRefreshWorkflow(
- OperationExecutor operationExecutor,
- OperationHandle handle,
ObjectIdentifier materializedTableIdentifier,
- ResolvedCatalogMaterializedTable catalogMaterializedTable)
+ ResolvedCatalogMaterializedTable catalogMaterializedTable,
+ RefreshWorkflowContext refreshWorkflowContext)
throws Exception {
+ final WorkflowScheduler extends RefreshHandler> workflowScheduler =
+ refreshWorkflowContext.scheduler();
final IntervalFreshness freshness = catalogMaterializedTable.getDefinitionFreshness();
final String cronExpression = convertFreshnessToCron(freshness);
final CreateRefreshWorkflow createRefreshWorkflow =
@@ -388,9 +329,9 @@ private void createPeriodicRefreshWorkflow(
materializedTableIdentifier,
catalogMaterializedTable.getExpandedQuery(),
cronExpression,
- getSessionInitializationConf(operationExecutor),
+ refreshWorkflowContext.sessionInitializationConf(),
Map.of(),
- restEndpointUrl);
+ refreshWorkflowContext.restEndpointUrl());
final RefreshHandler refreshHandler =
workflowScheduler.createRefreshWorkflow(createRefreshWorkflow);
@@ -399,8 +340,6 @@ private void createPeriodicRefreshWorkflow(
final byte[] serializedRefreshHandler = refreshHandlerSerializer.serialize(refreshHandler);
updateRefreshHandler(
- operationExecutor,
- handle,
materializedTableIdentifier,
catalogMaterializedTable,
RefreshStatus.ACTIVATED,
@@ -410,8 +349,6 @@ private void createPeriodicRefreshWorkflow(
/** Sets a materialized table refresh status to {@code SUSPENDED}. */
private void suspendMaterializedTable(
- OperationExecutor operationExecutor,
- OperationHandle handle,
ObjectIdentifier materializedTableIdentifier,
ResolvedCatalogMaterializedTable catalogMaterializedTable) {
AlterMaterializedTableChangeOperation alterOperation =
@@ -420,53 +357,60 @@ private void suspendMaterializedTable(
oldTable ->
List.of(TableChange.modifyRefreshStatus(RefreshStatus.SUSPENDED)),
catalogMaterializedTable);
- operationExecutor.callExecutableOperation(handle, alterOperation);
+ alterOperation.execute(operationCtx);
}
- private ResultFetcher callAlterMaterializedTableSuspend(
- OperationExecutor operationExecutor,
- OperationHandle handle,
+ private TableResultInternal callAlterMaterializedTableSuspend(
AlterMaterializedTableSuspendOperation op) {
ObjectIdentifier tableIdentifier = op.getTableIdentifier();
ResolvedCatalogMaterializedTable materializedTable =
- getCatalogMaterializedTable(operationExecutor, tableIdentifier);
+ getCatalogMaterializedTable(tableIdentifier);
- // Initialization phase doesn't support resume operation.
+ // Initialization phase doesn't support suspend operation.
if (RefreshStatus.INITIALIZING == materializedTable.getRefreshStatus()) {
- throw new SqlExecutionException(
+ throw new TableException(
String.format(
"Materialized table %s is being initialized and does not support suspend operation.",
tableIdentifier));
}
if (RefreshMode.CONTINUOUS == materializedTable.getRefreshMode()) {
- suspendContinuousRefreshJob(
- operationExecutor, handle, tableIdentifier, materializedTable);
+ suspendContinuousRefreshJob(tableIdentifier, materializedTable);
} else {
- suspendRefreshWorkflow(operationExecutor, handle, tableIdentifier, materializedTable);
+ suspendRefreshWorkflow(tableIdentifier, materializedTable);
}
- return ResultFetcher.fromTableResult(handle, TABLE_RESULT_OK, false);
+ return TABLE_RESULT_OK;
}
private ResolvedCatalogMaterializedTable suspendContinuousRefreshJob(
- OperationExecutor operationExecutor,
- OperationHandle handle,
- ObjectIdentifier tableIdentifier,
- ResolvedCatalogMaterializedTable materializedTable) {
+ ObjectIdentifier tableIdentifier, ResolvedCatalogMaterializedTable materializedTable) {
+ final ContinuousRefreshHandler refreshHandler;
+ final String savepointPath;
try {
- ContinuousRefreshHandler refreshHandler =
+ refreshHandler =
deserializeContinuousHandler(materializedTable.getSerializedRefreshHandler());
if (RefreshStatus.SUSPENDED == materializedTable.getRefreshStatus()) {
- throw new SqlExecutionException(
+ throw new TableException(
String.format(
"Materialized table %s continuous refresh job has been suspended, jobId is %s.",
tableIdentifier, refreshHandler.getJobId()));
}
- String savepointPath = stopJobWithSavepoint(operationExecutor, handle, refreshHandler);
+ final String savepointDirectory = requireSavepointDirectory();
+ savepointPath =
+ refreshJobController.stopWithSavepoint(refreshHandler, savepointDirectory);
+ } catch (Exception e) {
+ throw new TableException(
+ String.format(
+ "Failed to suspend the continuous refresh job for materialized table %s.",
+ tableIdentifier),
+ e);
+ }
- ContinuousRefreshHandler updateRefreshHandler =
+ // The job has stopped, so from here on a failure must name the savepoint.
+ try {
+ final ContinuousRefreshHandler updateRefreshHandler =
new ContinuousRefreshHandler(
refreshHandler.getExecutionTarget(),
refreshHandler.getClusterId(),
@@ -474,59 +418,60 @@ private ResolvedCatalogMaterializedTable suspendContinuousRefreshJob(
savepointPath);
return updateRefreshHandler(
- operationExecutor,
- handle,
tableIdentifier,
materializedTable,
RefreshStatus.SUSPENDED,
updateRefreshHandler.asSummaryString(),
serializeContinuousHandler(updateRefreshHandler));
} catch (Exception e) {
- throw new SqlExecutionException(
+ final String jobId = refreshHandler.getJobId();
+ LOG.error(
+ "Stopped the continuous refresh job {} of materialized table {} with savepoint {}, but failed to record it in the catalog.",
+ jobId,
+ tableIdentifier,
+ savepointPath,
+ e);
+ throw new TableException(
String.format(
- "Failed to suspend the continuous refresh job for materialized table %s.",
- tableIdentifier),
+ "Stopped the continuous refresh job %s of materialized table %s with savepoint %s, but failed to record the savepoint in the catalog, which still reports the table as ACTIVATED. To resume from the savepoint, set 'execution.state-recovery.path' to %s before RESUME, and unset it afterwards (RESET in SQL); otherwise RESUME starts the refresh job without restoring state.",
+ jobId, tableIdentifier, savepointPath, savepointPath),
e);
}
}
private void suspendRefreshWorkflow(
- OperationExecutor operationExecutor,
- OperationHandle handle,
- ObjectIdentifier tableIdentifier,
- ResolvedCatalogMaterializedTable materializedTable) {
+ ObjectIdentifier tableIdentifier, ResolvedCatalogMaterializedTable materializedTable) {
if (RefreshStatus.SUSPENDED == materializedTable.getRefreshStatus()) {
- throw new SqlExecutionException(
+ throw new TableException(
String.format(
"Materialized table %s refresh workflow has been suspended.",
tableIdentifier));
}
- if (workflowScheduler == null) {
- throw new SqlExecutionException(
- "The workflow scheduler must be configured when suspending materialized table in full refresh mode.");
- }
+ WorkflowScheduler extends RefreshHandler> workflowScheduler =
+ requireRefreshWorkflowContext(
+ "The workflow scheduler must be configured when suspending materialized table in full refresh mode.")
+ .scheduler();
try {
RefreshHandlerSerializer> refreshHandlerSerializer =
workflowScheduler.getRefreshHandlerSerializer();
RefreshHandler refreshHandler =
refreshHandlerSerializer.deserialize(
- materializedTable.getSerializedRefreshHandler(), userCodeClassLoader);
+ materializedTable.getSerializedRefreshHandler(),
+ operationCtx.getResourceManager().getUserClassLoader());
ModifyRefreshWorkflow modifyRefreshWorkflow =
new SuspendRefreshWorkflow(refreshHandler);
workflowScheduler.modifyRefreshWorkflow(modifyRefreshWorkflow);
updateRefreshHandler(
- operationExecutor,
- handle,
tableIdentifier,
materializedTable,
RefreshStatus.SUSPENDED,
refreshHandler.asSummaryString(),
materializedTable.getSerializedRefreshHandler());
} catch (Exception e) {
- throw new SqlExecutionException(
+ throw new TableException(
String.format(
"Failed to suspend the refresh workflow for materialized table %s.",
tableIdentifier),
@@ -534,17 +479,15 @@ private void suspendRefreshWorkflow(
}
}
- private ResultFetcher callAlterMaterializedTableResume(
- OperationExecutor operationExecutor,
- OperationHandle handle,
+ private TableResultInternal callAlterMaterializedTableResume(
AlterMaterializedTableResumeOperation op) {
ObjectIdentifier tableIdentifier = op.getTableIdentifier();
ResolvedCatalogMaterializedTable catalogMaterializedTable =
- getCatalogMaterializedTable(operationExecutor, tableIdentifier);
+ getCatalogMaterializedTable(tableIdentifier);
// Initialization phase doesn't support resume operation.
if (RefreshStatus.INITIALIZING == catalogMaterializedTable.getRefreshStatus()) {
- throw new SqlExecutionException(
+ throw new TableException(
String.format(
"Materialized table %s is being initialized and does not support resume operation.",
tableIdentifier));
@@ -552,55 +495,50 @@ private ResultFetcher callAlterMaterializedTableResume(
if (RefreshMode.CONTINUOUS == catalogMaterializedTable.getRefreshMode()) {
resumeContinuousRefreshJob(
- operationExecutor,
- handle,
- tableIdentifier,
- catalogMaterializedTable,
- op.getDynamicOptions());
+ tableIdentifier, catalogMaterializedTable, op.getDynamicOptions());
} else {
resumeRefreshWorkflow(
- operationExecutor,
- handle,
- tableIdentifier,
- catalogMaterializedTable,
- op.getDynamicOptions());
+ tableIdentifier, catalogMaterializedTable, op.getDynamicOptions());
}
- return ResultFetcher.fromTableResult(handle, TABLE_RESULT_OK, false);
+ return TABLE_RESULT_OK;
}
private void resumeContinuousRefreshJob(
- OperationExecutor operationExecutor,
- OperationHandle handle,
ObjectIdentifier tableIdentifier,
ResolvedCatalogMaterializedTable catalogMaterializedTable,
Map dynamicOptions) {
- ContinuousRefreshHandler refreshHandler =
+ final ContinuousRefreshHandler refreshHandler =
deserializeContinuousHandler(
catalogMaterializedTable.getSerializedRefreshHandler());
// Repeated resume continuous refresh job is not supported
if (RefreshStatus.ACTIVATED == catalogMaterializedTable.getRefreshStatus()) {
- JobStatus jobStatus = getJobStatus(operationExecutor, handle, refreshHandler);
+ final JobStatus jobStatus = refreshJobController.getJobStatus(refreshHandler);
if (!jobStatus.isGloballyTerminalState()) {
- throw new SqlExecutionException(
+ throw new TableException(
String.format(
"Materialized table %s continuous refresh job has been resumed, jobId is %s.",
tableIdentifier, refreshHandler.getJobId()));
}
+ // an ACTIVATED handler never carries a savepoint, so only a configured path keeps state
+ if (operationCtx.getTableConfig().getOptional(SAVEPOINT_PATH).isEmpty()) {
+ LOG.warn(
+ "Materialized table {} is ACTIVATED, but its continuous refresh job {} has already reached {}. Starting a new refresh job without restoring state; to restore a savepoint instead, set 'execution.state-recovery.path' to it before RESUME, and unset it afterwards (RESET in SQL).",
+ tableIdentifier,
+ refreshHandler.getJobId(),
+ jobStatus);
+ }
}
- Optional restorePath = refreshHandler.getRestorePath();
+ checkRestartTargetBeforeChanges(
+ tableIdentifier, catalogMaterializedTable.getSerializedRefreshHandler());
+ final Optional restorePath = refreshHandler.getRestorePath();
try {
executeContinuousRefreshJob(
- operationExecutor,
- handle,
- catalogMaterializedTable,
- tableIdentifier,
- dynamicOptions,
- restorePath);
+ catalogMaterializedTable, tableIdentifier, dynamicOptions, restorePath);
} catch (Exception e) {
- throw new SqlExecutionException(
+ throw new TableException(
String.format(
"Failed to resume the continuous refresh job for materialized table %s.",
tableIdentifier),
@@ -609,44 +547,40 @@ private void resumeContinuousRefreshJob(
}
private void resumeRefreshWorkflow(
- OperationExecutor operationExecutor,
- OperationHandle handle,
ObjectIdentifier tableIdentifier,
ResolvedCatalogMaterializedTable catalogMaterializedTable,
Map dynamicOptions) {
// Repeated resume refresh workflow is not supported
if (RefreshStatus.ACTIVATED == catalogMaterializedTable.getRefreshStatus()) {
- throw new SqlExecutionException(
+ throw new TableException(
String.format(
"Materialized table %s refresh workflow has been resumed.",
tableIdentifier));
}
- if (workflowScheduler == null) {
- throw new SqlExecutionException(
- "The workflow scheduler must be configured when resuming materialized table in full refresh mode.");
- }
+ WorkflowScheduler extends RefreshHandler> workflowScheduler =
+ requireRefreshWorkflowContext(
+ "The workflow scheduler must be configured when resuming materialized table in full refresh mode.")
+ .scheduler();
try {
RefreshHandlerSerializer> refreshHandlerSerializer =
workflowScheduler.getRefreshHandlerSerializer();
RefreshHandler refreshHandler =
refreshHandlerSerializer.deserialize(
catalogMaterializedTable.getSerializedRefreshHandler(),
- userCodeClassLoader);
+ operationCtx.getResourceManager().getUserClassLoader());
ModifyRefreshWorkflow modifyRefreshWorkflow =
new ResumeRefreshWorkflow(refreshHandler, dynamicOptions);
workflowScheduler.modifyRefreshWorkflow(modifyRefreshWorkflow);
updateRefreshHandler(
- operationExecutor,
- handle,
tableIdentifier,
catalogMaterializedTable,
RefreshStatus.ACTIVATED,
refreshHandler.asSummaryString(),
catalogMaterializedTable.getSerializedRefreshHandler());
} catch (Exception e) {
- throw new SqlExecutionException(
+ throw new TableException(
String.format(
"Failed to resume the refresh workflow for materialized table %s.",
tableIdentifier),
@@ -655,8 +589,38 @@ private void resumeRefreshWorkflow(
}
private void executeContinuousRefreshJob(
- OperationExecutor operationExecutor,
- OperationHandle handle,
+ CatalogMaterializedTable catalogMaterializedTable,
+ ObjectIdentifier materializedTableIdentifier,
+ Map dynamicOptions,
+ Optional restorePath) {
+ recordContinuousRefreshJob(
+ materializedTableIdentifier,
+ catalogMaterializedTable,
+ submitContinuousRefreshJob(
+ catalogMaterializedTable,
+ materializedTableIdentifier,
+ dynamicOptions,
+ restorePath));
+ }
+
+ private void executeContinuousRefreshJobOrCancelIt(
+ CatalogMaterializedTable catalogMaterializedTable,
+ ObjectIdentifier materializedTableIdentifier) {
+ final ContinuousRefreshHandler refreshHandler =
+ submitContinuousRefreshJob(
+ catalogMaterializedTable,
+ materializedTableIdentifier,
+ Map.of(),
+ Optional.empty());
+ try {
+ recordContinuousRefreshJob(
+ materializedTableIdentifier, catalogMaterializedTable, refreshHandler);
+ } catch (Exception e) {
+ throw cancelUnrecordedRefreshJob(materializedTableIdentifier, refreshHandler, e);
+ }
+ }
+
+ private ContinuousRefreshHandler submitContinuousRefreshJob(
CatalogMaterializedTable catalogMaterializedTable,
ObjectIdentifier materializedTableIdentifier,
Map dynamicOptions,
@@ -673,10 +637,10 @@ private void executeContinuousRefreshJob(
restorePath.ifPresent(s -> customConfig.set(SAVEPOINT_PATH, s));
// Do not override the user-defined checkpoint interval
- if (!operationExecutor
- .getSessionContext()
- .getSessionConf()
- .contains(CheckpointingOptions.CHECKPOINTING_INTERVAL)) {
+ if (operationCtx
+ .getTableConfig()
+ .getOptional(CheckpointingOptions.CHECKPOINTING_INTERVAL)
+ .isEmpty()) {
final Duration freshness =
validateAndGetIntervalFreshness(catalogMaterializedTable).toDuration();
@@ -689,67 +653,83 @@ private void executeContinuousRefreshJob(
catalogMaterializedTable.getExpandedQuery(),
dynamicOptions);
- JobExecutionResult result =
- executeRefreshJob(insertStatement, customConfig, operationExecutor, handle);
- ContinuousRefreshHandler continuousRefreshHandler =
- new ContinuousRefreshHandler(
- result.executionTarget, result.clusterId, result.jobId);
- byte[] serializedBytes = serializeContinuousHandler(continuousRefreshHandler);
+ final RefreshJobResult result =
+ jobSubmitter.submitRefreshJob(customConfig, insertStatement);
+ refreshJobController.register(result);
+ return new ContinuousRefreshHandler(
+ result.getTarget().getExecutionTarget(),
+ result.getTarget().getClusterId(),
+ result.getJobId());
+ }
+ private void recordContinuousRefreshJob(
+ ObjectIdentifier materializedTableIdentifier,
+ CatalogMaterializedTable catalogMaterializedTable,
+ ContinuousRefreshHandler refreshHandler) {
updateRefreshHandler(
- operationExecutor,
- handle,
materializedTableIdentifier,
- resolveCatalogMaterializedTable(operationExecutor, catalogMaterializedTable),
+ resolveCatalogMaterializedTable(catalogMaterializedTable),
RefreshStatus.ACTIVATED,
- continuousRefreshHandler.asSummaryString(),
- serializedBytes);
+ refreshHandler.asSummaryString(),
+ serializeContinuousHandler(refreshHandler));
}
- private ResultFetcher callAlterMaterializedTableRefreshOperation(
- OperationExecutor operationExecutor,
- OperationHandle handle,
- AlterMaterializedTableRefreshOperation alterMaterializedTableRefreshOperation) {
- ObjectIdentifier materializedTableIdentifier =
- alterMaterializedTableRefreshOperation.getTableIdentifier();
-
- Map partitionSpec =
- alterMaterializedTableRefreshOperation.getPartitionSpec();
-
- return refreshMaterializedTable(
- operationExecutor,
- handle,
- materializedTableIdentifier,
- partitionSpec,
- Map.of(),
- false,
- null);
+ private TableException cancelUnrecordedRefreshJob(
+ ObjectIdentifier tableIdentifier,
+ ContinuousRefreshHandler refreshHandler,
+ Exception recordFailure) {
+ final String jobId = refreshHandler.getJobId();
+ LOG.error(
+ "Failed to record the continuous refresh job {} of materialized table {} in the catalog; cancelling it.",
+ jobId,
+ tableIdentifier,
+ recordFailure);
+ try {
+ refreshJobController.cancel(refreshHandler);
+ } catch (Exception cancelFailure) {
+ LOG.error(
+ "Failed to cancel the continuous refresh job {} of materialized table {} (execution.target={}, cluster id={}), which the catalog does not record. It may still be running and must be cancelled manually.",
+ jobId,
+ tableIdentifier,
+ refreshHandler.getExecutionTarget(),
+ refreshHandler.getClusterId(),
+ cancelFailure);
+ return withSuppressed(
+ String.format(
+ "Failed to record the continuous refresh job %s of materialized table %s in the catalog, and cancelling it failed as well. The job may still be running (execution.target=%s, cluster id=%s) and must be cancelled manually.",
+ jobId,
+ tableIdentifier,
+ refreshHandler.getExecutionTarget(),
+ refreshHandler.getClusterId()),
+ cancelFailure,
+ recordFailure);
+ }
+ return new TableException(
+ String.format(
+ "Failed to record the continuous refresh job %s of materialized table %s in the catalog. The job was cancelled.",
+ jobId, tableIdentifier),
+ recordFailure);
}
- public ResultFetcher refreshMaterializedTable(
- OperationExecutor operationExecutor,
- OperationHandle handle,
- ObjectIdentifier materializedTableIdentifier,
- Map staticPartitions,
- Map dynamicOptions,
- boolean isPeriodic,
- @Nullable String scheduleTime) {
+ private TableResultInternal callAlterMaterializedTableRefreshOperation(
+ AlterMaterializedTableRefreshOperation op) {
+ ObjectIdentifier materializedTableIdentifier = op.getTableIdentifier();
ResolvedCatalogMaterializedTable materializedTable =
- getCatalogMaterializedTable(operationExecutor, materializedTableIdentifier);
+ getCatalogMaterializedTable(materializedTableIdentifier);
+
+ boolean isPeriodic = op.isPeriodic();
Map refreshPartitions =
isPeriodic
? getPeriodRefreshPartition(
- scheduleTime,
+ op.getScheduleTime(),
materializedTable.getDefinitionFreshness(),
materializedTableIdentifier,
materializedTable.getOptions(),
- operationExecutor
- .getTableEnvironment()
- .getConfig()
- .getLocalTimeZone())
- : staticPartitions;
+ operationCtx.getTableConfig().getLocalTimeZone())
+ : op.getPartitionSpec();
validatePartitionSpec(refreshPartitions, materializedTable);
+ checkSubmissionTargetBeforeChanges();
// Set job name, runtime mode
Configuration customConfig = new Configuration();
@@ -770,40 +750,34 @@ public ResultFetcher refreshMaterializedTable(
materializedTableIdentifier,
materializedTable.getExpandedQuery(),
refreshPartitions,
- dynamicOptions);
+ op.getDynamicOptions());
try {
LOG.info(
"Starting refresh of the materialized table {}, statement: {}",
materializedTableIdentifier,
insertStatement);
- JobExecutionResult result =
- executeRefreshJob(insertStatement, customConfig, operationExecutor, handle);
-
- Map clusterInfo = new HashMap<>();
- clusterInfo.put(
- StringData.fromString(TARGET.key()),
- StringData.fromString(result.executionTarget));
- Optional clusterIdKeyName = getClusterIdKeyName(result.executionTarget);
- clusterIdKeyName.ifPresent(
- s ->
- clusterInfo.put(
- StringData.fromString(s),
- StringData.fromString(result.clusterId)));
-
- return ResultFetcher.fromResults(
- handle,
+ final RefreshJobResult result =
+ jobSubmitter.submitRefreshJob(customConfig, insertStatement);
+
+ final Row row = Row.of(result.getJobId(), result.getTarget().toClusterInfo());
+ ResolvedSchema schema =
ResolvedSchema.of(
- Column.physical(JOB_ID, DataTypes.STRING()),
+ Column.physical("job id", DataTypes.STRING()),
Column.physical(
- CLUSTER_INFO,
- DataTypes.MAP(DataTypes.STRING(), DataTypes.STRING()))),
- Collections.singletonList(
- GenericRowData.of(
- StringData.fromString(result.jobId),
- new GenericMapData(clusterInfo))));
+ "cluster info",
+ DataTypes.MAP(DataTypes.STRING(), DataTypes.STRING())));
+
+ return TableResultImpl.builder()
+ .resultKind(ResultKind.SUCCESS_WITH_CONTENT)
+ .schema(schema)
+ .resultProvider(
+ new StaticResultProvider(
+ List.of(row),
+ DefaultMaterializedTableExecutor::refreshRowToInternalRow))
+ .build();
} catch (Exception e) {
- throw new SqlExecutionException(
+ throw new TableException(
String.format(
"Failed to refresh the materialized table %s.",
materializedTableIdentifier),
@@ -845,7 +819,7 @@ static Map getPeriodRefreshPartition(
TimeZone.getTimeZone(localZoneId),
-freshness.toDuration().toMillis());
if (partFiledValue == null) {
- throw new SqlExecutionException(
+ throw new TableException(
String.format(
"Failed to parse a valid partition value for the field '%s' in materialized table %s using the scheduler time '%s' based on the date format '%s'.",
partField,
@@ -902,7 +876,7 @@ private void validatePartitionSpec(
}
@VisibleForTesting
- protected static String getRefreshStatement(
+ static String getRefreshStatement(
ObjectIdentifier tableIdentifier,
String definitionQuery,
Map partitionSpec,
@@ -929,103 +903,127 @@ protected static String getRefreshStatement(
return insertStatement.toString();
}
- private ResultFetcher callAlterMaterializedTableChangeOperation(
- OperationExecutor operationExecutor,
- OperationHandle handle,
+ private TableResultInternal callAlterMaterializedTableChangeOperation(
AlterMaterializedTableChangeOperation op) {
ObjectIdentifier tableIdentifier = op.getTableIdentifier();
ResolvedCatalogMaterializedTable oldMaterializedTable =
- getCatalogMaterializedTable(operationExecutor, tableIdentifier);
+ getCatalogMaterializedTable(tableIdentifier);
if (RefreshMode.FULL == oldMaterializedTable.getRefreshMode()) {
// directly apply the alter operation
- return operationExecutor.callExecutableOperation(
- handle, op.copyAsTableChangeOperation());
+ return op.copyAsTableChangeOperation().execute(operationCtx);
}
if (RefreshStatus.ACTIVATED == oldMaterializedTable.getRefreshStatus()) {
- // 1. suspend the materialized table
- ResolvedCatalogMaterializedTable suspendMaterializedTable =
- suspendContinuousRefreshJob(
- operationExecutor, handle, tableIdentifier, oldMaterializedTable);
-
- // 2. alter materialized table schema & query definition
- AlterMaterializedTableChangeOperation alterMaterializedTableChangeOperation =
- new AlterMaterializedTableChangeOperation(
- op.getTableIdentifier(),
- oldTable -> op.getTableChanges(),
- suspendMaterializedTable,
- op.getAsQueryOperation());
- operationExecutor.callExecutableOperation(
- handle, alterMaterializedTableChangeOperation);
-
- // 3. resume the materialized table
- try {
- executeContinuousRefreshJob(
- operationExecutor,
- handle,
- alterMaterializedTableChangeOperation.getNewTable(),
- tableIdentifier,
- Map.of(),
- Optional.empty());
- } catch (Exception e) {
- // Roll back the changes to the materialized table and restore the continuous
- // refresh job
- LOG.warn(
- "Failed to start the continuous refresh job for materialized table {} using new query {}, rollback to origin query {}.",
- tableIdentifier,
- op.getNewTable().getExpandedQuery(),
- suspendMaterializedTable.getExpandedQuery(),
- e);
+ alterActivatedContinuousMaterializedTable(op, oldMaterializedTable);
+ } else if (RefreshStatus.SUSPENDED == oldMaterializedTable.getRefreshStatus()) {
+ alterSuspendedContinuousMaterializedTable(op, oldMaterializedTable);
+ } else {
+ throw new TableException(
+ String.format(
+ "Materialized table %s is being initialized and does not support alter operation.",
+ tableIdentifier));
+ }
- AlterMaterializedTableChangeOperation rollbackChangeOperation =
- generateRollbackAlterMaterializedTableOperation(
- suspendMaterializedTable, alterMaterializedTableChangeOperation);
- operationExecutor.callExecutableOperation(handle, rollbackChangeOperation);
+ return TABLE_RESULT_OK;
+ }
- ContinuousRefreshHandler continuousRefreshHandler =
- deserializeContinuousHandler(
- suspendMaterializedTable.getSerializedRefreshHandler());
- executeContinuousRefreshJob(
- operationExecutor,
- handle,
+ /** Stops the refresh job, applies the change and restarts it, rolling back on failure. */
+ private void alterActivatedContinuousMaterializedTable(
+ AlterMaterializedTableChangeOperation op,
+ ResolvedCatalogMaterializedTable oldMaterializedTable) {
+ final ObjectIdentifier tableIdentifier = op.getTableIdentifier();
+ // check the target before the stop, so a restart that could never run keeps the job
+ // running and the table untouched
+ checkRestartTargetBeforeChanges(
+ tableIdentifier, oldMaterializedTable.getSerializedRefreshHandler());
+
+ // 1. suspend the materialized table
+ ResolvedCatalogMaterializedTable suspendMaterializedTable =
+ suspendContinuousRefreshJob(tableIdentifier, oldMaterializedTable);
+
+ // 2. alter materialized table schema & query definition
+ final AlterMaterializedTableChangeOperation alterMaterializedTableChangeOperation =
+ new AlterMaterializedTableChangeOperation(
+ op.getTableIdentifier(),
+ oldTable -> op.getTableChanges(),
suspendMaterializedTable,
- tableIdentifier,
- Map.of(),
- continuousRefreshHandler.getRestorePath());
+ op.getAsQueryOperation());
+ alterMaterializedTableChangeOperation.execute(operationCtx);
- throw new SqlExecutionException(
- String.format(
- "Failed to start the continuous refresh job using new query %s when altering materialized table %s select query.",
- op.getNewTable().getExpandedQuery(), tableIdentifier),
- e);
- }
- } else if (RefreshStatus.SUSPENDED == oldMaterializedTable.getRefreshStatus()) {
- // alter schema & definition query & refresh handler (reset savepoint path of refresh
- // handler)
- List tableChanges = new ArrayList<>(op.getTableChanges());
- TableChange.ModifyRefreshHandler modifyRefreshHandler =
- generateResetSavepointTableChange(
- oldMaterializedTable.getSerializedRefreshHandler());
- tableChanges.add(modifyRefreshHandler);
-
- AlterMaterializedTableChangeOperation alterMaterializedTableChangeOperation =
- new AlterMaterializedTableChangeOperation(
- tableIdentifier,
- oldTable -> tableChanges,
- oldMaterializedTable,
- op.getAsQueryOperation());
+ // 3. resume the materialized table
+ try {
+ executeContinuousRefreshJob(
+ alterMaterializedTableChangeOperation.getNewTable(),
+ tableIdentifier,
+ Map.of(),
+ Optional.empty());
+ } catch (Exception e) {
+ final ContinuousRefreshHandler suspendedRefreshHandler =
+ deserializeContinuousHandler(
+ suspendMaterializedTable.getSerializedRefreshHandler());
+ LOG.warn(
+ "Failed to start the continuous refresh job for materialized table {} using new query {}, rollback to origin query {} with savepoint {}.",
+ tableIdentifier,
+ op.getNewTable().getExpandedQuery(),
+ suspendMaterializedTable.getExpandedQuery(),
+ suspendedRefreshHandler.getRestorePath().orElse(null),
+ e);
- operationExecutor.callExecutableOperation(
- handle, alterMaterializedTableChangeOperation);
- } else {
- throw new SqlExecutionException(
+ // Roll back the changes to the materialized table and restore the continuous
+ // refresh job
+ generateRollbackAlterMaterializedTableOperation(
+ suspendMaterializedTable, alterMaterializedTableChangeOperation)
+ .execute(operationCtx);
+ executeContinuousRefreshJob(
+ suspendMaterializedTable,
+ tableIdentifier,
+ Map.of(),
+ suspendedRefreshHandler.getRestorePath());
+
+ throw new TableException(
String.format(
- "Materialized table %s is being initialized and does not support alter operation.",
- tableIdentifier));
+ "Failed to start the continuous refresh job using new query %s when altering materialized table %s select query.",
+ op.getNewTable().getExpandedQuery(), tableIdentifier),
+ e);
}
+ }
+
+ private void alterSuspendedContinuousMaterializedTable(
+ AlterMaterializedTableChangeOperation op,
+ ResolvedCatalogMaterializedTable oldMaterializedTable) {
+ // alter schema & definition query & refresh handler (reset savepoint path of refresh
+ // handler)
+ List tableChanges = new ArrayList<>(op.getTableChanges());
+ tableChanges.add(
+ generateResetSavepointTableChange(
+ deserializeContinuousHandler(
+ oldMaterializedTable.getSerializedRefreshHandler())));
- return ResultFetcher.fromTableResult(handle, TABLE_RESULT_OK, false);
+ AlterMaterializedTableChangeOperation alterMaterializedTableChangeOperation =
+ new AlterMaterializedTableChangeOperation(
+ op.getTableIdentifier(),
+ oldTable -> tableChanges,
+ oldMaterializedTable,
+ op.getAsQueryOperation());
+
+ alterMaterializedTableChangeOperation.execute(operationCtx);
+ }
+
+ private static TableException withSuppressed(
+ String message, Exception cause, Exception suppressed) {
+ final TableException failure = new TableException(message, cause);
+ failure.addSuppressed(suppressed);
+ return failure;
+ }
+
+ private static void cleanUpAfterFailure(
+ Runnable cleanup, Exception failure, String cleanupFailureMessage) {
+ try {
+ cleanup.run();
+ } catch (Exception cleanupFailure) {
+ throw withSuppressed(cleanupFailureMessage, cleanupFailure, failure);
+ }
}
private AlterMaterializedTableChangeOperation generateRollbackAlterMaterializedTableOperation(
@@ -1040,9 +1038,7 @@ private AlterMaterializedTableChangeOperation generateRollbackAlterMaterializedT
}
private TableChange.ModifyRefreshHandler generateResetSavepointTableChange(
- byte[] serializedContinuousHandler) {
- ContinuousRefreshHandler continuousRefreshHandler =
- deserializeContinuousHandler(serializedContinuousHandler);
+ ContinuousRefreshHandler continuousRefreshHandler) {
ContinuousRefreshHandler resetContinuousRefreshHandler =
new ContinuousRefreshHandler(
continuousRefreshHandler.getExecutionTarget(),
@@ -1054,18 +1050,21 @@ private TableChange.ModifyRefreshHandler generateResetSavepointTableChange(
serializeContinuousHandler(resetContinuousRefreshHandler));
}
- private ResultFetcher callDropMaterializedTableOperation(
- OperationExecutor operationExecutor,
- OperationHandle handle,
+ private TableResultInternal callDropMaterializedTableOperation(
DropMaterializedTableOperation dropMaterializedTableOperation) {
ObjectIdentifier tableIdentifier = dropMaterializedTableOperation.getTableIdentifier();
- boolean tableExists = operationExecutor.tableExists(tableIdentifier);
+ boolean tableExists =
+ operationCtx
+ .getCatalogManager()
+ .getCatalog(tableIdentifier.getCatalogName())
+ .map(catalog -> catalog.tableExists(tableIdentifier.toObjectPath()))
+ .orElse(false);
if (!tableExists) {
if (dropMaterializedTableOperation.isIfExists()) {
LOG.info(
"Materialized table {} does not exists, skip the drop operation.",
tableIdentifier);
- return ResultFetcher.fromTableResult(handle, TABLE_RESULT_OK, false);
+ return TABLE_RESULT_OK;
} else {
throw new ValidationException(
String.format(
@@ -1075,7 +1074,7 @@ private ResultFetcher callDropMaterializedTableOperation(
}
ResolvedCatalogMaterializedTable materializedTable =
- getCatalogMaterializedTable(operationExecutor, tableIdentifier);
+ getCatalogMaterializedTable(tableIdentifier);
RefreshStatus refreshStatus = materializedTable.getRefreshStatus();
if (RefreshStatus.ACTIVATED == refreshStatus || RefreshStatus.SUSPENDED == refreshStatus) {
RefreshMode refreshMode = materializedTable.getRefreshMode();
@@ -1083,8 +1082,7 @@ private ResultFetcher callDropMaterializedTableOperation(
deleteRefreshWorkflow(tableIdentifier, materializedTable);
} else if (RefreshMode.CONTINUOUS == refreshMode
&& RefreshStatus.ACTIVATED == refreshStatus) {
- cancelContinuousRefreshJob(
- operationExecutor, handle, tableIdentifier, materializedTable);
+ cancelContinuousRefreshJob(tableIdentifier, materializedTable);
}
} else if (RefreshStatus.INITIALIZING == refreshStatus) {
throw new ValidationException(
@@ -1093,65 +1091,65 @@ private ResultFetcher callDropMaterializedTableOperation(
tableIdentifier.asSerializableString()));
}
- operationExecutor.callExecutableOperation(handle, dropMaterializedTableOperation);
+ dropMaterializedTableOperation.execute(operationCtx);
- return ResultFetcher.fromTableResult(handle, TABLE_RESULT_OK, false);
+ return TABLE_RESULT_OK;
}
private void cancelContinuousRefreshJob(
- OperationExecutor operationExecutor,
- OperationHandle handle,
- ObjectIdentifier tableIdentifier,
- CatalogMaterializedTable materializedTable) {
- ContinuousRefreshHandler refreshHandler =
+ ObjectIdentifier tableIdentifier, CatalogMaterializedTable materializedTable) {
+ final ContinuousRefreshHandler refreshHandler =
deserializeContinuousHandler(materializedTable.getSerializedRefreshHandler());
- // get job running status
- JobStatus jobStatus = getJobStatus(operationExecutor, handle, refreshHandler);
- if (!jobStatus.isTerminalState()) {
- try {
- cancelJob(operationExecutor, handle, refreshHandler);
- } catch (Exception e) {
- jobStatus = getJobStatus(operationExecutor, handle, refreshHandler);
- if (!jobStatus.isTerminalState()) {
- throw new SqlExecutionException(
- String.format(
- "Failed to drop the materialized table %s because the continuous refresh job %s could not be canceled."
- + " The current status of the continuous refresh job is %s.",
- tableIdentifier, refreshHandler.getJobId(), jobStatus),
- e);
- } else {
- LOG.warn(
- "An exception occurred while canceling the continuous refresh job {} for materialized table {},"
- + " but since the job is in a terminal state, skip the cancel operation.",
- refreshHandler.getJobId(),
- tableIdentifier);
- }
- }
- } else {
+ if (refreshJobController.getJobStatus(refreshHandler).isTerminalState()) {
LOG.info(
"No need to cancel the continuous refresh job {} for materialized table {} as it is not currently running.",
refreshHandler.getJobId(),
tableIdentifier);
+ return;
+ }
+
+ final Exception cancelFailure;
+ try {
+ refreshJobController.cancel(refreshHandler);
+ return;
+ } catch (Exception e) {
+ cancelFailure = e;
}
+
+ final JobStatus jobStatus = refreshJobController.getJobStatus(refreshHandler);
+ if (!jobStatus.isTerminalState()) {
+ throw new TableException(
+ String.format(
+ "Failed to drop the materialized table %s because the continuous refresh job %s could not be canceled."
+ + " The current status of the continuous refresh job is %s.",
+ tableIdentifier, refreshHandler.getJobId(), jobStatus),
+ cancelFailure);
+ }
+ LOG.warn(
+ "An exception occurred while canceling the continuous refresh job {} for materialized table {},"
+ + " but since the job is in a terminal state, skip the cancel operation.",
+ refreshHandler.getJobId(),
+ tableIdentifier,
+ cancelFailure);
}
private void deleteRefreshWorkflow(
ObjectIdentifier tableIdentifier, CatalogMaterializedTable catalogMaterializedTable) {
- if (workflowScheduler == null) {
- throw new SqlExecutionException(
- "The workflow scheduler must be configured when dropping materialized table in full refresh mode.");
- }
+ WorkflowScheduler extends RefreshHandler> workflowScheduler =
+ requireRefreshWorkflowContext(
+ "The workflow scheduler must be configured when dropping materialized table in full refresh mode.")
+ .scheduler();
try {
RefreshHandlerSerializer> refreshHandlerSerializer =
workflowScheduler.getRefreshHandlerSerializer();
RefreshHandler refreshHandler =
refreshHandlerSerializer.deserialize(
catalogMaterializedTable.getSerializedRefreshHandler(),
- userCodeClassLoader);
+ operationCtx.getResourceManager().getUserClassLoader());
DeleteRefreshWorkflow deleteRefreshWorkflow = new DeleteRefreshWorkflow(refreshHandler);
workflowScheduler.deleteRefreshWorkflow(deleteRefreshWorkflow);
} catch (Exception e) {
- throw new SqlExecutionException(
+ throw new TableException(
String.format(
"Failed to delete the refresh workflow for materialized table %s.",
tableIdentifier),
@@ -1159,106 +1157,13 @@ private void deleteRefreshWorkflow(
}
}
- /**
- * Retrieves the session configuration for initializing the periodic refresh job. The function
- * filters out default context configurations and removes unnecessary configurations such as
- * resources download directory and workflow scheduler related configurations.
- *
- * @param operationExecutor The OperationExecutor instance used to access the session context.
- * @return A Map containing the session configurations for initializing session for executing
- * the periodic refresh job.
- */
- private Map getSessionInitializationConf(OperationExecutor operationExecutor) {
- Map sessionConf =
- operationExecutor.getSessionContext().getSessionConf().toMap();
-
- // we only keep the session conf that is not in the default context or the conf value is
- // different from the default context.
- Map defaultContextConf =
- operationExecutor.getSessionContext().getDefaultContext().getFlinkConfig().toMap();
- sessionConf
- .entrySet()
- .removeIf(
- entry -> {
- String key = entry.getKey();
- String value = entry.getValue();
- return defaultContextConf.containsKey(key)
- && defaultContextConf.get(key).equals(value);
- });
-
- // remove useless conf
- sessionConf.remove(TableConfigOptions.RESOURCES_DOWNLOAD_DIR.key());
- sessionConf.keySet().removeIf(key -> key.startsWith(WORKFLOW_SCHEDULER_PREFIX));
-
- return sessionConf;
- }
-
- private static JobStatus getJobStatus(
- OperationExecutor operationExecutor,
- OperationHandle handle,
- ContinuousRefreshHandler refreshHandler) {
- ResultFetcher resultFetcher =
- operationExecutor.callDescribeJobOperation(
- getTableEnvironment(operationExecutor, refreshHandler),
- handle,
- new DescribeJobOperation(refreshHandler.getJobId()));
- List result = fetchAllResults(resultFetcher);
- String jobStatus = result.get(0).getString(2).toString();
- return JobStatus.valueOf(jobStatus);
- }
-
- private static void cancelJob(
- OperationExecutor operationExecutor,
- OperationHandle handle,
- ContinuousRefreshHandler refreshHandler) {
- operationExecutor.callStopJobOperation(
- getTableEnvironment(operationExecutor, refreshHandler),
- handle,
- new StopJobOperation(refreshHandler.getJobId(), false, false));
- }
-
- private static String stopJobWithSavepoint(
- OperationExecutor executor,
- OperationHandle handle,
- ContinuousRefreshHandler refreshHandler) {
- // check savepoint dir is configured
- Optional savepointDir =
- executor.getSessionContext().getSessionConf().getOptional(SAVEPOINT_DIRECTORY);
- if (savepointDir.isEmpty()) {
- throw new ValidationException(
- "Savepoint directory is not configured, can't stop job with savepoint.");
- }
- String jobId = refreshHandler.getJobId();
- ResultFetcher resultFetcher =
- executor.callStopJobOperation(
- getTableEnvironment(executor, refreshHandler),
- handle,
- new StopJobOperation(jobId, true, false));
- List results = fetchAllResults(resultFetcher);
- return results.get(0).getString(0).toString();
- }
-
- private static TableEnvironmentInternal getTableEnvironment(
- OperationExecutor executor, ContinuousRefreshHandler refreshHandler) {
-
- String target = refreshHandler.getExecutionTarget();
- Configuration sessionConfiguration = new Configuration();
- sessionConfiguration.set(TARGET, target);
- Optional clusterIdKeyName = getClusterIdKeyName(target);
- clusterIdKeyName.ifPresent(
- s -> sessionConfiguration.setString(s, refreshHandler.getClusterId()));
-
- return executor.getTableEnvironment(
- executor.getSessionContext().getSessionState().resourceManager,
- sessionConfiguration);
- }
-
private ContinuousRefreshHandler deserializeContinuousHandler(byte[] serializedRefreshHandler) {
try {
return ContinuousRefreshHandlerSerializer.INSTANCE.deserialize(
- serializedRefreshHandler, userCodeClassLoader);
+ serializedRefreshHandler,
+ operationCtx.getResourceManager().getUserClassLoader());
} catch (IOException | ClassNotFoundException e) {
- throw new SqlExecutionException("Failed to deserialize ContinuousRefreshHandler.", e);
+ throw new TableException("Failed to deserialize ContinuousRefreshHandler.", e);
}
}
@@ -1266,14 +1171,17 @@ private byte[] serializeContinuousHandler(ContinuousRefreshHandler refreshHandle
try {
return ContinuousRefreshHandlerSerializer.INSTANCE.serialize(refreshHandler);
} catch (IOException e) {
- throw new SqlExecutionException("Failed to serialize ContinuousRefreshHandler.", e);
+ throw new TableException("Failed to serialize ContinuousRefreshHandler.", e);
}
}
private ResolvedCatalogMaterializedTable getCatalogMaterializedTable(
- OperationExecutor operationExecutor, ObjectIdentifier tableIdentifier) {
+ ObjectIdentifier tableIdentifier) {
ResolvedCatalogBaseTable> resolvedCatalogBaseTable =
- operationExecutor.getTable(tableIdentifier);
+ operationCtx
+ .getCatalogManager()
+ .getTableOrError(tableIdentifier)
+ .getResolvedTable();
if (MATERIALIZED_TABLE != resolvedCatalogBaseTable.getTableKind()) {
throw new ValidationException(
String.format(
@@ -1285,17 +1193,11 @@ private ResolvedCatalogMaterializedTable getCatalogMaterializedTable(
}
private ResolvedCatalogMaterializedTable resolveCatalogMaterializedTable(
- OperationExecutor operationExecutor, CatalogMaterializedTable materializedTable) {
- return operationExecutor
- .getSessionContext()
- .getSessionState()
- .catalogManager
- .resolveCatalogMaterializedTable(materializedTable);
+ CatalogMaterializedTable materializedTable) {
+ return operationCtx.getCatalogManager().resolveCatalogMaterializedTable(materializedTable);
}
private ResolvedCatalogMaterializedTable updateRefreshHandler(
- OperationExecutor operationExecutor,
- OperationHandle operationHandle,
ObjectIdentifier materializedTableIdentifier,
ResolvedCatalogMaterializedTable catalogMaterializedTable,
RefreshStatus refreshStatus,
@@ -1311,16 +1213,14 @@ private ResolvedCatalogMaterializedTable updateRefreshHandler(
oldTable -> tableChanges,
catalogMaterializedTable);
// update RefreshHandler to Catalog
- operationExecutor.callExecutableOperation(
- operationHandle, alterMaterializedTableChangeOperation);
+ alterMaterializedTableChangeOperation.execute(operationCtx);
- return resolveCatalogMaterializedTable(
- operationExecutor, alterMaterializedTableChangeOperation.getNewTable());
+ return resolveCatalogMaterializedTable(alterMaterializedTableChangeOperation.getNewTable());
}
/** Generate insert statement for materialized table. */
@VisibleForTesting
- protected static String getInsertStatement(
+ static String getInsertStatement(
ObjectIdentifier materializedTableIdentifier,
String definitionQuery,
Map dynamicOptions) {
@@ -1346,123 +1246,52 @@ private static String generateTableWithDynamicOptions(
return builder.toString();
}
- private static List fetchAllResults(ResultFetcher resultFetcher) {
- Long token = 0L;
- List results = new ArrayList<>();
- while (token != null) {
- ResultSet result = resultFetcher.fetchResults(token, Integer.MAX_VALUE);
- results.addAll(result.getData());
- token = result.getNextToken();
+ private void checkSubmissionTargetBeforeChanges() {
+ if (!jobSubmitter.supportsApplicationTargets()) {
+ RefreshJobTarget.forNewJob(operationCtx.getTableConfig());
}
- return results;
}
- private static JobExecutionResult executeRefreshJob(
- String script,
- Configuration executionConfig,
- OperationExecutor operationExecutor,
- OperationHandle operationHandle) {
- String executeTarget = operationExecutor.getSessionContext().getSessionConf().get(TARGET);
- if (executeTarget == null || executeTarget.isEmpty() || "local".equals(executeTarget)) {
- String errorMessage =
- String.format(
- "Unsupported execution target detected: %s."
- + "Currently, only the following execution targets are supported: "
- + "'remote', 'yarn-session', 'yarn-application', 'kubernetes-session', 'kubernetes-application'. ",
- executeTarget);
- LOG.error(errorMessage);
- throw new ValidationException(errorMessage);
- }
-
- if (executeTarget.endsWith("application")) {
- return executeApplicationJob(script, executionConfig, operationExecutor);
- } else {
- return executeNonApplicationJob(
- script, executionConfig, operationExecutor, operationHandle);
+ private void checkRestartTargetBeforeChanges(
+ ObjectIdentifier tableIdentifier, byte[] serializedRefreshHandler) {
+ if (!jobSubmitter.supportsApplicationTargets()) {
+ RefreshJobTarget.forRestart(
+ operationCtx.getTableConfig(),
+ tableIdentifier,
+ deserializeContinuousHandler(serializedRefreshHandler));
}
}
- private static JobExecutionResult executeNonApplicationJob(
- String script,
- Configuration executionConfig,
- OperationExecutor operationExecutor,
- OperationHandle operationHandle) {
- String executeTarget = operationExecutor.getSessionContext().getSessionConf().get(TARGET);
- String clusterId =
- operationExecutor
- .getSessionClusterId()
- .orElseThrow(
- () -> {
- String errorMessage =
- String.format(
- "No cluster ID found when executing materialized table refresh job. Execution target is : %s",
- executeTarget);
- LOG.error(errorMessage);
- return new ValidationException(errorMessage);
- });
-
- ResultFetcher resultFetcher =
- operationExecutor.executeStatement(operationHandle, executionConfig, script);
- List results = fetchAllResults(resultFetcher);
- String jobId = results.get(0).getString(0).toString();
-
- return new JobExecutionResult(executeTarget, clusterId, jobId);
- }
-
- private static JobExecutionResult executeApplicationJob(
- String script, Configuration executionConfig, OperationExecutor operationExecutor) {
- List arguments = new ArrayList<>();
- arguments.add("--" + SqlDriver.OPTION_SQL_SCRIPT.getLongOpt());
- arguments.add(script);
-
- Configuration mergedConfig =
- new Configuration(operationExecutor.getSessionContext().getSessionConf());
- mergedConfig.addAll(executionConfig);
- JobID jobId = new JobID();
- mergedConfig.set(PIPELINE_FIXED_JOB_ID, jobId.toString());
-
- ApplicationConfiguration applicationConfiguration =
- new ApplicationConfiguration(
- arguments.toArray(new String[0]), SqlDriver.class.getName());
- try {
- String clusterId =
- new ApplicationClusterDeployer(new DefaultClusterClientServiceLoader())
- .run(mergedConfig, applicationConfiguration)
- .toString();
-
- return new JobExecutionResult(mergedConfig.get(TARGET), clusterId, jobId.toString());
- } catch (Throwable t) {
- LOG.error("Failed to deploy script {} to application cluster.", script, t);
- throw new SqlGatewayException("Failed to deploy script to cluster.", t);
- }
+ private String requireSavepointDirectory() {
+ return operationCtx
+ .getTableConfig()
+ .getOptional(CheckpointingOptions.SAVEPOINT_DIRECTORY)
+ .orElseThrow(
+ () ->
+ new ValidationException(
+ "Savepoint directory is not configured, can't stop job with savepoint."));
}
- private static class JobExecutionResult {
-
- private final String executionTarget;
- private final String clusterId;
- private final String jobId;
-
- public JobExecutionResult(String executionTarget, String clusterId, String jobId) {
- this.executionTarget = executionTarget;
- this.clusterId = clusterId;
- this.jobId = jobId;
- }
+ private static IntervalFreshness validateAndGetIntervalFreshness(
+ final CatalogMaterializedTable catalogMaterializedTable) {
+ return Optional.ofNullable(catalogMaterializedTable.getDefinitionFreshness())
+ .orElseThrow(() -> new TableException("Freshness cannot be null"));
}
- private static Optional getClusterIdKeyName(String targetName) {
- if (targetName.startsWith("yarn")) {
- return Optional.of("yarn.application.id");
- } else if (targetName.startsWith("kubernetes")) {
- return Optional.of("kubernetes.cluster-id");
- } else {
- return Optional.empty();
- }
+ private RefreshWorkflowContext requireRefreshWorkflowContext(String missingSchedulerMessage) {
+ return jobSubmitter
+ .getRefreshWorkflowContext()
+ .orElseThrow(() -> new TableException(missingSchedulerMessage));
}
- private static IntervalFreshness validateAndGetIntervalFreshness(
- final CatalogMaterializedTable catalogMaterializedTable) {
- return Optional.ofNullable(catalogMaterializedTable.getDefinitionFreshness())
- .orElseThrow(() -> new SqlExecutionException("Freshness cannot be null"));
+ private static RowData refreshRowToInternalRow(Row row) {
+ final Map internalClusterInfo = new HashMap<>();
+ row.