This is an automated email from the ASF dual-hosted git repository.
snuyanzin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 9a72fde4824 [FLINK-39662][table] Support `ALTER MATERIALIZED TABLE ...
SET (...)` conversion
9a72fde4824 is described below
commit 9a72fde482412d1cf41c215a229c5a6cb3e1cc69
Author: Ramin Gharib <[email protected]>
AuthorDate: Tue May 19 12:13:55 2026 +0200
[FLINK-39662][table] Support `ALTER MATERIALIZED TABLE ... SET (...)`
conversion
---
.../docs/sql/materialized-table/statements.md | 30 ++++++++++
.../docs/sql/materialized-table/statements.md | 30 ++++++++++
.../service/MaterializedTableStatementITCase.java | 28 ++++++++++
.../operations/converters/SqlNodeConverters.java | 2 +
.../SqlAlterMaterializedTableOptionsConverter.java | 64 ++++++++++++++++++++++
...erializedTableNodeToOperationConverterTest.java | 46 +++++++++++++++-
6 files changed, 197 insertions(+), 3 deletions(-)
diff --git a/docs/content.zh/docs/sql/materialized-table/statements.md
b/docs/content.zh/docs/sql/materialized-table/statements.md
index fce01e00f69..260027f6d02 100644
--- a/docs/content.zh/docs/sql/materialized-table/statements.md
+++ b/docs/content.zh/docs/sql/materialized-table/statements.md
@@ -398,6 +398,7 @@ ALTER MATERIALIZED TABLE [catalog_name.][db_name.]table_name
| DROP {column_name | (column_name, column_name, ...) | PRIMARY KEY |
CONSTRAINT constraint_name | WATERMARK | DISTRIBUTION }
| SUSPEND | RESUME [WITH (key1=val1, key2=val2, ...)]
| REFRESH [PARTITION partition_spec] |
+ | SET (key1=val1, key2=val2, ...)
| RESET (key1, key2, ...)
| AS <select_statement>
<schema_component>:
@@ -549,6 +550,35 @@ ALTER MATERIALIZED TABLE my_materialized_table RESUME;
ALTER MATERIALIZED TABLE my_materialized_table RESUME WITH
('sink.parallelism'='10');
```
+## SET
+
+```
+ALTER MATERIALIZED TABLE [catalog_name.][db_name.]table_name SET (key1=val1,
key2=val2, ...)
+```
+
+`SET` is used to add or overwrite table options of a materialized table.
Existing options not listed in the statement are preserved.
+
+**Key handling:**
+- Keys are applied in the order they appear in the statement.
+- If the same key is listed multiple times, the last value in the list wins.
+- The empty option list `SET ()` is rejected with a validation error.
+
+**Example:**
+
+```sql
+-- Add a new option and overwrite an existing one
+ALTER MATERIALIZED TABLE my_materialized_table SET ('format' = 'json',
'sink.parallelism' = '4');
+
+-- Duplicate keys: the last value wins. After this statement, 'format' is
'csv'.
+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:
+- `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.
+- `CONTINUOUS` mode, `INITIALIZING` status: the statement is rejected.
+
## RESET
```
diff --git a/docs/content/docs/sql/materialized-table/statements.md
b/docs/content/docs/sql/materialized-table/statements.md
index 638e6611d65..5b0628219cb 100644
--- a/docs/content/docs/sql/materialized-table/statements.md
+++ b/docs/content/docs/sql/materialized-table/statements.md
@@ -397,6 +397,7 @@ ALTER MATERIALIZED TABLE [catalog_name.][db_name.]table_name
| DROP {column_name | (column_name, column_name, ...) | PRIMARY KEY |
CONSTRAINT constraint_name | WATERMARK | DISTRIBUTION }
| SUSPEND | RESUME [WITH (key1=val1, key2=val2, ...)]
| REFRESH [PARTITION partition_spec] |
+ | SET (key1=val1, key2=val2, ...)
| RESET (key1, key2, ...)
| AS <select_statement>
<schema_component>:
@@ -549,6 +550,35 @@ ALTER MATERIALIZED TABLE my_materialized_table RESUME;
ALTER MATERIALIZED TABLE my_materialized_table RESUME WITH
('sink.parallelism'='10');
```
+## SET
+
+```
+ALTER MATERIALIZED TABLE [catalog_name.][db_name.]table_name SET (key1=val1,
key2=val2, ...)
+```
+
+`SET` is used to add or overwrite table options of a materialized table.
Existing options not listed in the statement are preserved.
+
+**Key handling:**
+- Keys are applied in the order they appear in the statement.
+- If the same key is listed multiple times, the last value in the list wins.
+- The empty option list `SET ()` is rejected with a validation error.
+
+**Example:**
+
+```sql
+-- Add a new option and overwrite an existing one
+ALTER MATERIALIZED TABLE my_materialized_table SET ('format' = 'json',
'sink.parallelism' = '4');
+
+-- Duplicate keys: the last value wins. After this statement, 'format' is
'csv'.
+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:
+- `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.
+- `CONTINUOUS` mode, `INITIALIZING` status: the statement is rejected.
+
## RESET
```
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 0fc7c51304d..4e128e2cd8d 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
@@ -1155,6 +1155,34 @@ class MaterializedTableStatementITCase extends
AbstractMaterializedTableStatemen
.isEqualTo(TableDistribution.of(Kind.HASH, 2,
List.of("order_id")));
}
+ @Test
+ void testAlterMaterializedTableSetInFullMode() throws Exception {
+ createAndVerifyCreateMaterializedTableWithData(
+ "users_shops", List.of(), Map.of(), RefreshMode.FULL);
+
+ ObjectIdentifier userShopsIdentifier =
getObjectIdentifier("users_shops");
+ ResolvedCatalogMaterializedTable oldTable =
getTable(userShopsIdentifier);
+
+ String alterMaterializedTableSetDDL =
+ "ALTER MATERIALIZED TABLE users_shops SET ('format' = 'json',
'k1' = 'v1')";
+ OperationHandle alterMaterializedTableSetHandle =
+ executeStatement(alterMaterializedTableSetDDL);
+ awaitOperationTermination(service, sessionHandle,
alterMaterializedTableSetHandle);
+
+ ResolvedCatalogMaterializedTable newTable =
getTable(userShopsIdentifier);
+
+ // overriding existing key and adding a new key
+ assertThat(newTable.getOptions()).containsEntry("format",
"json").containsEntry("k1", "v1");
+
+ // unchanged: schema, query, distribution, freshness, refresh handler
+
assertThat(newTable.getResolvedSchema()).isEqualTo(oldTable.getResolvedSchema());
+
assertThat(newTable.getExpandedQuery()).isEqualTo(oldTable.getExpandedQuery());
+
assertThat(newTable.getDistribution()).isEqualTo(oldTable.getDistribution());
+
assertThat(newTable.getDefinitionFreshness()).isEqualTo(oldTable.getDefinitionFreshness());
+ assertThat(newTable.getSerializedRefreshHandler())
+ .isEqualTo(oldTable.getSerializedRefreshHandler());
+ }
+
@Test
void testAlterMaterializedTableResetInFullMode() throws Exception {
createAndVerifyCreateMaterializedTableWithData(
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlNodeConverters.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlNodeConverters.java
index 7a1d9b67a52..7e03c203664 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlNodeConverters.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlNodeConverters.java
@@ -30,6 +30,7 @@ import
org.apache.flink.table.planner.operations.converters.materializedtable.Sq
import
org.apache.flink.table.planner.operations.converters.materializedtable.SqlAlterMaterializedTableDropSchemaConverter.SqlAlterMaterializedTableDropWatermarkConverter;
import
org.apache.flink.table.planner.operations.converters.materializedtable.SqlAlterMaterializedTableDropSchemaConverter.SqlAlterMaterializedTableSchemaDropColumnConverter;
import
org.apache.flink.table.planner.operations.converters.materializedtable.SqlAlterMaterializedTableModifyDistributionConverter;
+import
org.apache.flink.table.planner.operations.converters.materializedtable.SqlAlterMaterializedTableOptionsConverter;
import
org.apache.flink.table.planner.operations.converters.materializedtable.SqlAlterMaterializedTableRefreshConverter;
import
org.apache.flink.table.planner.operations.converters.materializedtable.SqlAlterMaterializedTableResetConverter;
import
org.apache.flink.table.planner.operations.converters.materializedtable.SqlAlterMaterializedTableResumeConverter;
@@ -148,6 +149,7 @@ public class SqlNodeConverters {
register(new SqlAlterMaterializedTableSchemaDropColumnConverter());
register(new SqlAlterMaterializedTableDropWatermarkConverter());
register(new SqlAlterMaterializedTableModifySchemaConverter());
+ register(new SqlAlterMaterializedTableOptionsConverter());
register(new SqlAlterMaterializedTableRefreshConverter());
register(new SqlAlterMaterializedTableResetConverter());
register(new SqlAlterMaterializedTableResumeConverter());
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/materializedtable/SqlAlterMaterializedTableOptionsConverter.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/materializedtable/SqlAlterMaterializedTableOptionsConverter.java
new file mode 100644
index 00000000000..096d5f2a2a7
--- /dev/null
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/materializedtable/SqlAlterMaterializedTableOptionsConverter.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.planner.operations.converters.materializedtable;
+
+import org.apache.flink.sql.parser.SqlParseUtils;
+import
org.apache.flink.sql.parser.ddl.materializedtable.SqlAlterMaterializedTableOptions;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.catalog.ResolvedCatalogMaterializedTable;
+import org.apache.flink.table.catalog.TableChange;
+import org.apache.flink.table.operations.Operation;
+import
org.apache.flink.table.operations.materializedtable.AlterMaterializedTableChangeOperation;
+
+import java.util.List;
+import java.util.Map;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+/** A converter for {@link SqlAlterMaterializedTableOptions}. */
+public class SqlAlterMaterializedTableOptionsConverter
+ extends
AbstractAlterMaterializedTableConverter<SqlAlterMaterializedTableOptions> {
+
+ @Override
+ protected Operation convertToOperation(
+ SqlAlterMaterializedTableOptions sqlAlterTable,
+ ResolvedCatalogMaterializedTable oldTable,
+ ConvertContext context) {
+ return new AlterMaterializedTableChangeOperation(
+ resolveIdentifier(sqlAlterTable, context),
+ gatherTableChanges(sqlAlterTable, context),
+ oldTable);
+ }
+
+ @Override
+ protected Function<ResolvedCatalogMaterializedTable, List<TableChange>>
gatherTableChanges(
+ SqlAlterMaterializedTableOptions sqlAlterTable, ConvertContext
context) {
+ final Map<String, String> options =
+ SqlParseUtils.extractMap(sqlAlterTable.getPropertyList());
+ if (options.isEmpty()) {
+ throw new ValidationException(
+ EX_MSG_PREFIX + "ALTER MATERIALIZED TABLE SET does not
support empty options.");
+ }
+ final List<TableChange> changes =
+ options.entrySet().stream()
+ .map(e -> TableChange.set(e.getKey(), e.getValue()))
+ .collect(Collectors.toList());
+ return oldTable -> changes;
+ }
+}
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlMaterializedTableNodeToOperationConverterTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlMaterializedTableNodeToOperationConverterTest.java
index 45e952a4b3c..1b458494542 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlMaterializedTableNodeToOperationConverterTest.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlMaterializedTableNodeToOperationConverterTest.java
@@ -481,7 +481,7 @@ class SqlMaterializedTableNodeToOperationConverterTest
AlterMaterializedTableRefreshOperation op =
(AlterMaterializedTableRefreshOperation) operation;
-
assertThat(op.getTableIdentifier().toString()).isEqualTo("`builtin`.`default`.`mtbl1`");
+
assertThat(op.getTableIdentifier()).hasToString("`builtin`.`default`.`mtbl1`");
assertThat(op.getPartitionSpec())
.containsExactly(Map.entry("ds1", "1"), Map.entry("ds2", "2"));
}
@@ -495,7 +495,7 @@ class SqlMaterializedTableNodeToOperationConverterTest
AlterMaterializedTableRefreshOperation op =
(AlterMaterializedTableRefreshOperation) operation;
-
assertThat(op.getTableIdentifier().toString()).isEqualTo("`builtin`.`default`.`mtbl1`");
+
assertThat(op.getTableIdentifier()).hasToString("`builtin`.`default`.`mtbl1`");
assertThat(op.getPartitionSpec()).isEmpty();
}
@@ -523,6 +523,32 @@ class SqlMaterializedTableNodeToOperationConverterTest
.isEqualTo("ALTER MATERIALIZED TABLE builtin.default.mtbl1
RESUME WITH (k1: [v1])");
}
+ @Test
+ void testAlterMaterializedTableSet() {
+ final String sql =
+ "ALTER MATERIALIZED TABLE base_mtbl SET ('format' = 'json2',
'k1' = 'v1', 'k2' = 'v2', 'k2' = 'newV2')";
+ Operation operation = parse(sql);
+
assertThat(operation).isInstanceOf(AlterMaterializedTableChangeOperation.class);
+
+ AlterMaterializedTableChangeOperation op =
+ (AlterMaterializedTableChangeOperation) operation;
+
assertThat(op.getTableIdentifier()).hasToString("`builtin`.`default`.`base_mtbl`");
+ assertThat(op.getTableChanges())
+ .containsExactlyInAnyOrder(
+ TableChange.set("format", "json2"),
+ TableChange.set("k1", "v1"),
+ TableChange.set("k2", "newV2"));
+ assertThat(op.getNewTable().getOptions())
+ .containsOnly(
+ Map.entry("connector", "filesystem"),
+ Map.entry("format", "json2"),
+ Map.entry("k1", "v1"),
+ Map.entry("k2", "newV2"));
+ assertThat(op.asSummaryString())
+ .startsWith("ALTER MATERIALIZED TABLE
builtin.default.base_mtbl\n")
+ .contains(" SET 'format' = 'json2'", " SET 'k1' = 'v1'", "
SET 'k2' = 'newV2'");
+ }
+
@Test
void testAlterMaterializedTableReset() {
final String sql = "ALTER MATERIALIZED TABLE base_mtbl RESET
('format', 'unknown_key')";
@@ -531,7 +557,7 @@ class SqlMaterializedTableNodeToOperationConverterTest
AlterMaterializedTableChangeOperation op =
(AlterMaterializedTableChangeOperation) operation;
-
assertThat(op.getTableIdentifier().toString()).isEqualTo("`builtin`.`default`.`base_mtbl`");
+
assertThat(op.getTableIdentifier()).hasToString("`builtin`.`default`.`base_mtbl`");
assertThat(op.getTableChanges())
.containsExactlyInAnyOrder(
TableChange.reset("format"),
TableChange.reset("unknown_key"));
@@ -679,6 +705,7 @@ class SqlMaterializedTableNodeToOperationConverterTest
list.addAll(alterModifyWithInvalidSchema());
list.addAll(alterQuery());
list.addAll(alterDrop());
+ list.addAll(alterSet());
list.addAll(alterReset());
return list;
}
@@ -1022,6 +1049,19 @@ class SqlMaterializedTableNodeToOperationConverterTest
+ "The column `m_p` is a persisted column.
Dropping of persisted columns is not supported."));
}
+ private static Collection<TestSpec> alterSet() {
+ return List.of(
+ TestSpec.of(
+ "ALTER MATERIALIZED TABLE base_mtbl SET ()",
+ "ALTER MATERIALIZED TABLE SET does not support empty
options."),
+ TestSpec.of(
+ "ALTER MATERIALIZED TABLE unknown_mtbl SET ('format' =
'json2')",
+ "Materialized table `builtin`.`default`.`unknown_mtbl`
doesn't exist."),
+ TestSpec.of(
+ "ALTER MATERIALIZED TABLE t3 SET ('format' = 'json2')",
+ "ALTER MATERIALIZED TABLE for a table is not
allowed"));
+ }
+
private static Collection<TestSpec> alterReset() {
return List.of(
TestSpec.of(
