[FLINK-41209][table] Run materialized table jobs in any TableEnvironment

docs/content.zh/docs/sql/materialized-table/deployment.md

@@ -98,7 +98,7 @@ Catalog 必须支持创建物化表，目前只有 [Paimon Catalog](https://paim
 ### 在 Standalone 集群运行刷新作业
 
 ```sql
-Flink SQL> SET 'execution.mode' = 'remote';
+Flink SQL> SET 'execution.target' = 'remote';
 [INFO] Execute statement succeeded.
 
 FLINK SQL> CREATE MATERIALIZED TABLE my_materialized_table
@@ -113,7 +113,7 @@ FLINK SQL> CREATE MATERIALIZED TABLE my_materialized_table
 **Kubernetes session 模式：**
 
 ```sql
-Flink SQL> SET 'execution.mode' = 'kubernetes-session';
+Flink SQL> SET 'execution.target' = 'kubernetes-session';
 [INFO] Execute statement succeeded.
 
 Flink SQL> SET 'kubernetes.cluster-id' = 'flink-cluster-mt-session-1';
@@ -124,12 +124,12 @@ FLINK SQL> CREATE MATERIALIZED TABLE my_materialized_table
 [INFO] Execute statement succeeded.
 ```
 
-设置 `execution.mode` 为 `kubernetes-session` 并设置参数 `kubernetes.cluster-id` 指向一个已经存在的 Kubernetes session 集群.
+设置 `execution.target` 为 `kubernetes-session` 并设置参数 `kubernetes.cluster-id` 指向一个已经存在的 Kubernetes session 集群.
 
 **YARN session 模式：**
 
 ```sql
-Flink SQL> SET 'execution.mode' = 'yarn-session';
+Flink SQL> SET 'execution.target' = 'yarn-session';
 [INFO] Execute statement succeeded.
 
 Flink SQL> SET 'yarn.application.id' = 'application-xxxx';
@@ -139,14 +139,14 @@ FLINK SQL> CREATE MATERIALIZED TABLE my_materialized_table
 > ...
 [INFO] Execute statement succeeded.
 ```
-设置 `execution.mode` 为 `yarn-session` 并设置参数 `yarn.application.id` 指向一个已经存在的 YARN session 集群。
+设置 `execution.target` 为 `yarn-session` 并设置参数 `yarn.application.id` 指向一个已经存在的 YARN session 集群。
 
 ### 在 application 模式下运行刷新作业
 
 **Kubernetes application 模式：**
 
 ```sql
-Flink SQL> SET 'execution.mode' = 'kubernetes-application';
+Flink SQL> SET 'execution.target' = 'kubernetes-application';
 [INFO] Execute statement succeeded.
 
 Flink SQL> SET 'kubernetes.cluster-id' = 'flink-cluster-mt-application-1';
@@ -156,23 +156,23 @@ FLINK SQL> CREATE MATERIALIZED TABLE my_materialized_table
 > ...
 [INFO] Execute statement succeeded.
 ```
-设置 `execution.mode` 为 `kubernetes-application` ，`kubernetes.cluster-id` 是一个可选配置，如果未配置，在提交作业时会自动生成。
+设置 `execution.target` 为 `kubernetes-application` ，`kubernetes.cluster-id` 是一个可选配置，如果未配置，在提交作业时会自动生成。
 
 **YARN application 模式：**
 
 ```sql
-Flink SQL> SET 'execution.mode' = 'yarn-application';
+Flink SQL> SET 'execution.target' = 'yarn-application';
 [INFO] Execute statement succeeded.
 
 FLINK SQL> CREATE MATERIALIZED TABLE my_materialized_table
 > ...
 [INFO] Execute statement succeeded.
 ```
-设置 `execution.mode` 为 `yarn-application` ，`yarn.application.id` 无需配置。
+设置 `execution.target` 为 `yarn-application` ，`yarn.application.id` 无需配置。
 
 ## 运维操作
 
-集群信息（如 `execution.mode` 或 `kubernetes.cluster-id`）已持久化在 Catalog 中，暂停或恢复物化表刷新作业时无需重复设置。
+集群信息（如 `execution.target` 或 `kubernetes.cluster-id`）已持久化在 Catalog 中，暂停或恢复物化表刷新作业时无需重复设置。
 
 ### 暂停刷新作业
 ```sql

docs/content/docs/sql/materialized-table/deployment.md

@@ -49,7 +49,7 @@ Materialized Table refresh jobs currently support execution in these cluster env
 
 ## Flink SQL Gateway Deployment
 
-Materialized Tables must be created through SQL Gateway, which requires specific configurations for metadata persistence and job scheduling.
+Materialized Tables in full refresh mode must be created and operated through SQL Gateway, which schedules their refresh workflows. SQL Gateway requires specific configurations for metadata persistence and job scheduling. Materialized Tables in continuous refresh mode can also be created from any `TableEnvironment` (see [Using Table API](#using-table-api)).
 
 ### Configure Catalog Store
 
@@ -99,7 +99,7 @@ Example using SQL Client:
 ### Refresh Jobs Running on Standalone Cluster
 
 ```sql
-Flink SQL> SET 'execution.mode' = 'remote';
+Flink SQL> SET 'execution.target' = 'remote';
 [INFO] Execute statement succeeded.
 
 FLINK SQL> CREATE MATERIALIZED TABLE my_materialized_table
@@ -114,7 +114,7 @@ For session modes, pre-create session cluster as documented in  [yarn-session]({
 **Kubernetes session mode:**
 
 ```sql
-Flink SQL> SET 'execution.mode' = 'kubernetes-session';
+Flink SQL> SET 'execution.target' = 'kubernetes-session';
 [INFO] Execute statement succeeded.
 
 Flink SQL> SET 'kubernetes.cluster-id' = 'flink-cluster-mt-session-1';
@@ -124,12 +124,12 @@ Flink SQL> CREATE MATERIALIZED TABLE my_materialized_table
 > ... ;
 [INFO] Execute statement succeeded.
 ```
-Set `execution.mode` to `kubernetes-session` and specify a valid `kubernetes.cluster-id` corresponding to an existing Kubernetes session cluster.
+Set `execution.target` to `kubernetes-session` and specify a valid `kubernetes.cluster-id` corresponding to an existing Kubernetes session cluster.
 
 **YARN session mode:**
 
 ```sql
-Flink SQL> SET 'execution.mode' = 'yarn-session';
+Flink SQL> SET 'execution.target' = 'yarn-session';
 [INFO] Execute statement succeeded.
 
 Flink SQL> SET 'yarn.application.id' = 'application-xxxx';
@@ -139,14 +139,14 @@ Flink SQL> CREATE MATERIALIZED TABLE my_materialized_table
 > ... ;
 [INFO] Execute statement succeeded.
 ```
-Set `execution.mode` to `yarn-session` and specify a valid `yarn.application.id` corresponding to an existing YARN session cluster.
+Set `execution.target` to `yarn-session` and specify a valid `yarn.application.id` corresponding to an existing YARN session cluster.
 
 ### Refresh Jobs Running in Application Mode
 
 **Kubernetes application mode:**
 
 ```sql
-Flink SQL> SET 'execution.mode' = 'kubernetes-application';
+Flink SQL> SET 'execution.target' = 'kubernetes-application';
 [INFO] Execute statement succeeded.
 
 Flink SQL> SET 'kubernetes.cluster-id' = 'flink-cluster-mt-application-1';
@@ -156,23 +156,23 @@ Flink SQL> CREATE MATERIALIZED TABLE my_materialized_table
 > ... ;
 [INFO] Execute statement succeeded.
 ```
-Set `execution.mode` to `kubernetes-application`. The `kubernetes.cluster-id` is optional; if not set, it will be automatically generated.
+Set `execution.target` to `kubernetes-application`. The `kubernetes.cluster-id` is optional; if not set, it will be automatically generated.
 
 **YARN application mode:**
 
 ```sql
-Flink SQL> SET 'execution.mode' = 'yarn-application';
+Flink SQL> SET 'execution.target' = 'yarn-application';
 [INFO] Execute statement succeeded.
 
 Flink SQL> CREATE MATERIALIZED TABLE my_materialized_table
 > ... ;
 [INFO] Execute statement succeeded.
 ```
-Only set `execution.mode` to yarn-application. The `yarn.application.id` doesn’t need to be set; it will be automatically generated during submission.
+Only set `execution.target` to yarn-application. The `yarn.application.id` doesn’t need to be set; it will be automatically generated during submission.
 
 ## Maintenance Operations
 
-Cluster information (e.g., `execution.mode` or `kubernetes.cluster-id`) is already persisted in the catalog and does not need to be set when suspend or resume the refresh jobs of Materialized Table.
+Cluster information (e.g., `execution.target` or `kubernetes.cluster-id`) is already persisted in the catalog and does not need to be set when suspending or dropping a Materialized Table. Resuming a Materialized Table, or altering an activated one, submits a new refresh job to the session's `execution.target` and cluster id, so set them as when creating the Materialized Table.
 
 ### Suspend Refresh Job
 
@@ -199,3 +199,29 @@ Flink SQL> ALTER MATERIALIZED TABLE my_materialized_table
 > ... ;
 [INFO] Execute statement succeeded.
 ```
+
+# Using Table API
+
+Materialized tables in continuous refresh mode can also be managed from any `TableEnvironment`. Full refresh mode still requires SQL Gateway, which schedules the refresh workflows.
+
+Configure the session cluster on the program's `TableConfig`, or in the configuration the program runs with:
+
+| Session cluster | Options |
+|---|---|
+| Standalone | `execution.target: remote`, `rest.address`, `rest.port` |
+| YARN session | `execution.target: yarn-session`, `yarn.application.id` |
+| Kubernetes session | `execution.target: kubernetes-session`, `kubernetes.cluster-id` |
+
+```java
+TableEnvironment tEnv = TableEnvironment.create(EnvironmentSettings.inStreamingMode());
+tEnv.getConfig().set("execution.target", "remote");
+tEnv.getConfig().set("rest.address", "flink-jobmanager");
+tEnv.getConfig().set("execution.checkpointing.savepoint-dir", "s3://my-bucket/savepoints");
+
+tEnv.executeSql("CREATE MATERIALIZED TABLE my_materialized_table ...");
+```
+
+* A statement that submits a refresh job fails before anything changes if the target can't host one: an unset, `local` or `embedded` target, an application target, or a session target without its cluster id. Application mode requires SQL Gateway.
+* Suspending a materialized table, and altering an activated one, stop the refresh job with a savepoint and require `execution.checkpointing.savepoint-dir`.
+* Any program that uses the same catalog, and SQL Gateway, can operate the materialized table. The refresh job is reached through the execution target and cluster id recorded in the catalog. For `remote`, configure the same `rest.address` as the program that submitted it.
+* Resuming a materialized table, and altering an activated one, restart the refresh job on the program's own target, which must match the recorded one. A refresh job that SQL Gateway deployed in its own application cluster can be suspended, refreshed and dropped from a program, but must be resumed or altered through SQL Gateway.

docs/content/docs/sql/materialized-table/statements.md

@@ -696,7 +696,7 @@ ALTER MATERIALIZED TABLE my_materialized_table SET ('format' = 'json', 'sink.par
 ALTER MATERIALIZED TABLE my_materialized_table SET ('format' = 'json', 'format' = 'csv');
 ```
 
-<span class="label label-danger">Note</span> When run through the Flink SQL Gateway, the behavior depends on the refresh mode and current refresh status:
+<span class="label label-danger">Note</span> When run through the Flink SQL Gateway or a `TableEnvironment`, the behavior depends on the refresh mode and current refresh status:
 - `FULL` mode: the change is applied to the catalog. The refresh workflow is not touched.
 - `CONTINUOUS` mode, `ACTIVATED` status: the running refresh job is stopped with savepoint, the change is applied to the catalog, and a new refresh job is started using the updated options. The new job does **not** restore from the savepoint taken during suspend, so streaming state is reset.
 - `CONTINUOUS` mode, `SUSPENDED` status: the change is applied to the catalog and the savepoint stored in the refresh handler is cleared, so the next `RESUME` will also start a fresh job.
@@ -730,7 +730,7 @@ ALTER MATERIALIZED TABLE my_materialized_table RESET ('format', 'sink.parallelis
 ALTER MATERIALIZED TABLE my_materialized_table RESET ('format', 'format');
 ```
 
-<span class="label label-danger">Note</span> When run through the Flink SQL Gateway, the behavior depends on the refresh mode and current refresh status:
+<span class="label label-danger">Note</span> When run through the Flink SQL Gateway or a `TableEnvironment`, the behavior depends on the refresh mode and current refresh status:
 - `FULL` mode: the change is applied to the catalog. The refresh workflow is not touched.
 - `CONTINUOUS` mode, `ACTIVATED` status: the running refresh job is stopped with savepoint, the change is applied to the catalog, and a new refresh job is started using the updated options. The new job does **not** restore from the savepoint taken during suspend, so streaming state is reset.
 - `CONTINUOUS` mode, `SUSPENDED` status: the change is applied to the catalog and the savepoint stored in the refresh handler is cleared, so the next `RESUME` will also start a fresh job.

flink-clients/src/main/java/org/apache/flink/client/deployment/executors/AbstractSessionClusterExecutor.java

@@ -19,6 +19,7 @@
 package org.apache.flink.client.deployment.executors;
 
 import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.JobID;
 import org.apache.flink.api.dag.Pipeline;
 import org.apache.flink.client.ClientUtils;
 import org.apache.flink.client.deployment.ClusterClientFactory;
@@ -27,15 +28,18 @@
 import org.apache.flink.client.program.ClusterClient;
 import org.apache.flink.client.program.ClusterClientProvider;
 import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.DeploymentOptions;
 import org.apache.flink.core.execution.CacheSupportedPipelineExecutor;
 import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.core.execution.JobRetrievalSupportedPipelineExecutor;
 import org.apache.flink.core.execution.JobStatusChangedListener;
 import org.apache.flink.core.execution.JobStatusChangedListenerUtils;
 import org.apache.flink.core.execution.PipelineExecutor;
 import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
 import org.apache.flink.streaming.api.graph.StreamGraph;
 import org.apache.flink.util.AbstractID;
 import org.apache.flink.util.concurrent.ExecutorThreadFactory;
+import org.apache.flink.util.concurrent.FutureUtils;
 import org.apache.flink.util.function.FunctionUtils;
 
 import org.slf4j.Logger;
@@ -63,7 +67,7 @@
 @Internal
 public class AbstractSessionClusterExecutor<
                 ClusterID, ClientFactory extends ClusterClientFactory<ClusterID>>
-        implements CacheSupportedPipelineExecutor {
+        implements CacheSupportedPipelineExecutor, JobRetrievalSupportedPipelineExecutor {
     private static final Logger LOG = LoggerFactory.getLogger(AbstractSessionClusterExecutor.class);
     private final ExecutorService executorService =
             Executors.newFixedThreadPool(
@@ -135,6 +139,28 @@ public CompletableFuture<JobClient> execute(
         }
     }
 
+    @Override
+    public CompletableFuture<JobClient> retrieveJobClient(
+            JobID jobId, Configuration configuration, ClassLoader userCodeClassloader) {
+        try {
+            final ClusterID clusterId = clusterClientFactory.getClusterId(configuration);
+            checkState(
+                    clusterId != null,
+                    "No cluster id is configured for execution target '%s'.",
+                    configuration.get(DeploymentOptions.TARGET));
+
+            try (ClusterDescriptor<ClusterID> clusterDescriptor =
+                    clusterClientFactory.createClusterDescriptor(
+                            new Configuration(configuration))) {
+                return CompletableFuture.completedFuture(
+                        new ClusterClientJobClientAdapter<>(
+                                clusterDescriptor.retrieve(clusterId), jobId, userCodeClassloader));
+            }
+        } catch (Exception e) {
+            return FutureUtils.completedExceptionally(e);
+        }
+    }
+
     @Override
     public CompletableFuture<Set<AbstractID>> listCompletedClusterDatasetIds(
             Configuration configuration, ClassLoader userCodeClassloader) throws Exception {

flink-core/src/main/java/org/apache/flink/core/execution/JobRetrievalSupportedPipelineExecutor.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.core.execution;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.configuration.Configuration;
+
+import java.util.concurrent.CompletableFuture;
+
+/** A pipeline executor that can reach jobs it did not submit. */
+@Internal
+public interface JobRetrievalSupportedPipelineExecutor extends PipelineExecutor {
+
+    /**
+     * A JobClient for an existing job on the cluster the configuration names. Fails if the cluster
+     * can't be retrieved.
+     */
+    CompletableFuture<JobClient> retrieveJobClient(
+            JobID jobId, Configuration configuration, ClassLoader userCodeClassloader);
+}

flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/SqlGatewayServiceImpl.java

@@ -355,6 +355,17 @@ public <ClusterID> ClusterID deployScript(
         Configuration mergedConfig = Configuration.fromMap(session.getSessionConfig());
         mergedConfig.addAll(executionConfig);
 
+        try {
+            return deployApplicationCluster(mergedConfig, scriptUri, script);
+        } catch (Throwable t) {
+            LOG.error("Failed to deploy script to cluster.", t);
+            throw new SqlGatewayException("Failed to deploy script to cluster.", t);
+        }
+    }
+
+    public static <ClusterID> ClusterID deployApplicationCluster(
+            Configuration mergedConfig, @Nullable URI scriptUri, @Nullable String script)
+            throws Exception {
         List<String> arguments = new ArrayList<>();
         if (scriptUri != null) {
             arguments.add("--" + SqlDriver.OPTION_SQL_FILE.getLongOpt());
@@ -368,13 +379,8 @@ public <ClusterID> ClusterID deployScript(
         ApplicationConfiguration applicationConfiguration =
                 new ApplicationConfiguration(
                         arguments.toArray(new String[0]), SqlDriver.class.getName());
-        try {
-            return new ApplicationClusterDeployer(new DefaultClusterClientServiceLoader())
-                    .run(mergedConfig, applicationConfiguration);
-        } catch (Throwable t) {
-            LOG.error("Failed to deploy script to cluster.", t);
-            throw new SqlGatewayException("Failed to deploy script to cluster.", t);
-        }
+        return new ApplicationClusterDeployer(new DefaultClusterClientServiceLoader())
+                .run(mergedConfig, applicationConfiguration);
     }
 
     @Override