[FLINK-37203] Alter table comment or column comment (apache#3893)

Co-authored-by: lvyanquan <lvyanquan.lyq@alibaba-inc.com>

Author: 2 people

flink-cdc-cli/src/test/java/org/apache/flink/cdc/cli/parser/YamlPipelineDefinitionParserTest.java

@@ -45,6 +45,7 @@
 
 import static org.apache.flink.cdc.common.event.SchemaChangeEventType.ADD_COLUMN;
 import static org.apache.flink.cdc.common.event.SchemaChangeEventType.ALTER_COLUMN_TYPE;
+import static org.apache.flink.cdc.common.event.SchemaChangeEventType.ALTER_TABLE_COMMENT;
 import static org.apache.flink.cdc.common.event.SchemaChangeEventType.CREATE_TABLE;
 import static org.apache.flink.cdc.common.event.SchemaChangeEventType.DROP_COLUMN;
 import static org.apache.flink.cdc.common.event.SchemaChangeEventType.DROP_TABLE;
@@ -222,6 +223,7 @@ void testSchemaEvolutionTypesConfiguration() throws Exception {
                 null,
                 null,
                 ImmutableSet.of(
+                        ALTER_TABLE_COMMENT,
                         ADD_COLUMN,
                         ALTER_COLUMN_TYPE,
                         CREATE_TABLE,
@@ -234,6 +236,7 @@ void testSchemaEvolutionTypesConfiguration() throws Exception {
                 null,
                 null,
                 ImmutableSet.of(
+                        ALTER_TABLE_COMMENT,
                         ADD_COLUMN,
                         ALTER_COLUMN_TYPE,
                         CREATE_TABLE,
@@ -246,6 +249,7 @@ void testSchemaEvolutionTypesConfiguration() throws Exception {
                 "[column, table]",
                 "[drop]",
                 ImmutableSet.of(
+                        ALTER_TABLE_COMMENT,
                         ADD_COLUMN,
                         ALTER_COLUMN_TYPE,
                         CREATE_TABLE,
@@ -256,12 +260,18 @@ void testSchemaEvolutionTypesConfiguration() throws Exception {
                 null,
                 null,
                 ImmutableSet.of(
-                        ADD_COLUMN, ALTER_COLUMN_TYPE, CREATE_TABLE, DROP_COLUMN, RENAME_COLUMN));
+                        ALTER_TABLE_COMMENT,
+                        ADD_COLUMN,
+                        ALTER_COLUMN_TYPE,
+                        CREATE_TABLE,
+                        DROP_COLUMN,
+                        RENAME_COLUMN));
         testSchemaEvolutionTypesParsing(
                 "lenient",
                 null,
                 "[]",
                 ImmutableSet.of(
+                        ALTER_TABLE_COMMENT,
                         ADD_COLUMN,
                         ALTER_COLUMN_TYPE,
                         CREATE_TABLE,
@@ -594,6 +604,7 @@ void testParsingFullDefinitionFromString() throws Exception {
                                             .put("bootstrap-servers", "localhost:9092")
                                             .build()),
                             ImmutableSet.of(
+                                    ALTER_TABLE_COMMENT,
                                     DROP_COLUMN,
                                     ALTER_COLUMN_TYPE,
                                     ADD_COLUMN,
@@ -620,6 +631,7 @@ void testParsingFullDefinitionFromString() throws Exception {
                             null,
                             new Configuration(),
                             ImmutableSet.of(
+                                    ALTER_TABLE_COMMENT,
                                     DROP_COLUMN,
                                     ALTER_COLUMN_TYPE,
                                     ADD_COLUMN,
@@ -706,6 +718,7 @@ void testParsingFullDefinitionFromString() throws Exception {
                             null,
                             new Configuration(),
                             ImmutableSet.of(
+                                    ALTER_TABLE_COMMENT,
                                     DROP_COLUMN,
                                     ALTER_COLUMN_TYPE,
                                     ADD_COLUMN,
@@ -742,6 +755,7 @@ void testParsingFullDefinitionFromString() throws Exception {
                             null,
                             new Configuration(),
                             ImmutableSet.of(
+                                    ALTER_TABLE_COMMENT,
                                     DROP_COLUMN,
                                     ALTER_COLUMN_TYPE,
                                     ADD_COLUMN,
@@ -797,6 +811,7 @@ void testParsingFullDefinitionFromString() throws Exception {
                                             .put("password", "")
                                             .build()),
                             ImmutableSet.of(
+                                    ALTER_TABLE_COMMENT,
                                     DROP_COLUMN,
                                     ALTER_COLUMN_TYPE,
                                     ADD_COLUMN,

flink-cdc-common/src/main/java/org/apache/flink/cdc/common/event/AlterColumnTypeEvent.java

@@ -22,6 +22,8 @@
 import org.apache.flink.cdc.common.schema.Schema;
 import org.apache.flink.cdc.common.types.DataType;
 
+import javax.annotation.Nullable;
+
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Objects;
@@ -44,26 +46,46 @@ public class AlterColumnTypeEvent implements SchemaChangeEventWithPreSchema, Sch
 
     private final Map<String, DataType> oldTypeMapping;
 
-    public AlterColumnTypeEvent(TableId tableId, Map<String, DataType> typeMapping) {
+    /** key => column name, value => column comment after changing. */
+    private final Map<String, String> comments;
+
+    public AlterColumnTypeEvent(
+            TableId tableId,
+            Map<String, DataType> typeMapping,
+            Map<String, DataType> oldTypeMapping,
+            Map<String, String> comments) {
         this.tableId = tableId;
         this.typeMapping = typeMapping;
-        this.oldTypeMapping = new HashMap<>();
+        this.oldTypeMapping = oldTypeMapping;
+        this.comments = comments;
+    }
+
+    public AlterColumnTypeEvent(TableId tableId, Map<String, DataType> typeMapping) {
+        this(tableId, typeMapping, new HashMap<>());
     }
 
     public AlterColumnTypeEvent(
             TableId tableId,
             Map<String, DataType> typeMapping,
             Map<String, DataType> oldTypeMapping) {
-        this.tableId = tableId;
-        this.typeMapping = typeMapping;
-        this.oldTypeMapping = oldTypeMapping;
+        this(tableId, typeMapping, oldTypeMapping, new HashMap<>());
+    }
+
+    public void addColumnComment(String columnName, @Nullable String comment) {
+        if (comment != null) {
+            this.comments.put(columnName, comment);
+        }
     }
 
     /** Returns the type mapping. */
     public Map<String, DataType> getTypeMapping() {
         return typeMapping;
     }
 
+    public Map<String, String> getComments() {
+        return comments;
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) {
@@ -75,12 +97,13 @@ public boolean equals(Object o) {
         AlterColumnTypeEvent that = (AlterColumnTypeEvent) o;
         return Objects.equals(tableId, that.tableId)
                 && Objects.equals(typeMapping, that.typeMapping)
-                && Objects.equals(oldTypeMapping, that.oldTypeMapping);
+                && Objects.equals(oldTypeMapping, that.oldTypeMapping)
+                && Objects.equals(comments, that.comments);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(tableId, typeMapping, oldTypeMapping);
+        return Objects.hash(tableId, typeMapping, oldTypeMapping, comments);
     }
 
     @Override
@@ -93,13 +116,17 @@ public String toString() {
                     + typeMapping
                     + ", oldTypeMapping="
                     + oldTypeMapping
+                    + ", comments="
+                    + comments
                     + '}';
         } else {
             return "AlterColumnTypeEvent{"
                     + "tableId="
                     + tableId
                     + ", typeMapping="
                     + typeMapping
+                    + ", comments="
+                    + comments
                     + '}';
         }
     }

flink-cdc-common/src/main/java/org/apache/flink/cdc/common/event/AlterTableCommentEvent.java

@@ -0,0 +1,89 @@
+/*
+ * 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.cdc.common.event;
+
+import org.apache.flink.cdc.common.annotation.PublicEvolving;
+import org.apache.flink.cdc.common.utils.Preconditions;
+
+import java.util.Objects;
+
+/**
+ * A {@link SchemaChangeEvent} that represents an {@code ALTER TABLE COMMENT} or {@code ALTER TABLE
+ * SET COMMENT} DDL.
+ */
+@PublicEvolving
+public class AlterTableCommentEvent implements SchemaChangeEvent {
+
+    private static final long serialVersionUID = 1L;
+
+    private final TableId tableId;
+    private final String comment;
+
+    public AlterTableCommentEvent(TableId tableId, String comment) {
+        Preconditions.checkArgument(comment != null, "comment should not be null.");
+        this.tableId = tableId;
+        this.comment = comment;
+    }
+
+    public String getComment() {
+        return comment;
+    }
+
+    @Override
+    public SchemaChangeEventType getType() {
+        return SchemaChangeEventType.ALTER_TABLE_COMMENT;
+    }
+
+    @Override
+    public SchemaChangeEvent copy(TableId newTableId) {
+        return new AlterTableCommentEvent(newTableId, comment);
+    }
+
+    @Override
+    public TableId tableId() {
+        return tableId;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (!(o instanceof AlterTableCommentEvent)) {
+            return false;
+        }
+        AlterTableCommentEvent that = (AlterTableCommentEvent) o;
+        return Objects.equals(tableId, that.tableId) && Objects.equals(comment, that.comment);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(tableId, comment);
+    }
+
+    @Override
+    public String toString() {
+        return "AlterTableCommentEvent{"
+                + "tableId="
+                + tableId
+                + ", comment='"
+                + comment
+                + '\''
+                + '}';
+    }
+}

flink-cdc-common/src/main/java/org/apache/flink/cdc/common/event/SchemaChangeEventType.java

@@ -28,7 +28,8 @@ public enum SchemaChangeEventType {
     DROP_COLUMN("drop.column"),
     DROP_TABLE("drop.table"),
     RENAME_COLUMN("rename.column"),
-    TRUNCATE_TABLE("truncate.table");
+    TRUNCATE_TABLE("truncate.table"),
+    ALTER_TABLE_COMMENT("alter.table.comment");
 
     private final String tag;
 
@@ -55,6 +56,8 @@ public static SchemaChangeEventType ofEvent(SchemaChangeEvent event) {
             return RENAME_COLUMN;
         } else if (event instanceof TruncateTableEvent) {
             return TRUNCATE_TABLE;
+        } else if (event instanceof AlterTableCommentEvent) {
+            return ALTER_TABLE_COMMENT;
         } else {
             throw new RuntimeException("Unknown schema change event type: " + event.getClass());
         }
@@ -76,6 +79,8 @@ public static SchemaChangeEventType ofTag(String tag) {
                 return RENAME_COLUMN;
             case "truncate.table":
                 return TRUNCATE_TABLE;
+            case "alter.table.comment":
+                return ALTER_TABLE_COMMENT;
             default:
                 throw new RuntimeException("Unknown schema change event type: " + tag);
         }

flink-cdc-common/src/main/java/org/apache/flink/cdc/common/event/SchemaChangeEventTypeFamily.java

@@ -21,6 +21,7 @@
 
 import static org.apache.flink.cdc.common.event.SchemaChangeEventType.ADD_COLUMN;
 import static org.apache.flink.cdc.common.event.SchemaChangeEventType.ALTER_COLUMN_TYPE;
+import static org.apache.flink.cdc.common.event.SchemaChangeEventType.ALTER_TABLE_COMMENT;
 import static org.apache.flink.cdc.common.event.SchemaChangeEventType.CREATE_TABLE;
 import static org.apache.flink.cdc.common.event.SchemaChangeEventType.DROP_COLUMN;
 import static org.apache.flink.cdc.common.event.SchemaChangeEventType.DROP_TABLE;
@@ -44,7 +45,9 @@ public class SchemaChangeEventTypeFamily {
 
     public static final SchemaChangeEventType[] RENAME = {RENAME_COLUMN};
 
-    public static final SchemaChangeEventType[] TABLE = {CREATE_TABLE, DROP_TABLE, TRUNCATE_TABLE};
+    public static final SchemaChangeEventType[] TABLE = {
+        CREATE_TABLE, DROP_TABLE, TRUNCATE_TABLE, ALTER_TABLE_COMMENT
+    };
 
     public static final SchemaChangeEventType[] COLUMN = {
         ADD_COLUMN, ALTER_COLUMN_TYPE, DROP_COLUMN, RENAME_COLUMN
@@ -57,7 +60,8 @@ public class SchemaChangeEventTypeFamily {
         DROP_COLUMN,
         DROP_TABLE,
         RENAME_COLUMN,
-        TRUNCATE_TABLE
+        TRUNCATE_TABLE,
+        ALTER_TABLE_COMMENT
     };
 
     public static final SchemaChangeEventType[] NONE = {};

flink-cdc-common/src/main/java/org/apache/flink/cdc/common/event/visitor/AlterTableCommentEventVisitor.java

@@ -0,0 +1,28 @@
+/*
+ * 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.cdc.common.event.visitor;
+
+import org.apache.flink.cdc.common.annotation.Internal;
+import org.apache.flink.cdc.common.event.AlterTableCommentEvent;
+
+/** Visitor for {@link AlterTableCommentEvent}s. */
+@Internal
+@FunctionalInterface
+public interface AlterTableCommentEventVisitor<T, E extends Throwable> {
+    T visit(AlterTableCommentEvent event) throws E;
+}

flink-cdc-common/src/main/java/org/apache/flink/cdc/common/event/visitor/SchemaChangeEventVisitor.java

@@ -20,6 +20,7 @@
 import org.apache.flink.cdc.common.annotation.Internal;
 import org.apache.flink.cdc.common.event.AddColumnEvent;
 import org.apache.flink.cdc.common.event.AlterColumnTypeEvent;
+import org.apache.flink.cdc.common.event.AlterTableCommentEvent;
 import org.apache.flink.cdc.common.event.CreateTableEvent;
 import org.apache.flink.cdc.common.event.DropColumnEvent;
 import org.apache.flink.cdc.common.event.DropTableEvent;
@@ -38,7 +39,8 @@ public static <T, E extends Throwable> T visit(
             DropColumnEventVisitor<T, E> dropColumnEventVisitor,
             DropTableEventVisitor<T, E> dropTableEventVisitor,
             RenameColumnEventVisitor<T, E> renameColumnEventVisitor,
-            TruncateTableEventVisitor<T, E> truncateTableEventVisitor)
+            TruncateTableEventVisitor<T, E> truncateTableEventVisitor,
+            AlterTableCommentEventVisitor<T, E> alterTableCommentEventVisitor)
             throws E {
         if (event instanceof AddColumnEvent) {
             if (addColumnVisitor == null) {
@@ -75,6 +77,11 @@ public static <T, E extends Throwable> T visit(
                 return null;
             }
             return truncateTableEventVisitor.visit((TruncateTableEvent) event);
+        } else if (event instanceof AlterTableCommentEvent) {
+            if (alterTableCommentEventVisitor == null) {
+                return null;
+            }
+            return alterTableCommentEventVisitor.visit((AlterTableCommentEvent) event);
         } else {
             throw new IllegalArgumentException(
                     "Unknown schema change event type " + event.getType());