[FLINK-38355][table][FLIP-546] Support `CREATE OR ALTER MATERIALIZED TABLE` syntax #27199

raminqaf · 2025-11-05T13:54:57Z

What is the purpose of the change

This pull request implements the CREATE OR ALTER MATERIALIZED TABLE syntax as proposed in FLIP-546. This new command provides an idempotent way to manage materialized tables, enabling declarative deployment patterns for CI/CD pipelines and infrastructure-as-code workflows.

The command intelligently routes to either CREATE or ALTER logic:

If the materialized table does not exist, it creates a new table (behaves like CREATE MATERIALIZED TABLE)
If the materialized table already exists, it modifies the query definition (behaves like ALTER MATERIALIZED TABLE AS)

This eliminates the need for complex DROP-IF-EXISTS patterns and makes materialized table management more robust and predictable in automated deployment scenarios.

Brief change log

Added CREATE OR ALTER MATERIALIZED TABLE SQL syntax to the Flink SQL parser
Extended SqlCreateOrAlterMaterializedTable to handle both create and alter operations based on table existence
Refactored materialized table parsing and building logic into SqlCreateOrAlterMaterializedTableConverter for better encapsulation
Added getOriginalQuery() method to CatalogMaterializedTable interface (aligning with CatalogView interface)
Added comprehensive tests for CREATE OR ALTER behavior in both create and alter scenarios
Updated documentation to include CREATE OR ALTER MATERIALIZED TABLE syntax and examples

Verifying this change

This change added tests and can be verified as follows:

Added unit tests in MaterializedTableStatementParserTest to verify SQL parsing of CREATE OR ALTER syntax
Added integration tests in SqlMaterializedTableNodeToOperationConverterTest:
- testCreateOrAlterMaterializedTable() - verifies CREATE behavior when table doesn't exist
- testCreateOrAlterMaterializedTableForExistingTable() - verifies ALTER behavior when table exists with schema evolution
Extended integration tests in SqlGatewayRestEndpointMaterializedTableITCase to validate end-to-end behavior via REST API
Verified that existing CREATE and ALTER tests continue to pass, ensuring backward compatibility
Manually verified idempotent behavior by running the same CREATE OR ALTER statement multiple times

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): no
The public API, i.e., is any changed class annotated with @Public(Evolving): yes - CatalogMaterializedTable interface extended with getOriginalQuery() method
The serializers: no
The runtime per-record code paths (performance sensitive): no
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
The S3 file system connector: no

Documentation

Does this pull request introduce a new feature? yes
If yes, how is the feature documented? docs - Added new section in docs/content/docs/dev/table/materialized-table/statements.md explaining CREATE OR ALTER syntax, behavior, examples, and use cases

flinkbot · 2025-11-05T14:02:46Z

CI report:

3033271 UNKNOWN
2cd5eae Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

...l-parser/src/test/java/org/apache/flink/sql/parser/MaterializedTableStatementParserTest.java

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

...che/flink/table/planner/operations/converters/SqlAlterMaterializedTableAsQueryConverter.java

davidradl · 2025-11-06T11:31:13Z

.../apache/flink/table/planner/operations/SqlMaterializedTableNodeToOperationConverterTest.java

                .isEqualTo(newTable.getUnresolvedSchema().getPrimaryKey());
        assertThat(oldTable.getUnresolvedSchema().getWatermarkSpecs())
                .isEqualTo(newTable.getUnresolvedSchema().getWatermarkSpecs());
-        assertThat(oldTable.getDefinitionQuery()).isNotEqualTo(newTable.getDefinitionQuery());


I am curious what you think, we have removed DefinitionQuery, but we still have DefinitionFreshness. Should we rename DefinitionFreshness so it does not include the word Definition - maybe change just to freshness to simplify.

It's a valid point but we have deprecated the getFreshness() method in CatalogMaterializedTable interface.

https://github.com/apache/flink/blob/master/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/CatalogMaterializedTable.java#L123-L141

...s/src/test/java/org/apache/flink/table/file/testutils/catalog/TestFileSystemCatalogTest.java

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

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

...l-parser/src/test/java/org/apache/flink/sql/parser/MaterializedTableStatementParserTest.java

.../apache/flink/table/operations/materializedtable/AlterMaterializedTableAsQueryOperation.java

...le-api-java/src/test/java/org/apache/flink/table/catalog/CatalogBaseTableResolutionTest.java

snuyanzin · 2025-11-13T10:58:57Z

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

+        writer.keyword("MATERIALIZED TABLE");
+        getTableName().unparse(writer, leftPrec, rightPrec);
+
+        if (!getColumnList().isEmpty()
+                || !getTableConstraints().isEmpty()
+                || getWatermark().isPresent()) {
+            SqlUnparseUtils.unparseTableSchema(
+                    writer,
+                    leftPrec,
+                    rightPrec,
+                    getColumnList(),
+                    getTableConstraints(),
+                    getWatermark().orElse(null));
+        }
+
+        getComment()
+                .ifPresent(
+                        comment -> {
+                            writer.newlineAndIndent();
+                            writer.keyword("COMMENT");
+                            comment.unparse(writer, leftPrec, rightPrec);
+                        });
+
+        if (getDistribution() != null) {
+            writer.newlineAndIndent();
+            getDistribution().unparse(writer, leftPrec, rightPrec);
+        }
+
+        if (!getPartitionKeyList().isEmpty()) {
+            writer.newlineAndIndent();
+            writer.keyword("PARTITIONED BY");
+            SqlWriter.Frame partitionedByFrame = writer.startList("(", ")");
+            getPartitionKeyList().unparse(writer, leftPrec, rightPrec);
+            writer.endList(partitionedByFrame);
+        }
+
+        if (!getPropertyList().isEmpty()) {
+            writer.newlineAndIndent();
+            writer.keyword("WITH");
+            SqlWriter.Frame withFrame = writer.startList("(", ")");
+            for (SqlNode property : getPropertyList()) {
+                SqlUnparseUtils.printIndent(writer);
+                property.unparse(writer, leftPrec, rightPrec);
+            }
+            writer.newlineAndIndent();
+            writer.endList(withFrame);
+        }
+
+        if (getFreshness() != null) {
+            writer.newlineAndIndent();
+            writer.keyword("FRESHNESS");
+            writer.keyword("=");
+            getFreshness().unparse(writer, leftPrec, rightPrec);
+        }
+
+        if (getRefreshMode() != null) {
+            writer.newlineAndIndent();
+            writer.keyword("REFRESH_MODE");
+            writer.keyword("=");
+            writer.keyword(getRefreshMode().name());
+        }
+
+        writer.newlineAndIndent();
+        writer.keyword("AS");
+        writer.newlineAndIndent();
+        getAsQuery().unparse(writer, leftPrec, rightPrec);
+    }


I guess it is duplicating what we have in SqlCreateMaterializedTable
I wonder if we can extract this part into a separate method and reuse for both?

I have removed the code from the SqlCreateMaterializedTable and only the child class SqlCreateOrAlterMaterializedTable is implementing the logic now. Or do we need both classes to implement it?

Well, it means we need to reimplement it again if want to support something like ALTER MATERIALIZED TABLE ...
so yes, better to reuse

snuyanzin · 2025-11-13T11:00:25Z

...ink-table-api-java/src/test/java/org/apache/flink/table/api/internal/ShowCreateUtilTest.java

+                        "SELECT id, name FROM tbl_a",
                        "SELECT id, name FROM tbl_a"),


do we have at least one test where original and expanded are different?
If not then. probably should add it

snuyanzin · 2025-11-13T11:13:24Z

...he/flink/table/planner/operations/converters/SqlCreateOrAlterMaterializedTableConverter.java

+                ? handleAlter(sqlCreateOrAlterMaterializedTable, context)
+                : handleCreate(sqlCreateOrAlterMaterializedTable, context, identifier);


can we instead just redirect call to either SqlCreateMaterializedTableConverter or SqlAlterMaterializedTableConverter?

snuyanzin · 2025-11-13T15:26:49Z

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

+    public static final SqlSpecialOperator CREATE_OR_ALTER_OPERATOR =
+            new SqlSpecialOperator("CREATE OR ALTER MATERIALIZED TABLE", SqlKind.OTHER_DDL);
+
+    private final boolean isOrAlter;


do we really need this field if we have different operators?

snuyanzin · 2025-11-13T15:29:20Z

...ink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlCreateMaterializedTable.java

    @Override
    public SqlOperator getOperator() {
-        return OPERATOR;
+        return CREATE_OPERATOR;
    }


I guess if you drop this method override it will return the operator which is passed to the constructor and this is what we want, right?

snuyanzin · 2025-11-13T15:29:55Z

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

+    @Override
+    public SqlOperator getOperator() {
+        return isOrAlter ? CREATE_OR_ALTER_OPERATOR : CREATE_OPERATOR;
+    }


probably no need for this method if already passed operator to constructor

snuyanzin · 2025-11-13T15:30:49Z

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

+    @Override
+    public void unparse(SqlWriter writer, int leftPrec, int rightPrec) {
+        writer.keyword("CREATE");
+        if (isOrAlter) {


can we use

Suggested change

if (isOrAlter) {

if (getOperator() == CREATE_OR_ALTER_OPERATOR) {

?
and then no need for isOrAlter

snuyanzin · 2025-11-13T15:31:46Z

...l-parser/src/test/java/org/apache/flink/sql/parser/MaterializedTableStatementParserTest.java

+    private static final String CREATE_COMMAND = "CREATE ";
+    private static final String CREATE_OR_ALTER_COMMAND = "CREATE OR ALTER ";


Suggested change

private static final String CREATE_COMMAND = "CREATE ";

private static final String CREATE_OR_ALTER_COMMAND = "CREATE OR ALTER ";

private static final String CREATE_OPERATION = "CREATE ";

private static final String CREATE_OR_ALTER_OPERATION = "CREATE OR ALTER ";

snuyanzin · 2025-11-13T15:37:44Z

...he/flink/table/planner/operations/converters/SqlCreateOrAlterMaterializedTableConverter.java

+        final ObjectIdentifier identifier =
+                this.getIdentifier(sqlCreateOrAlterMaterializedTable, context);
+        final ResolvedCatalogMaterializedTable oldTable =
+                getExistingResolvedMaterializedTable(context, identifier);


Right before calling this method we already have derived these 2 values
an we reuse them instead doing same twice?

snuyanzin · 2025-11-13T15:40:02Z

...he/flink/table/planner/operations/converters/SqlCreateOrAlterMaterializedTableConverter.java

+                        oldTable, sqlCreateOrAlterMaterializedTable, context);
+
+        List<MaterializedTableChange> tableChanges =
+                buildTableChanges(sqlCreateOrAlterMaterializedTable, oldTable, context);


Suggested change

buildTableChanges(sqlCreateOrAlterMaterializedTable, oldTable, context);

buildTableChanges(sqlCreateOrAlterMaterializedTable, oldMaterializedTable, context);

here and in other places oldTable -> oldMaterializedTable

snuyanzin · 2025-11-13T15:40:56Z

...he/flink/table/planner/operations/converters/SqlCreateOrAlterMaterializedTableConverter.java

+        final MergeContext mergeContext =
+                this.getMergeContext(sqlCreateOrAlterMaterializedTable, context);
+
+        // Extract new columns


probably no need for this comment

snuyanzin · 2025-11-13T15:41:20Z

...he/flink/table/planner/operations/converters/SqlCreateOrAlterMaterializedTableConverter.java

+                        .refreshMode(refreshMode)
+                        .refreshStatus(RefreshStatus.INITIALIZING);
+
+        // Preserve refresh handler from old table


Suggested change

// Preserve refresh handler from old table

// Preserve refresh handler from old materialized table

snuyanzin · 2025-11-13T15:51:56Z

...he/flink/table/planner/operations/converters/SqlCreateOrAlterMaterializedTableConverter.java

+    private boolean tableExists(ConvertContext context, ObjectIdentifier identifier) {
+        return context.getCatalogManager().getTable(identifier).isPresent();
+    }


not sure we need this method at all given the fact that there is getExistingResolvedMaterializedTable
moreover tableExists doesn't check table type

snuyanzin · 2025-11-13T15:53:41Z

...he/flink/table/planner/operations/converters/SqlCreateOrAlterMaterializedTableConverter.java

+        return context.getCatalogManager().getTable(identifier).isPresent();
+    }
+
+    private CatalogMaterializedTable buildNewCatalogMaterializedTableFromOldTable(


ideally this method should be in alterMaterializedTable converter
may be we can move it in future when that converter becomes more mature

raminqaf changed the title ~~[FLINK-38355] Support CreateOrAlter MATERIALIZED TABLE~~ [FLINK-38355][table][FLIP-546] Support CREATE OR ALTER MATERIALIZED TABLE Nov 5, 2025

raminqaf changed the title ~~[FLINK-38355][table][FLIP-546] Support CREATE OR ALTER MATERIALIZED TABLE~~ [FLINK-38355][table][FLIP-546] Support CREATE OR ALTER MATERIALIZED TABLE syntax Nov 5, 2025

raminqaf force-pushed the FLINK-38355 branch from c656e71 to 578e536 Compare November 5, 2025 13:59

snuyanzin reviewed Nov 5, 2025

View reviewed changes

...l-parser/src/test/java/org/apache/flink/sql/parser/MaterializedTableStatementParserTest.java Outdated Show resolved Hide resolved

davidradl reviewed Nov 6, 2025

View reviewed changes

docs/content/docs/dev/table/materialized-table/statements.md Outdated Show resolved Hide resolved

davidradl reviewed Nov 6, 2025

View reviewed changes

...che/flink/table/planner/operations/converters/SqlAlterMaterializedTableAsQueryConverter.java Outdated Show resolved Hide resolved

davidradl reviewed Nov 6, 2025

View reviewed changes

...s/src/test/java/org/apache/flink/table/file/testutils/catalog/TestFileSystemCatalogTest.java Outdated Show resolved Hide resolved

github-actions bot added the community-reviewed PR has been reviewed by the community. label Nov 6, 2025

raminqaf force-pushed the FLINK-38355 branch from 26e232d to 0ab4a06 Compare November 6, 2025 16:12

snuyanzin reviewed Nov 6, 2025

View reviewed changes

flink-table/flink-sql-parser/src/main/codegen/data/Parser.tdd Outdated Show resolved Hide resolved

snuyanzin reviewed Nov 6, 2025

View reviewed changes

...-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlCreateOrAlterMaterializedTable.java Outdated Show resolved Hide resolved

snuyanzin reviewed Nov 6, 2025

View reviewed changes

...l-parser/src/test/java/org/apache/flink/sql/parser/MaterializedTableStatementParserTest.java Outdated Show resolved Hide resolved

snuyanzin reviewed Nov 6, 2025

View reviewed changes

.../apache/flink/table/operations/materializedtable/AlterMaterializedTableAsQueryOperation.java Outdated Show resolved Hide resolved

snuyanzin reviewed Nov 6, 2025

View reviewed changes

...le-api-java/src/test/java/org/apache/flink/table/catalog/CatalogBaseTableResolutionTest.java Outdated Show resolved Hide resolved

snuyanzin reviewed Nov 6, 2025

View reviewed changes

...le-api-java/src/test/java/org/apache/flink/table/catalog/CatalogBaseTableResolutionTest.java Outdated Show resolved Hide resolved

raminqaf force-pushed the FLINK-38355 branch 5 times, most recently from b08780d to ba9286e Compare November 12, 2025 10:17

raminqaf added 7 commits November 13, 2025 11:06

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

3e83e4d

[FLINK-38355] apply optional freshness to CoA

bfd9011

[FLINK-38355] Address nit feedbacks and renames

75b7868

[FLINK-38355] Inline docs similar to parser definition

8a60f93

[FLINK-38355] Revert deletion of SqlCreateMaterializedTable

9532910

[FLINK-38355] Use AbstractCreateMaterializedTableConverter

13dc166

[FLINK-38355] Resolve merge conflicts and update tests

2cd5eae

raminqaf force-pushed the FLINK-38355 branch from ba9286e to 2cd5eae Compare November 13, 2025 10:53

snuyanzin reviewed Nov 13, 2025

View reviewed changes

		? handleAlter(sqlCreateOrAlterMaterializedTable, context)
		: handleCreate(sqlCreateOrAlterMaterializedTable, context, identifier);

	if (isOrAlter) {
	if (getOperator() == CREATE_OR_ALTER_OPERATOR) {

		private static final String CREATE_COMMAND = "CREATE ";
		private static final String CREATE_OR_ALTER_COMMAND = "CREATE OR ALTER ";

	buildTableChanges(sqlCreateOrAlterMaterializedTable, oldTable, context);
	buildTableChanges(sqlCreateOrAlterMaterializedTable, oldMaterializedTable, context);

	// Preserve refresh handler from old table
	// Preserve refresh handler from old materialized table

[FLINK-38355][table][FLIP-546] Support CREATE OR ALTER MATERIALIZED TABLE syntax #27199

Are you sure you want to change the base?

[FLINK-38355][table][FLIP-546] Support CREATE OR ALTER MATERIALIZED TABLE syntax #27199

Conversation

raminqaf commented Nov 5, 2025

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

Uh oh!

flinkbot commented Nov 5, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

raminqaf Nov 13, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

[FLINK-38355][table][FLIP-546] Support `CREATE OR ALTER MATERIALIZED TABLE` syntax #27199

[FLINK-38355][table][FLIP-546] Support `CREATE OR ALTER MATERIALIZED TABLE` syntax #27199

flinkbot commented Nov 5, 2025 •

edited

Loading

raminqaf Nov 13, 2025 •

edited

Loading