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
commit 992062e0e27b418c41b8003c83c787751ece7811 Author: Sergey Nuyanzin <[email protected]> AuthorDate: Tue Mar 24 11:08:04 2026 +0100 [FLINK-39284][table] Make `CREATE OR ALTER MATERIALIZED TABLE` respect column comment change This closes #27807. --- .../operations/ddl/AlterTableChangeOperation.java | 4 ++- .../planner/utils/MaterializedTableUtils.java | 9 ++++++ ...reateOrAlterMaterializedTableConverterTest.java | 35 ++++++++++++++++++++++ 3 files changed, 47 insertions(+), 1 deletion(-) diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/AlterTableChangeOperation.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/AlterTableChangeOperation.java index fbd0b194d0b..d185ab4a385 100644 --- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/AlterTableChangeOperation.java +++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/AlterTableChangeOperation.java @@ -97,7 +97,9 @@ public class AlterTableChangeOperation extends AlterTableOperation { return String.format( " MODIFY %s COMMENT '%s'", EncodingUtils.escapeIdentifier(modifyColumnComment.getNewColumn().getName()), - modifyColumnComment.getNewComment()); + modifyColumnComment.getNewComment() == null + ? "" + : modifyColumnComment.getNewComment()); } else if (tableChange instanceof TableChange.ModifyPhysicalColumnType) { TableChange.ModifyPhysicalColumnType modifyPhysicalColumnType = (TableChange.ModifyPhysicalColumnType) tableChange; diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/utils/MaterializedTableUtils.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/utils/MaterializedTableUtils.java index ea578a2744a..8cc5ff524e8 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/utils/MaterializedTableUtils.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/utils/MaterializedTableUtils.java @@ -45,6 +45,7 @@ import org.apache.calcite.sql.SqlIntervalLiteral.IntervalValue; import org.apache.calcite.sql.SqlNode; import org.apache.calcite.sql.SqlNodeList; import org.apache.calcite.sql.type.SqlTypeFamily; +import org.apache.commons.lang3.StringUtils; import java.util.ArrayList; import java.util.HashMap; @@ -311,6 +312,14 @@ public class MaterializedTableUtils { + "Column mismatch at position %d: Original column is [%s], but new column is [%s].", i + 1, oldColumn, newColumn)); } + final String oldComment = oldColumn.getComment().orElse(null); + final String newComment = newColumn.getComment().orElse(null); + + if (StringUtils.isEmpty(oldComment) != StringUtils.isEmpty(newComment) + || StringUtils.isNotEmpty(oldComment) + && !Objects.equals(oldComment, newComment)) { + columnChanges.add(TableChange.modifyColumnComment(oldColumn, newComment)); + } } } diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlNodeToOperationSqlCreateOrAlterMaterializedTableConverterTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlNodeToOperationSqlCreateOrAlterMaterializedTableConverterTest.java index 499aca4db47..8a56e271d5f 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlNodeToOperationSqlCreateOrAlterMaterializedTableConverterTest.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlNodeToOperationSqlCreateOrAlterMaterializedTableConverterTest.java @@ -467,6 +467,41 @@ class SqlNodeToOperationSqlCreateOrAlterMaterializedTableConverterTest + " DROP WATERMARK"); } + @Test + void testCreateOrAlterMaterializedTableWithCommentChange() + throws TableAlreadyExistException, DatabaseNotExistException { + final String prepSql = + "CREATE MATERIALIZED TABLE mt1 (\n" + + " id INT COMMENT 'INT comment', t TIMESTAMP_LTZ(3)\n" + + ")\n" + + "AS SELECT 1 as id, current_timestamp as t"; + createMaterializedTableInCatalog(prepSql, "mt1"); + + final String sql = + "CREATE OR ALTER MATERIALIZED TABLE mt1 (\n" + + " id INT, t TIMESTAMP_LTZ(3) COMMENT 'Timestamp Comment'\n" + + ")\n" + + "AS SELECT 1 as id, current_timestamp as t"; + Operation operation = parse(sql); + + assertThat(operation).isInstanceOf(FullAlterMaterializedTableOperation.class); + + FullAlterMaterializedTableOperation op = (FullAlterMaterializedTableOperation) operation; + assertThat(op.getTableChanges()) + .containsExactly( + TableChange.modifyColumnComment( + Column.physical("id", DataTypes.INT()).withComment("INT comment"), + null), + TableChange.modifyColumnComment( + Column.physical("t", DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE(3)), + "Timestamp Comment")); + assertThat(operation.asSummaryString()) + .isEqualTo( + "CREATE OR ALTER MATERIALIZED TABLE builtin.default.mt1\n" + + " MODIFY `id` COMMENT '',\n" + + " MODIFY `t` COMMENT 'Timestamp Comment'"); + } + private void createMaterializedTableInCatalog(String sql, String materializedTableName) throws TableAlreadyExistException, DatabaseNotExistException { final ObjectPath objectPath =
