[FLINK-38596][pipeline-connector][maxcompute] Fix the column comment of CreateTableEvent and AddColumnEvent (#4215)

Author: linguoxuan

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute/src/main/java/org/apache/flink/cdc/connectors/maxcompute/utils/SchemaEvolutionUtils.java

@@ -122,10 +122,15 @@ public static void addColumns(
                 sqlBuilder
                         .append(addColumn.getAddColumn().getName())
                         .append(" ")
-                        .append(string(addColumn.getAddColumn().getType()))
-                        .append(" comment '")
-                        .append(addColumn.getAddColumn().getType().asSummaryString())
-                        .append("',");
+                        .append(string(addColumn.getAddColumn().getType()));
+                // Add comment if available
+                if (addColumn.getAddColumn().getComment() != null) {
+                    sqlBuilder
+                            .append(" comment '")
+                            .append(addColumn.getAddColumn().getComment())
+                            .append("'");
+                }
+                sqlBuilder.append(",");
             } else {
                 throw new UnsupportedOperationException(
                         "Not support position: "

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute/src/main/java/org/apache/flink/cdc/connectors/maxcompute/utils/TypeConvertUtils.java

@@ -117,6 +117,10 @@ public static Column toMaxCompute(
         if (!type.isNullable() || notNull) {
             columnBuilder.notNull();
         }
+        // Set column comment if available
+        if (flinkColumn.getComment() != null) {
+            columnBuilder.withComment(flinkColumn.getComment());
+        }
         return columnBuilder.build();
     }
 

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute/src/test/java/org/apache/flink/cdc/connectors/maxcompute/utils/TypeConvertUtilsTest.java

@@ -184,4 +184,46 @@ void testRecordConvert() {
                 "char,varchar,string,false,=01=02=03=04=05,=01=02=03=04=05=06=07=08=09=0A,0.00,1,2,12345,12345,123.456,123456.789,00:20:34.567,2003-10-20,1970-01-01T00:00,1970-01-01T00:00:00Z,1970-01-01T00:00:00Z";
         assertThat(arrayRecord).hasToString(expect);
     }
+
+    @Test
+    void testColumnCommentConversion() {
+        // Test column with comment
+        org.apache.flink.cdc.common.schema.Column columnWithComment =
+                org.apache.flink.cdc.common.schema.Column.physicalColumn(
+                        "user_id", DataTypes.BIGINT(), "Primary key for user ID");
+        com.aliyun.odps.Column maxComputeColumn =
+                TypeConvertUtils.toMaxCompute(columnWithComment, false);
+
+        assertThat(maxComputeColumn.getName()).isEqualTo("user_id");
+        assertThat(maxComputeColumn.getTypeInfo().getTypeName().toLowerCase()).isEqualTo("bigint");
+        assertThat(maxComputeColumn.getComment()).isEqualTo("Primary key for user ID");
+
+        // Test column without comment
+        org.apache.flink.cdc.common.schema.Column columnWithoutComment =
+                org.apache.flink.cdc.common.schema.Column.physicalColumn(
+                        "name", DataTypes.STRING());
+        com.aliyun.odps.Column maxComputeColumnNoComment =
+                TypeConvertUtils.toMaxCompute(columnWithoutComment, false);
+
+        assertThat(maxComputeColumnNoComment.getName()).isEqualTo("name");
+        assertThat(maxComputeColumnNoComment.getTypeInfo().getTypeName().toLowerCase())
+                .isEqualTo("string");
+        assertThat(maxComputeColumnNoComment.getComment()).isNull();
+
+        // Test schema conversion with comments
+        Schema schemaWithComments =
+                Schema.newBuilder()
+                        .physicalColumn("id", DataTypes.BIGINT(), "Primary key ID")
+                        .physicalColumn("username", DataTypes.STRING(), "User name")
+                        .physicalColumn("email", DataTypes.STRING(), "Email address")
+                        .build();
+
+        TableSchema maxComputeSchema = TypeConvertUtils.toMaxCompute(schemaWithComments);
+        List<com.aliyun.odps.Column> columns = maxComputeSchema.getAllColumns();
+
+        assertThat(columns).hasSize(3);
+        assertThat(columns.get(0).getComment()).isEqualTo("Primary key ID");
+        assertThat(columns.get(1).getComment()).isEqualTo("User name");
+        assertThat(columns.get(2).getComment()).isEqualTo("Email address");
+    }
 }