[FLINK-41214][table] Add Table#materializeInto for materialized tables

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

@@ -0,0 +1,81 @@
+---
+title: Table API
+weight: 5
+type: docs
+---
+<!--
+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.
+-->
+
+`Table#materializeInto` declares the query of a `Table` as the definition of a materialized table,
+and `execute()` submits the declaration with the semantics of
+[`CREATE OR ALTER MATERIALIZED TABLE ... AS <query>`]({{< ref "docs/sql/materialized-table/statements" >}}#or-alter).
+
+# Setup
+
+Continuous refresh jobs need a session cluster as the execution target of the `TableEnvironment`,
+and redeploying a running materialized table needs a savepoint directory. See
+[Using Table API]({{< ref "docs/sql/materialized-table/deployment" >}}#using-table-api) for the
+supported targets. The catalog must supply the storage of the materialized table.
+
+# Creating a Materialized Table
+
+```java
+MaterializedTable revenue = tEnv.from("orders")
+        .groupBy($("region"))
+        .select($("region"), $("amount").sum().as("revenue"))
+        .materializeInto("revenue", Duration.ofMinutes(1))
+        .execute();
+```
+
+The schema is derived from the stored SQL text of the query, and the path is resolved against the
+current catalog and database. The freshness is optional and defaults to
+[materialized-table.default-freshness.continuous]({{< ref "docs/dev/table/config" >}}#materialized-table-default-freshness-continuous).
+The refresh mode is derived from the freshness (see [Refresh Mode]({{< ref "docs/sql/materialized-table/overview" >}}#refresh-mode)).
+Full refresh mode needs a workflow scheduler and cannot be used from a plain `TableEnvironment`.
+
+# Redeploying
+
+`execute()` on an existing materialized table alters it, and a changed query follows the
+schema-evolution rules of
+[ALTER MATERIALIZED TABLE AS]({{< ref "docs/sql/materialized-table/statements" >}}#as-select_statement-1).
+A running continuous refresh job is stopped with a savepoint and restarted without restoring it,
+even if the declaration is unchanged, so streaming state is reset.
+
+An existing materialized table keeps its freshness, so a freshness passed to `materializeInto` is
+ignored. As with `CREATE OR ALTER` without `WITH` and `DISTRIBUTED BY` clauses, its options are
+reset and its distribution is dropped. To keep them, redeploy with SQL `CREATE OR ALTER` and
+restate the clauses.
+
+If the changes are rejected, for example because the new query drops a column, `execute()` fails
+and leaves the materialized table suspended with its previous definition. Resume it with
+`ALTER MATERIALIZED TABLE ... RESUME` before you redeploy, because redeploying a suspended
+materialized table clears its savepoint.
+
+If the new refresh job starts but the catalog cannot record it, the job is cancelled and the old
+job restarts from its savepoint. If that cancel fails, the materialized table is rolled back and
+left suspended instead of restarting the old job.
+
+# Limitations
+
+- The query of a refresh job is stored as SQL text, so it must be expressible in SQL. Inline
+  functions (unless a custom `SqlFactory` serializes them), tables created from a `TableDescriptor`
+  or a `DataStream`, and operations without a SQL form, such as session group windows, fail at
+  `materializeInto`.
+- As in SQL, referring to temporary tables, views, functions or models is not supported. They are
+  not rejected, but a refresh job started from another session can't resolve them.

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

@@ -0,0 +1,81 @@
+---
+title: Table API
+weight: 5
+type: docs
+---
+<!--
+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.
+-->
+
+`Table#materializeInto` declares the query of a `Table` as the definition of a materialized table,
+and `execute()` submits the declaration with the semantics of
+[`CREATE OR ALTER MATERIALIZED TABLE ... AS <query>`]({{< ref "docs/sql/materialized-table/statements" >}}#or-alter).
+
+# Setup
+
+Continuous refresh jobs need a session cluster as the execution target of the `TableEnvironment`,
+and redeploying a running materialized table needs a savepoint directory. See
+[Using Table API]({{< ref "docs/sql/materialized-table/deployment" >}}#using-table-api) for the
+supported targets. The catalog must supply the storage of the materialized table.
+
+# Creating a Materialized Table
+
+```java
+MaterializedTable revenue = tEnv.from("orders")
+        .groupBy($("region"))
+        .select($("region"), $("amount").sum().as("revenue"))
+        .materializeInto("revenue", Duration.ofMinutes(1))
+        .execute();
+```
+
+The schema is derived from the stored SQL text of the query, and the path is resolved against the
+current catalog and database. The freshness is optional and defaults to
+[materialized-table.default-freshness.continuous]({{< ref "docs/dev/table/config" >}}#materialized-table-default-freshness-continuous).
+The refresh mode is derived from the freshness (see [Refresh Mode]({{< ref "docs/sql/materialized-table/overview" >}}#refresh-mode)).
+Full refresh mode needs a workflow scheduler and cannot be used from a plain `TableEnvironment`.
+
+# Redeploying
+
+`execute()` on an existing materialized table alters it, and a changed query follows the
+schema-evolution rules of
+[ALTER MATERIALIZED TABLE AS]({{< ref "docs/sql/materialized-table/statements" >}}#as-select_statement-1).
+A running continuous refresh job is stopped with a savepoint and restarted without restoring it,
+even if the declaration is unchanged, so streaming state is reset.
+
+An existing materialized table keeps its freshness, so a freshness passed to `materializeInto` is
+ignored. As with `CREATE OR ALTER` without `WITH` and `DISTRIBUTED BY` clauses, its options are
+reset and its distribution is dropped. To keep them, redeploy with SQL `CREATE OR ALTER` and
+restate the clauses.
+
+If the changes are rejected, for example because the new query drops a column, `execute()` fails
+and leaves the materialized table suspended with its previous definition. Resume it with
+`ALTER MATERIALIZED TABLE ... RESUME` before you redeploy, because redeploying a suspended
+materialized table clears its savepoint.
+
+If the new refresh job starts but the catalog cannot record it, the job is cancelled and the old
+job restarts from its savepoint. If that cancel fails, the materialized table is rolled back and
+left suspended instead of restarting the old job.
+
+# Limitations
+
+- The query of a refresh job is stored as SQL text, so it must be expressible in SQL. Inline
+  functions (unless a custom `SqlFactory` serializes them), tables created from a `TableDescriptor`
+  or a `DataStream`, and operations without a SQL form, such as session group windows, fail at
+  `materializeInto`.
+- As in SQL, referring to temporary tables, views, functions or models is not supported. They are
+  not rejected, but a refresh job started from another session can't resolve them.

flink-python/pyflink/table/tests/test_table_completeness.py

@@ -42,6 +42,7 @@ def excluded_methods(cls):
             'asArgument',
             'process',
             'partitionBy',
+            'materializeInto',
         }
 
     @classmethod

flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/MaterializedPipeline.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.api;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.table.catalog.ObjectIdentifier;
+
+/**
+ * A pending declaration of one materialized table, created by {@link
+ * Table#materializeInto(String)}.
+ */
+@PublicEvolving
+public interface MaterializedPipeline {
+
+    /**
+     * The fully qualified identifier this declaration will create or alter.
+     *
+     * <p>Resolved against the current catalog and database at {@link Table#materializeInto}.
+     */
+    ObjectIdentifier getIdentifier();
+
+    /**
+     * Submits this declaration with the semantics of {@code CREATE OR ALTER MATERIALIZED TABLE}.
+     *
+     * <p>If the materialized table exists and its continuous refresh job is running, the job is
+     * stopped with a savepoint and a new refresh job is started, which does not restore that
+     * savepoint. This also happens if the declaration is unchanged. The stop requires a savepoint
+     * directory ({@code execution.checkpointing.savepoint-dir}); without one, this fails before the
+     * stop. If the changes are rejected or cannot be written after the stop, the materialized table
+     * is left suspended with its previous definition and the savepoint. If the new refresh job
+     * starts but the catalog cannot record it, that job is cancelled and the previous refresh job
+     * restarts from the savepoint. If the cancel fails, the materialized table is rolled back and
+     * left suspended instead.
+     *
+     * <p>A suspended materialized table is altered and stays suspended, and its stored savepoint is
+     * cleared, so a later {@code RESUME} starts the refresh job without state. A materialized table
+     * in full refresh mode is altered without changing its refresh workflow.
+     *
+     * <p>An existing materialized table keeps its freshness, comment and partition keys, so a
+     * freshness given to {@link Table#materializeInto(String, java.time.Duration)} is ignored. Its
+     * options are reset and its distribution is dropped, as by a {@code CREATE OR ALTER
+     * MATERIALIZED TABLE} without {@code WITH} and {@code DISTRIBUTED BY} clauses.
+     *
+     * <p>Returns once the catalog entry is written and, if a continuous refresh job is started,
+     * once that job has been submitted.
+     */
+    MaterializedTable execute();
+}

flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/MaterializedTable.java

@@ -0,0 +1,32 @@
+/*
+ * 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;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.table.catalog.ObjectIdentifier;
+
+/**
+ * A handle on an existing materialized table, returned by {@link MaterializedPipeline#execute()}.
+ */
+@PublicEvolving
+public interface MaterializedTable {
+
+    /** The fully qualified identifier of the materialized table. */
+    ObjectIdentifier getIdentifier();
+}

flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Table.java

@@ -33,6 +33,8 @@
 import org.apache.flink.table.operations.QueryOperation;
 import org.apache.flink.table.types.DataType;
 
+import java.time.Duration;
+
 /**
  * The {@link Table} object is the core abstraction of the Table API. Similar to how the DataStream
  * API has {@code DataStream}s, the Table API is built around {@link Table}s.
@@ -1306,6 +1308,52 @@ default TableResult executeInsert(
         return insertInto(descriptor, conflictStrategy, overwrite).execute();
     }
 
+    /**
+     * Declares this table's query as the definition of a materialized table at the given path, with
+     * the default freshness and an automatically chosen refresh mode.
+     *
+     * <p>The query is stored as SQL text, which also defines the schema of the materialized table,
+     * and re-run by the refresh job, also from other sessions, so it must be expressible in SQL.
+     *
+     * <p>The path is resolved against the current catalog and database when this method is called.
+     * See {@link TableEnvironment#useDatabase(String)} and {@link
+     * TableEnvironment#useCatalog(String)} for the rules on path resolution.
+     *
+     * <p>Example:
+     *
+     * <pre>{@code
+     * MaterializedTable revenue = tableEnv.from("orders")
+     *     .groupBy($("region"))
+     *     .select($("region"), $("amount").sum().as("revenue"))
+     *     .materializeInto("revenue")
+     *     .execute();
+     * }</pre>
+     *
+     * @param path The path of the materialized table.
+     * @return The declaration.
+     * @throws SqlParserException if the path is not a valid identifier
+     * @throws ValidationException if the query cannot be stored as SQL
+     */
+    MaterializedPipeline materializeInto(String path);
+
+    /**
+     * Like {@link #materializeInto(String)}, with the given freshness.
+     *
+     * <p>The refresh mode is chosen automatically: when the materialized table is created, a
+     * freshness at or above {@code materialized-table.refresh-mode.freshness-threshold} resolves to
+     * FULL, which requires a workflow scheduler and fails on a plain {@link TableEnvironment}.
+     *
+     * @param path The path of the materialized table.
+     * @param freshness The freshness, a positive whole number of seconds.
+     * @return The declaration.
+     * @throws SqlParserException if the path is not a valid identifier
+     * @throws ValidationException if the freshness is not positive, has a sub-second part, or
+     *     exceeds {@link Integer#MAX_VALUE} seconds
+     * @throws ValidationException if the query cannot be stored as SQL, see {@link
+     *     #materializeInto(String)}
+     */
+    MaterializedPipeline materializeInto(String path, Duration freshness);
+
     /**
      * Partitions the table by a set of partition keys.
      *