[FLINK-38993][table] Move materialized table query change validation logic from converter to operation #27488

snuyanzin · 2026-01-28T22:10:18Z

What is the purpose of the change

The issue is that right now validation logic (the one which is not related to common syntax) is hard coded in converter.
From another side converters do not allow to extend them.
The idea is to move validation to operation. Operations could be custom and could contain slightly relaxed validation logic.

Brief change log

Materialized table operations

Verifying this change

Mostly existed tests
however some new test also added

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): ( no)
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? ( no)
If yes, how is the feature documented? (not applicable )

flinkbot · 2026-01-28T22:18:48Z

CI report:

2565fcf Azure: FAILURE
fc011fa Azure: PENDING
53f52db UNKNOWN

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

liuyongvs

LGTM +1

snuyanzin · 2026-01-29T07:30:24Z

.../java/org/apache/flink/table/gateway/service/materializedtable/MaterializedTableManager.java

-            AlterMaterializedTableChangeOperation alterMaterializedTableChangeOperation =
-                    new AlterMaterializedTableChangeOperation(
-                            tableIdentifier,
-                            op.getTableChanges(),
-                            op.getCatalogMaterializedTable());
            return operationExecutor.callExecutableOperation(
-                    handle, alterMaterializedTableChangeOperation);


anyway here and in other places it is just a copy as AlterMaterializedTableChangeOperation since AlterMaterializedTableAsQueryOperation#execute fails with UnsupportedOperationException
not sure why it is done like that initially. Do not change the logic, just make code a bit more compact here

snuyanzin · 2026-01-29T07:32:21Z

.../java/org/apache/flink/table/gateway/service/materializedtable/MaterializedTableManager.java

-                                        + "Unsupported table change detected: %s. ",
-                                op.getTableIdentifier(), tableChange));
-            }
-        }


since now AlterMaterializedTableChangeOperation contains old table and a list of changes, there is no need to generate new list of changes to calculate rollbacked table

snuyanzin · 2026-01-29T07:38:48Z

...g/apache/flink/table/operations/materializedtable/AlterMaterializedTableChangeOperation.java

+        }
+
+        private void applyTableChanges(List<TableChange> tableChanges) {
+            isQueryChange = tableChanges.stream().anyMatch(t -> t instanceof ModifyDefinitionQuery);


depending on whether it is a query change or not the validation is slightly different

snuyanzin · 2026-01-29T07:39:29Z

flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/TableChange.java

     */
-    static ModifyDefinitionQuery modifyDefinitionQuery(String definitionQuery) {
-        return new ModifyDefinitionQuery(definitionQuery);
+    static ModifyDefinitionQuery modifyDefinitionQuery(


looks like it was missed when original query was added

…logic from converter to operation

snuyanzin · 2026-01-29T07:53:15Z

...table-planner/src/main/java/org/apache/flink/table/planner/utils/MaterializedTableUtils.java


+    // Used to build changes introduced by changed query like
+    // ALTER MATERIALIZED TABLE ... AS ...
+    public static List<TableChange> buildSchemaTableChanges(


generation of table change based on query change
In case of table schema changes the generation is happening at SchemaConverter

The reason why it is not used here is that SchemaConverter works with AST objects

AHeise

First pass. Looks very good already. My main question is if we can restructure the change application to avoid the 100 if(change instanceof ...) checks.

AHeise · 2026-01-29T09:38:15Z

...g/apache/flink/table/operations/materializedtable/AlterMaterializedTableChangeOperation.java

+                        .comment(oldTable.getComment())
+                        .partitionKeys(oldTable.getPartitionKeys())
+                        .options(oldTable.getOptions())


Does that mean that we currently don't support changing these fields?
(I'm fine if this is also the current behavior)

The only way how partition might be changed is

ALTER MATERIALIZED TABLE [catalog_name.][db_name.]table_name REFRESH [PARTITION (key1=val1, key2=val2, ...)]

and this is handled with completely different operation/converter AlterMaterializedTableRefreshOperation, SqlAlterMaterializedTableRefreshConverter and I don't see any specific validations there, just conversion to operation

other's change is impossible for no, yes

...g/apache/flink/table/operations/materializedtable/AlterMaterializedTableChangeOperation.java

AHeise · 2026-01-29T10:50:48Z

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

                        "(\n"
                                + "  `m` STRING METADATA VIRTUAL,\n"
-                                + "  `calc` AS 'a' || 'b',\n"
+                                + "  `calc` AS ['a' || 'b'],\n"


Why are these changes necessary?

together with validation resolving moved to operation level
square brackets mean unresolved expression

flink/flink-table/flink-table-common/src/main/java/org/apache/flink/table/expressions/SqlCallExpression.java

Lines 41 to 42 in ae45832

// indicates that this is an unresolved expression consistent with unresolved data types

private static final String FORMAT = "[%s]";

AHeise · 2026-01-29T10:52:42Z

...e-planner/src/test/java/org/apache/flink/table/planner/utils/MaterializedTableUtilsTest.java

+                        List.of(
+                                TableChange.add(physical("b", DataTypes.INT())),
+                                TableChange.dropColumn("a"))),
+                TestSpec.of(


Do we also need tests for column type changes (e.g. widening)? If that's not supported yet, ignore this comment.

added a test

AHeise · 2026-01-29T10:58:01Z

...g/apache/flink/table/operations/materializedtable/AlterMaterializedTableChangeOperation.java

+                handleColumns(tableChange);
+                handleDistribution(tableChange);
+                handleWatermark(tableChange);
+                handleUniqueConstraint(tableChange);
+                handleRefreshStatus(tableChange);
+                handleRefreshHandler(tableChange);
+                handleModifyDefinitionQuery(tableChange);


We could return true to short cut the handles on the first successful invocation.
Alterantively, we could build nested Handler classes that provide a list of supported change classes. Then we can have an index from type -> handler.
Going further we could have just 1 handler per change class, where we use a generic type to capture the handled change and automatically extract the change class to builld the inex.

WDYT?

probably I forgot to fix if condition at the very beginning
fixed that
thanks 👍

AHeise · 2026-01-29T10:58:46Z

...g/apache/flink/table/operations/materializedtable/AlterMaterializedTableChangeOperation.java

+                primaryKeyName = null;
+                primaryKeyColumns = null;


Do you need to verify if the drop constraint is actually for that particular primary key? Or is this always the case?

Do we need to sequence the changes somehow? If I change the PK from pk1 to pk2, do I get a DropConstraint and an AddUniqueConstraint. If so, we need to drop first before readding, right?
Can this also happen for other changes?
Since changes is a list, do we already have the sequence? If so, please document that.

For now it is impossible to have several primary keys for one [materialized ]table
for that reason while dropping there is no name, just this command

ALTER MATERIALIZED TABLE DROP PRIMARY KEY;

Do we need to sequence the changes somehow? If I change the PK from pk1 to pk2, do I get a DropConstraint and an AddUniqueConstraint. If so, we need to drop first before readding, right?
Can this also happen for other changes?
Since changes is a list, do we already have the sequence? If so, please document that.

may be in future we will have such cases
right now I don't know how I could tell to Flink just with changing query operation (what I should put in SQL...)

Anyway I will put the comment about ordering just in case

liuyongvs approved these changes Jan 29, 2026

View reviewed changes

snuyanzin force-pushed the flink38993 branch from 88cf755 to 29f1248 Compare January 29, 2026 07:28

snuyanzin commented Jan 29, 2026

View reviewed changes

[FLINK-38993][table] Move materialized table query change validation …

2565fcf

…logic from converter to operation

snuyanzin force-pushed the flink38993 branch from 29f1248 to 2565fcf Compare January 29, 2026 07:43

snuyanzin commented Jan 29, 2026

View reviewed changes

github-actions bot added the community-reviewed PR has been reviewed by the community. label Jan 29, 2026

AHeise reviewed Jan 29, 2026

View reviewed changes

snuyanzin added 2 commits January 29, 2026 13:56

Address feedback

fc011fa

Address feedback

53f52db

	// indicates that this is an unresolved expression consistent with unresolved data types
	private static final String FORMAT = "[%s]";

[FLINK-38993][table] Move materialized table query change validation logic from converter to operation #27488

Are you sure you want to change the base?

[FLINK-38993][table] Move materialized table query change validation logic from converter to operation #27488

Conversation

snuyanzin commented Jan 28, 2026

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 Jan 28, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

liuyongvs left a comment

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!

snuyanzin Jan 29, 2026 • 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!

AHeise left a comment

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!

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!

snuyanzin Jan 29, 2026 • 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!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

flinkbot commented Jan 28, 2026 •

edited

Loading

snuyanzin Jan 29, 2026 •

edited

Loading

snuyanzin Jan 29, 2026 •

edited

Loading