[FLINK-38856][table] Add support for `ALTER MATERIALIZED TABLE DROP` schema component #27381

snuyanzin · 2026-01-04T16:42:55Z

What is the purpose of the change

The PR implements ALTER MATERIALIZED TABLE ... DROP part of FLIP-550 like

-- drop a column (only drop of non persisted is allowed)
ALTER MATERIALIZED TABLE MyTable DROP col1;
 
-- drop several columns
ALTER MATERIALIZED TABLE MyTable DROP (col1, col2, col3);
 
-- drop a primary key
ALTER MATERIALIZED TABLE MyTable DROP PRIMARY KEY;

-- drop a constraint by name
ALTER MATERIALIZED TABLE MyTable DROP CONSTRAINT constraint_name;

-- drop a watermark
ALTER MATERIALIZED TABLE MyTable DROP WATERMARK;

Besides that it also refactors a bit SqlNodeToOperationConversion by extraction of function and analyze table functionality into dedicated converters.

Verifying this change

There are new tests for failed and success cases, also existing tests

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? (yes )
If yes, how is the feature documented? (will be done in a separate PR)

…rs` package

…ql objects

…hema component

…zedtable` package

…ction`, `DropFunction` to dedicated converters

snuyanzin · 2026-01-04T16:44:55Z

...er/src/main/java/org/apache/flink/table/planner/operations/SqlNodeToOperationConversion.java

@@ -315,182 +264,56 @@
        } else if (validated instanceof SqlCompileAndExecutePlan) {
            return Optional.of(
                    converter.convertCompileAndExecutePlan((SqlCompileAndExecutePlan) validated));
-        } else if (validated instanceof SqlAnalyzeTable) {
-            return Optional.of(converter.convertAnalyzeTable((SqlAnalyzeTable) validated));


all of these moved into separate converter classes
it allowed to make this class less than 1k lines

flinkbot · 2026-01-04T16:48:31Z

CI report:

2f765e1 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

davidradl · 2026-01-05T10:00:10Z

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

+        if (!identifier.isSimple()) {
+            throw new UnsupportedOperationException(
+                    String.format(
+                            "%sAlter nested row type %s is not supported yet.",


nit: from what I can see the exMsgPrefix comes from the TableKind enum. So would read something like
MATERIALIZED TABLEAlter nested row ....

I wonder if it would read better as Alter %s nested row ...

nit: from what I can see the exMsgPrefix comes from the TableKind enum. So would read something like
MATERIALIZED_TABLEAlter nested row ....

This is not true, it comes from https://github.com/snuyanzin/flink/blob/44838279dd0575d02c308b86ce752a87db86a9ee/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/materializedtable/AbstractAlterMaterializedTableConverter.java#L41-L42

or
https://github.com/snuyanzin/flink/blob/44838279dd0575d02c308b86ce752a87db86a9ee/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/table/AbstractAlterTableConverter.java#L45

TableKind enum is the second arg

Ok thanks - my bad. I saw

flink/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SchemaConverter.java

Line 82 in 94c6549

this.exMsgPrefix = String.format(ERROR_TEMPLATE, tableKindStr.toUpperCase(Locale.ROOT));

assuming it was picking it up from there. Strange it uses this name of variable in such different ways.

davidradl · 2026-01-05T10:02:46Z

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

+        if (oldTable.getResolvedSchema().getWatermarkSpecs().isEmpty()) {
+            throw new ValidationException(
+                    String.format(
+                            "%sThe current %s does not define any watermark strategy.",


similar comment . It will read strangely without a space before the first The

as mentioned above, there always will be \n

davidradl · 2026-01-05T10:04:40Z

.../src/test/java/org/apache/flink/table/planner/operations/SqlDdlToOperationConverterTest.java

                .isInstanceOf(ValidationException.class)
                .hasMessageContaining(
-                        "The base table does not define a primary key constraint named 'ct2'. Available constraint name: ['ct1'].");
+                        "The current table does not define a primary key constraint named 'ct2'. Available constraint name: ['ct1'].");


I suggest including the the message prefix in these tests - so we can see it reads well.

added
ideally it should be parameterized and refactored, however this PR is already large, so probably in a separate one

davidradl · 2026-01-05T10:06:47Z

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

        }
        |
-        <DROP> <DISTRIBUTION> {
+        <DROP>


I like this rationalisation of the grammar. Can we update the docs to match it?

I'm going to submit a separate PR for docs which should cover FLIP-550 (CREATE, ALTER, DROP)

FLINK-38861

snuyanzin · 2026-01-05T10:25:55Z

.../src/test/java/org/apache/flink/table/planner/operations/SqlDdlToOperationConverterTest.java

        assertThatThrownBy(() -> parse("alter table tb1 drop primary key"))
                .isInstanceOf(ValidationException.class)
-                .hasMessageContaining("The base table does not define any primary key.");
+                .hasMessageContaining("The current table does not define any primary key.");


Following same approach as suggested at #27302 (comment)

twalthr

Thanks for all these massive refactorings. Very valuable!

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

...r/src/main/java/org/apache/flink/table/planner/operations/converters/MergeTableLikeUtil.java

twalthr · 2026-01-05T14:53:38Z

...main/java/org/apache/flink/table/planner/operations/converters/SqlDropFunctionConverter.java

+        UnresolvedIdentifier unresolvedIdentifier =
+                UnresolvedIdentifier.of(sqlDropFunction.getFullName());
+        if (sqlDropFunction.isSystemFunction()) {
+            return new DropTempSystemFunctionOperation(


nit: Interesting that we don't check for isTemporary here. I hope the parser catches this.

looks like there is nothing in parser for that...
Test shows it fails in FunctionCatalog where there is a validation

flink/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/FunctionCatalog.java

Lines 134 to 146 in b597afa

public boolean dropTemporarySystemFunction(String name, boolean ignoreIfNotExist) {

final String normalizedName = FunctionIdentifier.normalizeName(name);

final CatalogFunction function = tempSystemFunctions.remove(normalizedName);

if (function == null && !ignoreIfNotExist) {

throw new ValidationException(

String.format(

"Could not drop temporary system function. A function named '%s' doesn't exist.",

name));

}

unregisterFunctionJarResources(function);

return function != null;

.../src/test/java/org/apache/flink/table/planner/operations/SqlDdlToOperationConverterTest.java

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

snuyanzin · 2026-01-05T17:41:34Z

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

+            throw new UnsupportedOperationException(
+                    String.format(
+                            "%sAltering the nested row type `%s` is not supported yet.",
+                            exMsgPrefix, String.join("`.`", identifier.names)));


Ideally quoting requires dialect which could be derived with help of planner.
In order to not make it complicated continue without this since this is only a message for Exception

snuyanzin · 2026-01-06T07:40:51Z

@flinkbot run azure

snuyanzin added 5 commits January 4, 2026 16:11

[hotfix] Move MergeTableAsUtil and SchemaBuilderUtil to `converte…

8ec464b

…rs` package

[FLINK-38856][table] Generalize ALTER TABLE ... DROP converters and s…

c1ee19f

…ql objects

[FLINK-38856][table] Add support for ALTER MATERIALIZED TABLE DROP sc…

0437f1d

…hema component

[FLINK-38856][table] Move materialized table converters to `materiali…

0925bd0

…zedtable` package

[FLINK-38856][table] Move AnalyzeTable, CreateFunction, `AlterFun…

4483827

…ction`, `DropFunction` to dedicated converters

snuyanzin commented Jan 4, 2026

View reviewed changes

davidradl reviewed Jan 5, 2026

View reviewed changes

snuyanzin commented Jan 5, 2026

View reviewed changes

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

twalthr reviewed Jan 5, 2026

View reviewed changes

Address feedback

9d335a5

snuyanzin commented Jan 5, 2026

View reviewed changes

Address feedback

2f765e1

twalthr approved these changes Jan 6, 2026

View reviewed changes

snuyanzin closed this in 9215bcd Jan 6, 2026

	public boolean dropTemporarySystemFunction(String name, boolean ignoreIfNotExist) {
	final String normalizedName = FunctionIdentifier.normalizeName(name);
	final CatalogFunction function = tempSystemFunctions.remove(normalizedName);

	if (function == null && !ignoreIfNotExist) {
	throw new ValidationException(
	String.format(
	"Could not drop temporary system function. A function named '%s' doesn't exist.",
	name));
	}
	unregisterFunctionJarResources(function);

	return function != null;

[FLINK-38856][table] Add support for ALTER MATERIALIZED TABLE DROP schema component #27381

[FLINK-38856][table] Add support for ALTER MATERIALIZED TABLE DROP schema component #27381

Uh oh!

Conversation

snuyanzin commented Jan 4, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What is the purpose of the change

Verifying this change

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

Documentation

Uh oh!

Choose a reason for hiding this comment

Uh oh!

flinkbot commented Jan 4, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

davidradl Jan 5, 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!

snuyanzin Jan 5, 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!

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!

twalthr left a comment

Choose a reason for hiding this comment

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!

Choose a reason for hiding this comment

Uh oh!

snuyanzin commented Jan 6, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

[FLINK-38856][table] Add support for `ALTER MATERIALIZED TABLE DROP` schema component #27381

[FLINK-38856][table] Add support for `ALTER MATERIALIZED TABLE DROP` schema component #27381

snuyanzin commented Jan 4, 2026 •

edited

Loading

flinkbot commented Jan 4, 2026 •

edited

Loading

davidradl Jan 5, 2026 •

edited

Loading

snuyanzin Jan 5, 2026 •

edited

Loading