[FLINK-38134][cdc-common] Cannot recognize DropColumnEvent in method SchemaMergingUtils.getSchemaDifference(#4065)

Author: tbpure

flink-cdc-common/src/main/java/org/apache/flink/cdc/common/utils/SchemaMergingUtils.java

@@ -29,6 +29,7 @@
 import org.apache.flink.cdc.common.event.AddColumnEvent;
 import org.apache.flink.cdc.common.event.AlterColumnTypeEvent;
 import org.apache.flink.cdc.common.event.CreateTableEvent;
+import org.apache.flink.cdc.common.event.DropColumnEvent;
 import org.apache.flink.cdc.common.event.SchemaChangeEvent;
 import org.apache.flink.cdc.common.event.TableId;
 import org.apache.flink.cdc.common.schema.Column;
@@ -212,6 +213,7 @@ public static List<SchemaChangeEvent> getSchemaDifference(
                     oldTypeMapping.put(columnName, beforeType);
                     newTypeMapping.put(columnName, afterType);
                 }
+                beforeColumns.remove(columnName);
             } else {
                 if (afterWhichColumnPosition == null) {
                     appendedColumns.add(
@@ -238,6 +240,10 @@ public static List<SchemaChangeEvent> getSchemaDifference(
                     new AlterColumnTypeEvent(tableId, newTypeMapping, oldTypeMapping));
         }
 
+        if (!beforeColumns.isEmpty()) {
+            schemaChangeEvents.add(
+                    new DropColumnEvent(tableId, new ArrayList<>(beforeColumns.keySet())));
+        }
         return schemaChangeEvents;
     }
 

flink-cdc-common/src/test/java/org/apache/flink/cdc/common/utils/SchemaMergingUtilsTest.java

@@ -27,6 +27,7 @@
 import org.apache.flink.cdc.common.event.AddColumnEvent;
 import org.apache.flink.cdc.common.event.AlterColumnTypeEvent;
 import org.apache.flink.cdc.common.event.CreateTableEvent;
+import org.apache.flink.cdc.common.event.DropColumnEvent;
 import org.apache.flink.cdc.common.event.SchemaChangeEvent;
 import org.apache.flink.cdc.common.event.TableId;
 import org.apache.flink.cdc.common.schema.Column;
@@ -417,7 +418,28 @@ TABLE_ID, of("id", BIGINT), of("name", VARCHAR(17), "id", BIGINT)))
                                 TABLE_ID,
                                 Collections.singletonMap("name", STRING),
                                 Collections.singletonMap("name", VARCHAR(17))));
-
+        Assertions.assertThat(
+                        getSchemaDifference(
+                                TABLE_ID,
+                                of("id", BIGINT, "name", STRING, "number", BIGINT),
+                                of("id", BIGINT)))
+                .as("test remove id while add gentle")
+                .containsExactly(new DropColumnEvent(TABLE_ID, Arrays.asList("number", "name")));
+        Assertions.assertThat(
+                        getSchemaDifference(
+                                TABLE_ID,
+                                of("id", BIGINT, "name", STRING, "number", BIGINT),
+                                of("id", BIGINT, "name", STRING, "gentle", STRING)))
+                .as("test remove id while add gentle")
+                .containsExactly(
+                        new AddColumnEvent(
+                                TABLE_ID,
+                                Collections.singletonList(
+                                        new AddColumnEvent.ColumnWithPosition(
+                                                Column.physicalColumn("gentle", STRING),
+                                                AddColumnEvent.ColumnPosition.AFTER,
+                                                "name"))),
+                        new DropColumnEvent(TABLE_ID, Collections.singletonList("number")));
         Stream.of(TINYINT, SMALLINT, INT)
                 .forEach(
                         type ->