[FLINK-38355][table] Support CREATE OR ALTER MATERIALIZED TABLE syntax

Author: raminqaf

docs/content.zh/docs/dev/table/materialized-table/statements.md

@@ -27,14 +27,14 @@ under the License.
 # 物化表语法
 
 Flink SQL 目前支持以下物化表操作：
-- [CREATE MATERIALIZED TABLE](#create-materialized-table)
+- [CREATE [OR ALTER] MATERIALIZED TABLE](#create-or-alter-materialized-table)
 - [ALTER MATERIALIZED TABLE](#alter-materialized-table)
 - [DROP MATERIALIZED TABLE](#drop-materialized-table)
 
-# CREATE MATERIALIZED TABLE
+# CREATE [OR ALTER] MATERIALIZED TABLE
 
 ```
-CREATE MATERIALIZED TABLE [catalog_name.][db_name.]table_name
+CREATE [OR ALTER] MATERIALIZED TABLE [catalog_name.][db_name.]table_name
 
 [(
     { <physical_column_definition> | <metadata_column_definition> | <computed_column_definition> }[ , ...n]
@@ -228,6 +228,30 @@ CREATE MATERIALIZED TABLE my_materialized_table
     AS SELECT * FROM kafka_catalog.db1.kafka_table;
 ```
 
+## OR ALTER
+
+The `OR ALTER` clause provides create-or-update semantics:
+
+- **If the materialized table does not exist**: Creates a new materialized table with the specified options
+- **If the materialized table exists**: Modifies the query definition (behaves like `ALTER MATERIALIZED TABLE AS`)
+
+This is particularly useful in declarative deployment scenarios where you want to define the desired state without checking if the materialized table already exists.
+
+**Behavior when materialized table exists:**
+
+The operation updates the materialized table similarly to [ALTER MATERIALIZED TABLE AS](#as-select_statement-1):
+
+**Full mode:**
+1. Updates the schema and query definition
+2. The materialized table is refreshed using the new query when the next refresh job is triggered
+
+**Continuous mode:**
+1. Pauses the current running refresh job
+2. Updates the schema and query definition
+3. Starts a new refresh job from the beginning
+
+See [ALTER MATERIALIZED TABLE AS](#as-select_statement-1) for more details.
+
 ## 示例
 
 假定 `materialized-table.refresh-mode.freshness-threshold` 为 30 分钟。
@@ -313,6 +337,46 @@ It might happen that types of columns are not the same, in that case implicit ca
 If for some of the combinations implicit cast is not supported then there will be validation error thrown. 
 Also, it is worth to note that reordering can also be done here. 
 
+Create or alter a materialized table executed twice:
+
+```sql
+-- First execution: creates the materialized table
+CREATE OR ALTER MATERIALIZED TABLE my_materialized_table
+    FRESHNESS = INTERVAL '10' SECOND
+    AS
+    SELECT
+        user_id,
+        COUNT(*) AS event_count,
+        SUM(amount) AS total_amount
+    FROM
+        kafka_catalog.db1.events
+    WHERE
+        event_type = 'purchase'
+    GROUP BY
+        user_id;
+
+-- Second execution: alters the query definition (adds avg_amount column)
+CREATE OR ALTER MATERIALIZED TABLE my_materialized_table
+    FRESHNESS = INTERVAL '10' SECOND
+    AS
+    SELECT
+        user_id,
+        COUNT(*) AS event_count,
+        SUM(amount) AS total_amount,
+        AVG(amount) AS avg_amount  -- Add a new nullable column at the end
+    FROM
+        kafka_catalog.db1.events
+    WHERE
+        event_type = 'purchase'
+    GROUP BY
+        user_id;
+```
+
+<span class="label label-danger">Note</span>
+- When altering an existing materialized table, schema evolution currently only supports adding `nullable` columns to the end of the original materialized table's schema.
+- In continuous mode, the new refresh job will not restore from the state of the original refresh job when altering.
+- All limitations from both CREATE and ALTER operations apply.
+
 ## 限制
 - Does not support explicitly specifying physical columns which are not used in the query
 - 不支持在 select 查询语句中引用临时表、临时视图或临时函数

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

@@ -27,14 +27,14 @@ under the License.
 # Materialized Table Statements
 
 Flink SQL supports the following Materialized Table statements for now:
-- [CREATE MATERIALIZED TABLE](#create-materialized-table)
+- [CREATE [OR ALTER] MATERIALIZED TABLE](#create-or-alter-materialized-table)
 - [ALTER MATERIALIZED TABLE](#alter-materialized-table)
 - [DROP MATERIALIZED TABLE](#drop-materialized-table)
 
-# CREATE MATERIALIZED TABLE
+# CREATE [OR ALTER] MATERIALIZED TABLE
 
 ```
-CREATE MATERIALIZED TABLE [catalog_name.][db_name.]table_name
+CREATE [OR ALTER] MATERIALIZED TABLE [catalog_name.][db_name.]table_name
 
 [(
     { <physical_column_definition> | <metadata_column_definition> | <computed_column_definition> }[ , ...n]
@@ -228,6 +228,30 @@ CREATE MATERIALIZED TABLE my_materialized_table
     AS SELECT * FROM kafka_catalog.db1.kafka_table;
 ```
 
+## OR ALTER
+
+The `OR ALTER` clause provides create-or-update semantics:
+
+- **If the materialized table does not exist**: Creates a new materialized table with the specified options
+- **If the materialized table exists**: Modifies the query definition (behaves like `ALTER MATERIALIZED TABLE AS`)
+
+This is particularly useful in declarative deployment scenarios where you want to define the desired state without checking if the materialized table already exists.
+
+**Behavior when materialized table exists:**
+
+The operation updates the materialized table similarly to [ALTER MATERIALIZED TABLE AS](#as-select_statement-1):
+
+**Full mode:**
+1. Updates the schema and query definition
+2. The materialized table is refreshed using the new query when the next refresh job is triggered
+
+**Continuous mode:**
+1. Pauses the current running refresh job
+2. Updates the schema and query definition
+3. Starts a new refresh job from the beginning
+
+See [ALTER MATERIALIZED TABLE AS](#as-select_statement-1) for more details.
+
 ## Examples
 
 Assuming `materialized-table.refresh-mode.freshness-threshold` is 30 minutes.
@@ -311,6 +335,46 @@ It might happen that types of columns are not the same, in that case implicit ca
 If for some of the combinations implicit cast is not supported then there will be validation error thrown.
 Also, it is worth to note that reordering can also be done here.
 
+Create or alter a materialized table executed twice:
+
+```sql
+-- First execution: creates the materialized table
+CREATE OR ALTER MATERIALIZED TABLE my_materialized_table
+    FRESHNESS = INTERVAL '10' SECOND
+    AS
+    SELECT
+        user_id,
+        COUNT(*) AS event_count,
+        SUM(amount) AS total_amount
+    FROM
+        kafka_catalog.db1.events
+    WHERE
+        event_type = 'purchase'
+    GROUP BY
+        user_id;
+
+-- Second execution: alters the query definition (adds avg_amount column)
+CREATE OR ALTER MATERIALIZED TABLE my_materialized_table
+    FRESHNESS = INTERVAL '10' SECOND
+    AS
+    SELECT
+        user_id,
+        COUNT(*) AS event_count,
+        SUM(amount) AS total_amount,
+        AVG(amount) AS avg_amount  -- Add a new nullable column at the end
+    FROM
+        kafka_catalog.db1.events
+    WHERE
+        event_type = 'purchase'
+    GROUP BY
+        user_id;
+```
+
+<span class="label label-danger">Note</span>
+- When altering an existing materialized table, schema evolution currently only supports adding `nullable` columns to the end of the original materialized table's schema.
+- In continuous mode, the new refresh job will not restore from the state of the original refresh job when altering.
+- All limitations from both CREATE and ALTER operations apply.
+
 ## Limitations
 
 - Does not support explicitly specifying physical columns which are not used in the query 

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

@@ -265,7 +265,7 @@ private void createMaterializedTableInFullMode(
         CreateRefreshWorkflow createRefreshWorkflow =
                 new CreatePeriodicRefreshWorkflow(
                         materializedTableIdentifier,
-                        catalogMaterializedTable.getDefinitionQuery(),
+                        catalogMaterializedTable.getExpandedQuery(),
                         cronExpression,
                         getSessionInitializationConf(operationExecutor),
                         Collections.emptyMap(),
@@ -569,7 +569,7 @@ private void executeContinuousRefreshJob(
         String insertStatement =
                 getInsertStatement(
                         materializedTableIdentifier,
-                        catalogMaterializedTable.getDefinitionQuery(),
+                        catalogMaterializedTable.getExpandedQuery(),
                         dynamicOptions);
 
         JobExecutionResult result =
@@ -651,7 +651,7 @@ public ResultFetcher refreshMaterializedTable(
         String insertStatement =
                 getRefreshStatement(
                         materializedTableIdentifier,
-                        materializedTable.getDefinitionQuery(),
+                        materializedTable.getExpandedQuery(),
                         refreshPartitions,
                         dynamicOptions);
 
@@ -868,8 +868,8 @@ private ResultFetcher callAlterMaterializedTableChangeOperation(
                 LOG.warn(
                         "Failed to start the continuous refresh job for materialized table {} using new query {}, rollback to origin query {}.",
                         tableIdentifier,
-                        op.getCatalogMaterializedTable().getDefinitionQuery(),
-                        suspendMaterializedTable.getDefinitionQuery(),
+                        op.getCatalogMaterializedTable().getExpandedQuery(),
+                        suspendMaterializedTable.getExpandedQuery(),
                         e);
 
                 AlterMaterializedTableChangeOperation rollbackChangeOperation =
@@ -891,7 +891,7 @@ private ResultFetcher callAlterMaterializedTableChangeOperation(
                 throw new SqlExecutionException(
                         String.format(
                                 "Failed to start the continuous refresh job using new query %s when altering materialized table %s select query.",
-                                op.getCatalogMaterializedTable().getDefinitionQuery(),
+                                op.getCatalogMaterializedTable().getExpandedQuery(),
                                 tableIdentifier),
                         e);
             }
@@ -944,8 +944,7 @@ private AlterMaterializedTableChangeOperation generateRollbackAlterMaterializedT
                                 oldMaterializedTable.getSerializedRefreshHandler()));
             } else if (tableChange instanceof TableChange.ModifyDefinitionQuery) {
                 rollbackChanges.add(
-                        TableChange.modifyDefinitionQuery(
-                                oldMaterializedTable.getDefinitionQuery()));
+                        TableChange.modifyDefinitionQuery(oldMaterializedTable.getExpandedQuery()));
             } else {
                 throw new ValidationException(
                         String.format(

flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/service/MaterializedTableStatementITCase.java

@@ -1268,7 +1268,7 @@ void testAlterMaterializedTableAsQueryInFullMode() throws Exception {
                 .isEqualTo(
                         Collections.singletonList(
                                 Column.physical("order_amount_sum", DataTypes.INT())));
-        assertThat(newTable.getDefinitionQuery())
+        assertThat(newTable.getExpandedQuery())
                 .isEqualTo(
                         String.format(
                                 "SELECT `tmp`.`user_id`, `tmp`.`shop_id`, `tmp`.`ds`, COUNT(`tmp`.`order_id`) AS `order_cnt`, SUM(`tmp`.`order_amount`) AS `order_amount_sum`\n"
@@ -1342,7 +1342,7 @@ void testAlterMaterializedTableAddDistribution() throws Exception {
                                         TEST_DEFAULT_DATABASE,
                                         "users_shops"));
 
-        assertThat(newTable.getDefinitionQuery()).isEqualTo(oldTable.getDefinitionQuery());
+        assertThat(newTable.getExpandedQuery()).isEqualTo(oldTable.getExpandedQuery());
 
         // the refresh handler in full mode should be the same as the old one
         assertThat(oldTable.getSerializedRefreshHandler())
@@ -1408,7 +1408,7 @@ void testAlterMaterializedTableAsQueryInFullModeWithSuspendStatus() throws Excep
                 .isEqualTo(
                         Collections.singletonList(
                                 Column.physical("order_amount_sum", DataTypes.INT())));
-        assertThat(newTable.getDefinitionQuery())
+        assertThat(newTable.getExpandedQuery())
                 .isEqualTo(
                         String.format(
                                 "SELECT `tmp`.`user_id`, `tmp`.`shop_id`, `tmp`.`ds`, COUNT(`tmp`.`order_id`) AS `order_cnt`, SUM(`tmp`.`order_amount`) AS `order_amount_sum`\n"
@@ -1495,7 +1495,7 @@ void testAlterMaterializedTableAsQueryInContinuousMode(@TempDir Path temporaryPa
                 .isEqualTo(oldTable.getResolvedSchema().getPrimaryKey());
         assertThat(newTable.getResolvedSchema().getWatermarkSpecs())
                 .isEqualTo(oldTable.getResolvedSchema().getWatermarkSpecs());
-        assertThat(newTable.getDefinitionQuery())
+        assertThat(newTable.getExpandedQuery())
                 .isEqualTo(
                         String.format(
                                 "SELECT COALESCE(`tmp`.`user_id`, CAST(0 AS BIGINT)) AS `user_id`, `tmp`.`shop_id`, COALESCE(`tmp`.`ds`, '') AS `ds`, SUM(`tmp`.`payment_amount_cents`) AS `payed_buy_fee_sum`, SUM(1) AS `pv`\n"
@@ -1594,7 +1594,7 @@ void testAlterMaterializedTableAsQueryInContinuousModeWithSuspendStatus(
 
         assertThat(getAddedColumns(newTable.getResolvedSchema(), oldTable.getResolvedSchema()))
                 .isEqualTo(Collections.singletonList(Column.physical("pv", DataTypes.INT())));
-        assertThat(newTable.getDefinitionQuery())
+        assertThat(newTable.getExpandedQuery())
                 .isEqualTo(
                         String.format(
                                 "SELECT `tmp`.`user_id`, `tmp`.`shop_id`, `tmp`.`ds`, SUM(`tmp`.`payment_amount_cents`) AS `payed_buy_fee_sum`, SUM(1) AS `pv`\n"

flink-table/flink-sql-parser/src/main/codegen/data/Parser.tdd

@@ -76,9 +76,9 @@
     "org.apache.flink.sql.parser.ddl.SqlCreateCatalog"
     "org.apache.flink.sql.parser.ddl.SqlCreateDatabase"
     "org.apache.flink.sql.parser.ddl.SqlCreateFunction"
-    "org.apache.flink.sql.parser.ddl.SqlCreateMaterializedTable"
     "org.apache.flink.sql.parser.ddl.SqlCreateModel"
     "org.apache.flink.sql.parser.ddl.SqlCreateModelAs"
+    "org.apache.flink.sql.parser.ddl.SqlCreateOrAlterMaterializedTable"
     "org.apache.flink.sql.parser.ddl.SqlCreateTable"
     "org.apache.flink.sql.parser.ddl.SqlCreateTable.TableCreationContext"
     "org.apache.flink.sql.parser.ddl.SqlCreateTableAs"

flink-table/flink-sql-parser/src/main/codegen/includes/parserImpls.ftl

@@ -1852,9 +1852,9 @@ SqlNode SqlReplaceTable() :
 }
 
 /**
-  * Parses a CREATE MATERIALIZED TABLE statement.
+  * Parses a CREATE [OR ALTER] MATERIALIZED TABLE statement.
 */
-SqlCreate SqlCreateMaterializedTable(Span s, boolean replace, boolean isTemporary) :
+SqlCreate SqlCreateOrAlterMaterializedTable(Span s, boolean replace, boolean isTemporary) :
 {
     final SqlParserPos startPos = s.pos();
     SqlIdentifier tableName;
@@ -1870,8 +1870,14 @@ SqlCreate SqlCreateMaterializedTable(Span s, boolean replace, boolean isTemporar
     SqlNode asQuery = null;
     SqlParserPos pos = startPos;
     boolean isColumnsIdentifiersOnly = false;
+    boolean isOrAlter = false;
 }
 {
+    [
+      <OR> <ALTER> {
+        isOrAlter = true;
+      }
+    ]
     <MATERIALIZED>
     {
         if (isTemporary) {
@@ -1946,7 +1952,7 @@ SqlCreate SqlCreateMaterializedTable(Span s, boolean replace, boolean isTemporar
     <AS>
     asQuery = OrderedQueryOrExpr(ExprContext.ACCEPT_QUERY)
     {
-        return new SqlCreateMaterializedTable(
+        return new SqlCreateOrAlterMaterializedTable(
             startPos.plus(getPos()),
             tableName,
             columnList,
@@ -1958,7 +1964,8 @@ SqlCreate SqlCreateMaterializedTable(Span s, boolean replace, boolean isTemporar
             propertyList,
             (SqlIntervalLiteral) freshness,
             refreshMode,
-            asQuery);
+            asQuery,
+            isOrAlter);
     }
 }
 
@@ -2646,7 +2653,7 @@ SqlCreate SqlCreateExtended(Span s, boolean replace) :
     (
         create = SqlCreateCatalog(s, replace)
         |
-        create = SqlCreateMaterializedTable(s, replace, isTemporary)
+        create = SqlCreateOrAlterMaterializedTable(s, replace, isTemporary)
         |
         create = SqlCreateTable(s, replace, isTemporary)
         |

flink-table/flink-sql-parser/src/main/codegen/templates/Parser.jj

@@ -4353,6 +4353,7 @@ SqlCreate SqlCreate() :
 {
     <CREATE> { s = span(); }
     [
+        LOOKAHEAD(2)
         <OR> <REPLACE> {
             replace = true;
         }

flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlAlterMaterializedTable.java

@@ -29,8 +29,8 @@
 import static java.util.Objects.requireNonNull;
 
 /**
- * Abstract class to describe statements like ALTER MATERIALIZED TABLE [catalogName.]
- * [dataBasesName.]tableName ...
+ * Abstract class to describe statements like ALTER MATERIALIZED TABLE
+ * [catalogName.][dataBasesName.]tableName ...
  */
 public abstract class SqlAlterMaterializedTable extends SqlCall {
 

flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlAlterMaterializedTableAsQuery.java

@@ -52,7 +52,9 @@ public List<SqlNode> getOperandList() {
     @Override
     public void unparse(SqlWriter writer, int leftPrec, int rightPrec) {
         super.unparse(writer, leftPrec, rightPrec);
+        writer.newlineAndIndent();
         writer.keyword("AS");
+        writer.newlineAndIndent();
         asQuery.unparse(writer, leftPrec, rightPrec);
     }
 }