[FLINK-38355] Support CreateOrAlter MATERIALIZED TABLE

Author: raminqaf

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

@@ -29,6 +29,7 @@ under the License.
 Flink SQL 目前支持以下物化表操作：
 - [CREATE MATERIALIZED TABLE](#create-materialized-table)
 - [ALTER MATERIALIZED TABLE](#alter-materialized-table)
+- [CREATE OR ALTER MATERIALIZED TABLE](#create-or-alter-materialized-table)
 - [DROP MATERIALIZED TABLE](#drop-materialized-table)
 
 # CREATE MATERIALIZED TABLE
@@ -389,6 +390,102 @@ GROUP BY
 - Schema 演进当前仅支持在原表 schema 尾部追加`可空列`。
 - 在持续模式下，新的流式作业不会从原来的流式作业的状态恢复。这可能会导致短暂的数据重复或丢失。
 
+
+# CREATE OR ALTER MATERIALIZED TABLE
+
+```
+CREATE OR ALTER MATERIALIZED TABLE [catalog_name.][db_name.]table_name
+
+[ ([ <table_constraint> ]) ]
+
+[COMMENT table_comment]
+
+[PARTITIONED BY (partition_column_name1, partition_column_name2, ...)]
+
+[WITH (key1=val1, key2=val2, ...)]
+
+FRESHNESS = INTERVAL '<num>' { SECOND | MINUTE | HOUR | DAY }
+
+[REFRESH_MODE = { CONTINUOUS | FULL }]
+
+AS <select_statement>
+
+<table_constraint>:
+  [CONSTRAINT constraint_name] PRIMARY KEY (column_name, ...) NOT ENFORCED
+```
+
+`CREATE OR ALTER MATERIALIZED TABLE` is a convenient statement that combines create and alter functionality:
+
+- **If the table does not exist**: Creates a new materialized table (behaves like [CREATE MATERIALIZED TABLE](#create-materialized-table))
+- **If the table exists**: Modifies the query definition (behaves like [ALTER MATERIALIZED TABLE AS](#as-select_statement-1))
+
+This is particularly useful in declarative deployment scenarios where you want to define the desired state of a materialized table without needing to check if it already exists.
+
+## Syntax Details
+
+The syntax is identical to `CREATE MATERIALIZED TABLE`. See the sections above for detailed explanations of:
+- [PRIMARY KEY](#primary-key)
+- [PARTITIONED BY](#partitioned-by)
+- [WITH Options](#with-options)
+- [FRESHNESS](#freshness)
+- [REFRESH_MODE](#refresh_mode)
+- [AS <select_statement>](#as-select_statement)
+
+## Behavior When Table Exists
+
+When the materialized table already exists, the operation behaves as if you ran `ALTER MATERIALIZED TABLE AS <select_statement>`:
+
+**Full mode:**
+1. Update the `schema` and `query definition` of the materialized table.
+2. The table is refreshed using the new query definition when the next refresh job is triggered.
+
+**Continuous mode:**
+1. Pause the current running refresh job.
+2. Update the `schema` and `query definition` of the materialized table.
+3. Start a new refresh job to refresh the materialized table from the beginning.
+
+See [ALTER MATERIALIZED TABLE AS](#as-select_statement-1) for more details.
+
+## Examples
+
+
+```sql
+-- First execution: creates the 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
+    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 table, schema evolution currently only supports adding `nullable` columns to the end of the original 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.
+
 # DROP MATERIALIZED TABLE
 
 ```text

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

@@ -29,6 +29,7 @@ under the License.
 Flink SQL supports the following Materialized Table statements for now:
 - [CREATE MATERIALIZED TABLE](#create-materialized-table)
 - [ALTER MATERIALIZED TABLE](#alter-materialized-table)
+- [CREATE OR ALTER MATERIALIZED TABLE](#create-or-alter-materialized-table)
 - [DROP MATERIALIZED TABLE](#drop-materialized-table)
 
 # CREATE MATERIALIZED TABLE
@@ -391,6 +392,101 @@ ALTER MATERIALIZED TABLE my_materialized_table
 - Schema evolution currently only supports adding `nullable` columns to the end of the original table's schema.
 - In continuous mode, the new refresh job will not restore from the state of the original refresh job. This may result in temporary data duplication or loss.
 
+# CREATE OR ALTER MATERIALIZED TABLE
+
+```
+CREATE OR ALTER MATERIALIZED TABLE [catalog_name.][db_name.]table_name
+
+[ ([ <table_constraint> ]) ]
+
+[COMMENT table_comment]
+
+[PARTITIONED BY (partition_column_name1, partition_column_name2, ...)]
+
+[WITH (key1=val1, key2=val2, ...)]
+
+FRESHNESS = INTERVAL '<num>' { SECOND | MINUTE | HOUR | DAY }
+
+[REFRESH_MODE = { CONTINUOUS | FULL }]
+
+AS <select_statement>
+
+<table_constraint>:
+  [CONSTRAINT constraint_name] PRIMARY KEY (column_name, ...) NOT ENFORCED
+```
+
+`CREATE OR ALTER MATERIALIZED TABLE` is a convenient statement that combines create and alter functionality:
+
+- **If the table does not exist**: Creates a new materialized table (behaves like [CREATE MATERIALIZED TABLE](#create-materialized-table))
+- **If the table exists**: Modifies the query definition (behaves like [ALTER MATERIALIZED TABLE AS](#as-select_statement-1))
+
+This is particularly useful in declarative deployment scenarios where you want to define the desired state of a materialized table without needing to check if it already exists.
+
+## Syntax Details
+
+The syntax is identical to `CREATE MATERIALIZED TABLE`. See the sections above for detailed explanations of:
+- [PRIMARY KEY](#primary-key)
+- [PARTITIONED BY](#partitioned-by)
+- [WITH Options](#with-options)
+- [FRESHNESS](#freshness)
+- [REFRESH_MODE](#refresh_mode)
+- [AS <select_statement>](#as-select_statement)
+
+## Behavior When Table Exists
+
+When the materialized table already exists, the operation behaves as if you ran `ALTER MATERIALIZED TABLE AS <select_statement>`:
+
+**Full mode:**
+1. Update the `schema` and `query definition` of the materialized table.
+2. The table is refreshed using the new query definition when the next refresh job is triggered.
+
+**Continuous mode:**
+1. Pause the current running refresh job.
+2. Update the `schema` and `query definition` of the materialized table.
+3. Start a new refresh job to refresh the materialized table from the beginning.
+
+See [ALTER MATERIALIZED TABLE AS](#as-select_statement-1) for more details.
+
+## Examples
+
+
+```sql
+-- First execution: creates the 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
+    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 table, schema evolution currently only supports adding `nullable` columns to the end of the original 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.
+
 # DROP MATERIALIZED TABLE
 
 ```text

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

@@ -166,8 +166,7 @@ public void close() throws Exception {
     public ResultFetcher callMaterializedTableOperation(
             OperationExecutor operationExecutor,
             OperationHandle handle,
-            MaterializedTableOperation op,
-            String statement) {
+            MaterializedTableOperation op) {
         if (op instanceof CreateMaterializedTableOperation) {
             return callCreateMaterializedTableOperation(
                     operationExecutor, handle, (CreateMaterializedTableOperation) op);
@@ -267,7 +266,7 @@ private void createMaterializedTableInFullMode(
         CreateRefreshWorkflow createRefreshWorkflow =
                 new CreatePeriodicRefreshWorkflow(
                         materializedTableIdentifier,
-                        catalogMaterializedTable.getDefinitionQuery(),
+                        catalogMaterializedTable.getExpandedQuery(),
                         cronExpression,
                         getSessionInitializationConf(operationExecutor),
                         Collections.emptyMap(),
@@ -570,7 +569,7 @@ private void executeContinuousRefreshJob(
         String insertStatement =
                 getInsertStatement(
                         materializedTableIdentifier,
-                        catalogMaterializedTable.getDefinitionQuery(),
+                        catalogMaterializedTable.getExpandedQuery(),
                         dynamicOptions);
 
         JobExecutionResult result =
@@ -652,7 +651,7 @@ public ResultFetcher refreshMaterializedTable(
         String insertStatement =
                 getRefreshStatement(
                         materializedTableIdentifier,
-                        materializedTable.getDefinitionQuery(),
+                        materializedTable.getExpandedQuery(),
                         refreshPartitions,
                         dynamicOptions);
 
@@ -869,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 =
@@ -892,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);
             }
@@ -945,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/main/java/org/apache/flink/table/gateway/service/operation/OperationExecutor.java

@@ -556,8 +556,7 @@ private ResultFetcher executeOperation(
             return sessionContext
                     .getSessionState()
                     .materializedTableManager
-                    .callMaterializedTableOperation(
-                            this, handle, (MaterializedTableOperation) op, statement);
+                    .callMaterializedTableOperation(this, handle, (MaterializedTableOperation) op);
         } else {
             return callOperation(tableEnv, handle, op);
         }

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

@@ -1083,7 +1083,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"
@@ -1223,7 +1223,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"
@@ -1310,7 +1310,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"
@@ -1409,7 +1409,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,7 +76,7 @@
     "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.SqlCreateOrAlterMaterializedTable"
     "org.apache.flink.sql.parser.ddl.SqlCreateModel"
     "org.apache.flink.sql.parser.ddl.SqlCreateModelAs"
     "org.apache.flink.sql.parser.ddl.SqlCreateTable"

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;
@@ -1866,8 +1866,14 @@ SqlCreate SqlCreateMaterializedTable(Span s, boolean replace, boolean isTemporar
     SqlNode freshness = null;
     SqlLiteral refreshMode = null;
     SqlNode asQuery = null;
+    boolean isOrAlter = false;
 }
 {
+    [
+      <OR> <ALTER> {
+        isOrAlter = true;
+      }
+    ]
     <MATERIALIZED>
     {
         if (isTemporary) {
@@ -1934,7 +1940,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,
             constraint,
@@ -1944,7 +1950,8 @@ SqlCreate SqlCreateMaterializedTable(Span s, boolean replace, boolean isTemporar
             propertyList,
             (SqlIntervalLiteral) freshness,
             refreshMode,
-            asQuery);
+            asQuery,
+            isOrAlter);
     }
 }
 
@@ -2632,7 +2639,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)
         |