Preserving original data type

junmuz · ymuzammil · commit 7c8c7c4cd209 · 2025-11-03T06:06:59.000-08:00
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlDataSource.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlDataSource.java
@@ -100,7 +100,9 @@ public MySqlSourceConfig getSourceConfig() {
 
     @Override
     public SupportedMetadataColumn[] supportedMetadataColumns() {
-        return new SupportedMetadataColumn[] {new OpTsMetadataColumn()};
+        return new SupportedMetadataColumn[] {
+            new OpTsMetadataColumn(), new RowKindMetadataColumn()
+        };
     }
 
     @Override
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlEventDeserializer.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlEventDeserializer.java
@@ -189,15 +189,27 @@ protected TableId getTableId(SourceRecord record) {
 
     @Override
     protected Map<String, String> getMetadata(SourceRecord record) {
+        return getMetadata(record, null);
+    }
+
+    protected Map<String, String> getMetadata(SourceRecord record, String opType) {
         Map<String, String> metadataMap = new HashMap<>();
         readableMetadataList.forEach(
                 (mySqlReadableMetadata -> {
-                    Object metadata = mySqlReadableMetadata.getConverter().read(record);
-                    if (mySqlReadableMetadata.equals(MySqlReadableMetadata.OP_TS)) {
+                    if (mySqlReadableMetadata.equals(MySqlReadableMetadata.ROW_KIND)) {
+                        if (appendOnly) {
+                            // In append-only mode, map row_kind to the original operation type
+                            metadataMap.put(mySqlReadableMetadata.getKey(), opType);
+                        }
+                        // Skip ROW_KIND in non-append-only mode since it throws
+                        // UnsupportedOperationException
+                    } else if (mySqlReadableMetadata.equals(MySqlReadableMetadata.OP_TS)) {
+                        Object metadata = mySqlReadableMetadata.getConverter().read(record);
                         metadataMap.put(
                                 mySqlReadableMetadata.getKey(),
                                 String.valueOf(((TimestampData) metadata).getMillisecond()));
                     } else {
+                        Object metadata = mySqlReadableMetadata.getConverter().read(record);
                         metadataMap.put(mySqlReadableMetadata.getKey(), String.valueOf(metadata));
                     }
                 }));
@@ -211,18 +223,20 @@ public List<DataChangeEvent> deserializeDataChangeRecord(SourceRecord record) th
             return super.deserializeDataChangeRecord(record);
         }
 
-        // Append-only mode: convert all operations to INSERT events
+        // Append-only mode: convert all operations to INSERT events, but preserve original
+        // operation type
         Envelope.Operation op = Envelope.operationFor(record);
         TableId tableId = getTableId(record);
 
         Struct value = (Struct) record.value();
         Schema valueSchema = record.valueSchema();
-        Map<String, String> meta = getMetadata(record);
 
         if (op == Envelope.Operation.CREATE || op == Envelope.Operation.READ) {
+            Map<String, String> meta = getMetadata(record, "+I");
             RecordData after = extractAfterData(value, valueSchema);
             return Collections.singletonList(DataChangeEvent.insertEvent(tableId, after, meta));
         } else if (op == Envelope.Operation.DELETE) {
+            Map<String, String> meta = getMetadata(record, "-D");
             // For DELETE: convert to INSERT with before data
             RecordData before = extractBeforeData(value, valueSchema);
             return Collections.singletonList(DataChangeEvent.insertEvent(tableId, before, meta));
@@ -233,13 +247,15 @@ public List<DataChangeEvent> deserializeDataChangeRecord(SourceRecord record) th
             List<DataChangeEvent> events = new ArrayList<>();
 
             if (changelogMode == DebeziumChangelogMode.ALL) {
+                Map<String, String> beforeMeta = getMetadata(record, "-U");
                 // Generate INSERT event for before data (UPDATE_BEFORE -> INSERT)
                 RecordData before = extractBeforeData(value, valueSchema);
-                events.add(DataChangeEvent.insertEvent(tableId, before, meta));
+                events.add(DataChangeEvent.insertEvent(tableId, before, beforeMeta));
             }
 
+            Map<String, String> afterMeta = getMetadata(record, "+U");
             // Generate INSERT event for after data (UPDATE_AFTER -> INSERT)
-            events.add(DataChangeEvent.insertEvent(tableId, after, meta));
+            events.add(DataChangeEvent.insertEvent(tableId, after, afterMeta));
             return events;
         } else {
             LOG.trace("Received {} operation, skip", op);
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/RowKindMetadataColumn.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/RowKindMetadataColumn.java
@@ -0,0 +1,51 @@
+/*
+ * 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.connectors.mysql.source;
+
+import org.apache.flink.cdc.common.source.SupportedMetadataColumn;
+import org.apache.flink.cdc.common.types.DataType;
+import org.apache.flink.cdc.common.types.DataTypes;
+
+import java.util.Map;
+
+/** A {@link SupportedMetadataColumn} for row_kind. */
+public class RowKindMetadataColumn implements SupportedMetadataColumn {
+
+    @Override
+    public String getName() {
+        return "row_kind";
+    }
+
+    @Override
+    public DataType getType() {
+        return DataTypes.STRING().notNull();
+    }
+
+    @Override
+    public Class<?> getJavaClass() {
+        return String.class;
+    }
+
+    @Override
+    public Object read(Map<String, String> metadata) {
+        if (metadata.containsKey("row_kind")) {
+            return metadata.get("row_kind");
+        }
+        throw new IllegalArgumentException("row_kind doesn't exist in the metadata: " + metadata);
+    }
+}
diff --git a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/MySqlToPaimonE2eITCase.java b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/MySqlToPaimonE2eITCase.java
@@ -203,6 +203,11 @@ public void testReadChangelogAsAppendOnlyWithPaimon() throws Exception {
                                 + "  server-time-zone: UTC\n"
                                 + "  scan.read-changelog-as-append-only.enabled: true\n"
                                 + "  use.legacy.json.format: false\n"
+                                + "  metadata.list: row_kind\n"
+                                + "\n"
+                                + "transform:\n"
+                                + "  - source-table: '^\\.+.\\.+$'\n"
+                                + "    projection: \\*, row_kind AS __internal__op_type\n"
                                 + "\n"
                                 + "sink:\n"
                                 + "  type: paimon\n"
@@ -261,11 +266,11 @@ public void testReadChangelogAsAppendOnlyWithPaimon() throws Exception {
                 database,
                 "readChangelogAsAppendOnly",
                 Arrays.asList(
-                        "1, One, Alice, 3.202, red, {\"key1\": \"value1\"}, null",
-                        "2, Two, Bob, 1.703, white, {\"key2\": \"value2\"}, null",
-                        "3, Three, Cecily, 4.105, red, {\"key3\": \"value3\"}, null",
-                        "4, Four, Derrida, 1.857, white, {\"key4\": \"value4\"}, null",
-                        "5, Five, Evelyn, 5.211, red, {\"K\": \"V\", \"k\": \"v\"}, null"));
+                        "1, One, Alice, 3.202, red, {\"key1\": \"value1\"}, null, +I",
+                        "2, Two, Bob, 1.703, white, {\"key2\": \"value2\"}, null, +I",
+                        "3, Three, Cecily, 4.105, red, {\"key3\": \"value3\"}, null, +I",
+                        "4, Four, Derrida, 1.857, white, {\"key4\": \"value4\"}, null, +I",
+                        "5, Five, Evelyn, 5.211, red, {\"K\": \"V\", \"k\": \"v\"}, null, +I"));
 
         LOG.info("Begin incremental reading stage with append-only enabled.");
 
@@ -301,23 +306,25 @@ public void testReadChangelogAsAppendOnlyWithPaimon() throws Exception {
         // after), plus delete records
         List<String> expectedAppendOnlyRecords =
                 Arrays.asList(
-                        "1, One, Alice, 3.202, red, {\"key1\": \"value1\"}, null", // Original
-                        "2, Two, Bob, 1.703, white, {\"key2\": \"value2\"}, null", // Original
-                        "3, Three, Cecily, 4.105, red, {\"key3\": \"value3\"}, null", // Original
-                        "4, Four, Derrida, 1.857, white, {\"key4\": \"value4\"}, null", // Original
-                        "5, Five, Evelyn, 5.211, red, {\"K\": \"V\", \"k\": \"v\"}, null", // Original
-                        "6, Six, Ferris, 9.813, null, null, null", // Insert
-                        "7, Seven, Grace, 2.117, null, null, null", // Insert
-                        "1, One, Alice, 3.202, red, {\"key1\": \"value1\"}, null", // Update id=1
-                        // (before)
-                        "1, One, Alice Updated, 3.202, red, {\"key1\": \"value1\"}, null", // Update
+                        "1, One, Alice, 3.202, red, {\"key1\": \"value1\"}, null, +I", // Original
+                        "2, Two, Bob, 1.703, white, {\"key2\": \"value2\"}, null, +I", // Original
+                        "3, Three, Cecily, 4.105, red, {\"key3\": \"value3\"}, null, +I", // Original
+                        "4, Four, Derrida, 1.857, white, {\"key4\": \"value4\"}, null, +I", // Original
+                        "5, Five, Evelyn, 5.211, red, {\"K\": \"V\", \"k\": \"v\"}, null, +I", // Original
+                        "6, Six, Ferris, 9.813, null, null, null, +I", // Insert
+                        "7, Seven, Grace, 2.117, null, null, null, +I", // Insert
+                        "1, One, Alice, 3.202, red, {\"key1\": \"value1\"}, null, -U", // Update
                         // id=1
-                        // (after)
-                        "2, Two, Bob, 1.703, white, {\"key2\": \"value2\"}, null", // Update id=2
                         // (before)
-                        "2, Two, Bob, 2.0, white, {\"key2\": \"value2\"}, null", // Update id=2
+                        "1, One, Alice Updated, 3.202, red, {\"key1\": \"value1\"}, null, +U", // Update
+                        // id=1 (after)
+                        "2, Two, Bob, 1.703, white, {\"key2\": \"value2\"}, null, -U", // Update
+                        // id=2
+                        // (before)
+                        "2, Two, Bob, 2.0, white, {\"key2\": \"value2\"}, null, +U", // Update id=2
                         // (after)
-                        "3, Three, Cecily, 4.105, red, {\"key3\": \"value3\"}, null" // Delete id=3
+                        "3, Three, Cecily, 4.105, red, {\"key3\": \"value3\"}, null, -D" // Delete
+                        // id=3
                         );
 
         validateSinkResult(

Original file line number	Diff line number	Diff line change
`@@ -100,7 +100,9 @@ public MySqlSourceConfig getSourceConfig() {`
`100`	`100`
`101`	`101`	`@Override`
`102`	`102`	`public SupportedMetadataColumn[] supportedMetadataColumns() {`
`103`		`- return new SupportedMetadataColumn[] {new OpTsMetadataColumn()};`
	`103`	`+ return new SupportedMetadataColumn[] {`
	`104`	`+ new OpTsMetadataColumn(), new RowKindMetadataColumn()`
	`105`	`+ };`
`104`	`106`	`}`
`105`	`107`
`106`	`108`	`@Override`