[cdc] Ignore decimal length and precision change by type-mapping (#5124)

JackeyLee007 · 李鹏程 · web-flow · commit 8abbc5e6f64e · 2025-02-25T10:57:19.000+08:00
Co-authored-by: 李鹏程 &lt;li.pc@topsports.com.cn&gt;
diff --git a/docs/layouts/shortcodes/generated/kafka_sync_database.html b/docs/layouts/shortcodes/generated/kafka_sync_database.html
@@ -86,6 +86,7 @@
                 <li>"char-to-string": maps MySQL CHAR(length)/VARCHAR(length) types to STRING.</li>
                 <li>"longtext-to-bytes": maps MySQL LONGTEXT types to BYTES.</li>
                 <li>"bigint-unsigned-to-bigint": maps MySQL BIGINT UNSIGNED, BIGINT UNSIGNED ZEROFILL, SERIAL to BIGINT. You should ensure overflow won't occur when using this option.</li>
+                <li>"decimal-no-change": Ignore decimal type change.</li>
             </ul>
         </td>
     </tr>
diff --git a/docs/layouts/shortcodes/generated/kafka_sync_table.html b/docs/layouts/shortcodes/generated/kafka_sync_table.html
@@ -58,6 +58,7 @@
                 <li>"char-to-string": maps MySQL CHAR(length)/VARCHAR(length) types to STRING.</li>
                 <li>"longtext-to-bytes": maps MySQL LONGTEXT types to BYTES.</li>
                 <li>"bigint-unsigned-to-bigint": maps MySQL BIGINT UNSIGNED, BIGINT UNSIGNED ZEROFILL, SERIAL to BIGINT. You should ensure overflow won't occur when using this option.</li>
+                <li>"decimal-no-change": Ignore decimal type change.</li>
             </ul>
         </td>
     </tr>
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SyncDatabaseActionBase.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SyncDatabaseActionBase.java
@@ -231,6 +231,7 @@ protected void buildSink(
                 .withInput(input)
                 .withParserFactory(parserFactory)
                 .withCatalogLoader(catalogLoader())
+                .withTypeMapping(typeMapping)
                 .withDatabase(database)
                 .withTables(tables)
                 .withMode(mode)
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SyncTableActionBase.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SyncTableActionBase.java
@@ -168,6 +168,7 @@ protected void buildSink(
                         .withParserFactory(parserFactory)
                         .withTable(fileStoreTable)
                         .withIdentifier(new Identifier(database, table))
+                        .withTypeMapping(typeMapping)
                         .withCatalogLoader(catalogLoader());
         String sinkParallelism = tableConfig.get(FlinkConnectorOptions.SINK_PARALLELISM.key());
         if (sinkParallelism != null) {
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/TypeMapping.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/TypeMapping.java
@@ -76,6 +76,7 @@ public enum TypeMappingMode {
         TO_STRING,
         CHAR_TO_STRING,
         LONGTEXT_TO_BYTES,
+        DECIMAL_NO_CHANGE,
         BIGINT_UNSIGNED_TO_BIGINT;
 
         private static final Map<String, TypeMappingMode> TYPE_MAPPING_OPTIONS =
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcSinkBuilder.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcSinkBuilder.java
@@ -21,6 +21,7 @@
 import org.apache.paimon.annotation.Experimental;
 import org.apache.paimon.catalog.CatalogLoader;
 import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.flink.action.cdc.TypeMapping;
 import org.apache.paimon.flink.utils.SingleOutputStreamOperatorUtils;
 import org.apache.paimon.schema.SchemaManager;
 import org.apache.paimon.table.BucketMode;
@@ -50,6 +51,7 @@ public class CdcSinkBuilder<T> {
     private Table table = null;
     private Identifier identifier = null;
     private CatalogLoader catalogLoader = null;
+    private TypeMapping typeMapping = null;
 
     @Nullable private Integer parallelism;
 
@@ -83,6 +85,11 @@ public CdcSinkBuilder<T> withCatalogLoader(CatalogLoader catalogLoader) {
         return this;
     }
 
+    public CdcSinkBuilder<T> withTypeMapping(TypeMapping typeMapping) {
+        this.typeMapping = typeMapping;
+        return this;
+    }
+
     public DataStreamSink<?> build() {
         Preconditions.checkNotNull(input, "Input DataStream can not be null.");
         Preconditions.checkNotNull(parserFactory, "Event ParserFactory can not be null.");
@@ -110,7 +117,8 @@ public DataStreamSink<?> build() {
                                 new UpdatedDataFieldsProcessFunction(
                                         new SchemaManager(dataTable.fileIO(), dataTable.location()),
                                         identifier,
-                                        catalogLoader))
+                                        catalogLoader,
+                                        typeMapping))
                         .name("Schema Evolution");
         schemaChangeProcessFunction.getTransformation().setParallelism(1);
         schemaChangeProcessFunction.getTransformation().setMaxParallelism(1);
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcSyncDatabaseSinkBuilder.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcSyncDatabaseSinkBuilder.java
@@ -22,6 +22,7 @@
 import org.apache.paimon.catalog.Identifier;
 import org.apache.paimon.flink.FlinkConnectorOptions;
 import org.apache.paimon.flink.action.MultiTablesSinkMode;
+import org.apache.paimon.flink.action.cdc.TypeMapping;
 import org.apache.paimon.flink.sink.FlinkWriteSink;
 import org.apache.paimon.flink.utils.SingleOutputStreamOperatorUtils;
 import org.apache.paimon.options.MemorySize;
@@ -74,6 +75,8 @@ public class FlinkCdcSyncDatabaseSinkBuilder<T> {
     //     Paimon tables. 2) in multiplex sink where it is used to
     //     initialize different writers to multiple tables.
     private CatalogLoader catalogLoader;
+    private TypeMapping typeMapping;
+
     // database to sync, currently only support single database
     private String database;
     private MultiTablesSinkMode mode;
@@ -122,6 +125,11 @@ public FlinkCdcSyncDatabaseSinkBuilder<T> withMode(MultiTablesSinkMode mode) {
         return this;
     }
 
+    public FlinkCdcSyncDatabaseSinkBuilder<T> withTypeMapping(TypeMapping typeMapping) {
+        this.typeMapping = typeMapping;
+        return this;
+    }
+
     public void build() {
         Preconditions.checkNotNull(input);
         Preconditions.checkNotNull(parserFactory);
@@ -154,7 +162,7 @@ private void buildCombinedCdcSink() {
                         parsed,
                         CdcDynamicTableParsingProcessFunction.DYNAMIC_SCHEMA_CHANGE_OUTPUT_TAG)
                 .keyBy(t -> t.f0)
-                .process(new MultiTableUpdatedDataFieldsProcessFunction(catalogLoader))
+                .process(new MultiTableUpdatedDataFieldsProcessFunction(catalogLoader, typeMapping))
                 .name("Schema Evolution");
 
         DataStream<CdcMultiplexRecord> converted =
@@ -202,7 +210,8 @@ private void buildDividedCdcSink() {
                                     new UpdatedDataFieldsProcessFunction(
                                             new SchemaManager(table.fileIO(), table.location()),
                                             Identifier.create(database, table.name()),
-                                            catalogLoader))
+                                            catalogLoader,
+                                            typeMapping))
                             .name("Schema Evolution");
             schemaChangeProcessFunction.getTransformation().setParallelism(1);
             schemaChangeProcessFunction.getTransformation().setMaxParallelism(1);
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/MultiTableUpdatedDataFieldsProcessFunction.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/MultiTableUpdatedDataFieldsProcessFunction.java
@@ -21,6 +21,7 @@
 import org.apache.paimon.catalog.Catalog;
 import org.apache.paimon.catalog.CatalogLoader;
 import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.flink.action.cdc.TypeMapping;
 import org.apache.paimon.schema.SchemaChange;
 import org.apache.paimon.schema.SchemaManager;
 import org.apache.paimon.table.FileStoreTable;
@@ -52,8 +53,9 @@ public class MultiTableUpdatedDataFieldsProcessFunction
 
     private final Map<Identifier, SchemaManager> schemaManagers = new HashMap<>();
 
-    public MultiTableUpdatedDataFieldsProcessFunction(CatalogLoader catalogLoader) {
-        super(catalogLoader);
+    public MultiTableUpdatedDataFieldsProcessFunction(
+            CatalogLoader catalogLoader, TypeMapping typeMapping) {
+        super(catalogLoader, typeMapping);
     }
 
     @Override
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/UpdatedDataFieldsProcessFunction.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/UpdatedDataFieldsProcessFunction.java
@@ -20,6 +20,7 @@
 
 import org.apache.paimon.catalog.CatalogLoader;
 import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.flink.action.cdc.TypeMapping;
 import org.apache.paimon.schema.SchemaChange;
 import org.apache.paimon.schema.SchemaManager;
 import org.apache.paimon.types.DataField;
@@ -53,8 +54,11 @@ public class UpdatedDataFieldsProcessFunction
     private Set<FieldIdentifier> latestFields;
 
     public UpdatedDataFieldsProcessFunction(
-            SchemaManager schemaManager, Identifier identifier, CatalogLoader catalogLoader) {
-        super(catalogLoader);
+            SchemaManager schemaManager,
+            Identifier identifier,
+            CatalogLoader catalogLoader,
+            TypeMapping typeMapping) {
+        super(catalogLoader, typeMapping);
         this.schemaManager = schemaManager;
         this.identifier = identifier;
         this.latestFields = new HashSet<>();
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/UpdatedDataFieldsProcessFunctionBase.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/UpdatedDataFieldsProcessFunctionBase.java
@@ -21,6 +21,7 @@
 import org.apache.paimon.catalog.Catalog;
 import org.apache.paimon.catalog.CatalogLoader;
 import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.flink.action.cdc.TypeMapping;
 import org.apache.paimon.schema.SchemaChange;
 import org.apache.paimon.schema.SchemaManager;
 import org.apache.paimon.schema.TableSchema;
@@ -52,6 +53,7 @@ public abstract class UpdatedDataFieldsProcessFunctionBase<I, O> extends Process
     protected final CatalogLoader catalogLoader;
     protected Catalog catalog;
     private boolean caseSensitive;
+    private TypeMapping typeMapping;
 
     private static final List<DataTypeRoot> STRING_TYPES =
             Arrays.asList(DataTypeRoot.CHAR, DataTypeRoot.VARCHAR);
@@ -71,8 +73,10 @@ public abstract class UpdatedDataFieldsProcessFunctionBase<I, O> extends Process
     private static final List<DataTypeRoot> TIMESTAMP_TYPES =
             Arrays.asList(DataTypeRoot.TIMESTAMP_WITHOUT_TIME_ZONE);
 
-    protected UpdatedDataFieldsProcessFunctionBase(CatalogLoader catalogLoader) {
+    protected UpdatedDataFieldsProcessFunctionBase(
+            CatalogLoader catalogLoader, TypeMapping typeMapping) {
         this.catalogLoader = catalogLoader;
+        this.typeMapping = typeMapping;
     }
 
     /**
@@ -214,6 +218,11 @@ protected List<SchemaChange> extractSchemaChanges(
             oldFields.put(oldField.name(), oldField);
         }
 
+        boolean allowDecimalTypeChange =
+                this.typeMapping == null
+                        || !this.typeMapping.containsMode(
+                                TypeMapping.TypeMappingMode.DECIMAL_NO_CHANGE);
+
         List<SchemaChange> result = new ArrayList<>();
         for (DataField newField : updatedDataFields) {
             String newFieldName = StringUtils.toLowerCaseIfNeed(newField.name(), caseSensitive);
@@ -232,6 +241,9 @@ protected List<SchemaChange> extractSchemaChanges(
                                         new String[] {newFieldName}, newField.description()));
                     }
                 } else {
+                    if (oldField.type().is(DataTypeRoot.DECIMAL) && !allowDecimalTypeChange) {
+                        continue;
+                    }
                     // update column type
                     result.add(SchemaChange.updateColumnType(newFieldName, newField.type()));
                     // update column comment
diff --git a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/SchemaEvolutionTest.java b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/SchemaEvolutionTest.java
@@ -210,7 +210,8 @@ public void testSchemaEvolution() throws Exception {
                                 new UpdatedDataFieldsProcessFunction(
                                         new SchemaManager(table.fileIO(), table.location()),
                                         identifier,
-                                        catalogLoader))
+                                        catalogLoader,
+                                        TypeMapping.defaultMapping()))
                         .name("Schema Evolution");
         schemaChangeProcessFunction.getTransformation().setParallelism(1);
         schemaChangeProcessFunction.getTransformation().setMaxParallelism(1);
diff --git a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaCanalSyncTableActionITCase.java b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaCanalSyncTableActionITCase.java
@@ -49,6 +49,7 @@
 import static org.apache.flink.streaming.connectors.kafka.table.KafkaConnectorOptions.TOPIC;
 import static org.apache.flink.streaming.connectors.kafka.table.KafkaConnectorOptions.TOPIC_PATTERN;
 import static org.apache.flink.streaming.connectors.kafka.table.KafkaConnectorOptions.VALUE_FORMAT;
+import static org.apache.paimon.flink.action.cdc.TypeMapping.TypeMappingMode.DECIMAL_NO_CHANGE;
 import static org.apache.paimon.flink.action.cdc.TypeMapping.TypeMappingMode.TO_STRING;
 import static org.apache.paimon.testutils.assertj.PaimonAssertions.anyCauseMatches;
 import static org.assertj.core.api.Assertions.assertThat;
@@ -1148,4 +1149,39 @@ public void testComputedColumnWithCaseInsensitive(boolean triggerSchemaRetrieval
     public void testWaterMarkSyncTable() throws Exception {
         testWaterMarkSyncTable(CANAL);
     }
+
+    @Test
+    @Timeout(60)
+    public void testDecimalNoChange() throws Exception {
+        String topic = "decimal-no-change";
+        createTestTopic(topic, 1, 1);
+        writeRecordsToKafka(topic, "kafka/canal/table/typenochange/canal-data-3.txt");
+        Map<String, String> kafkaConfig = getBasicKafkaConfig();
+        kafkaConfig.put(VALUE_FORMAT.key(), "canal-json");
+        kafkaConfig.put(TOPIC.key(), topic);
+        KafkaSyncTableAction action =
+                syncTableActionBuilder(kafkaConfig)
+                        .withTableConfig(getBasicTableConfig())
+                        .withCatalogConfig(
+                                Collections.singletonMap(
+                                        CatalogOptions.CASE_SENSITIVE.key(), "false"))
+                        .withTypeMappingModes(DECIMAL_NO_CHANGE.configString())
+                        .build();
+        runActionWithDefaultEnv(action);
+        RowType rowType =
+                RowType.of(
+                        new DataType[] {DataTypes.INT().notNull(), DataTypes.DECIMAL(10, 2)},
+                        new String[] {"k", "v"});
+        waitForResult(
+                Collections.singletonList("+I[1, 1.20]"),
+                getFileStoreTable(tableName),
+                rowType,
+                Collections.singletonList("k"));
+        writeRecordsToKafka(topic, "kafka/canal/table/typenochange/canal-data-4.txt");
+        waitForResult(
+                Arrays.asList("+I[1, 1.20]", "+I[2, 2.30]"),
+                getFileStoreTable(tableName),
+                rowType, // should not change
+                Collections.singletonList("k"));
+    }
 }
diff --git a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/FlinkCdcSyncDatabaseSinkITCase.java b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/FlinkCdcSyncDatabaseSinkITCase.java
@@ -23,6 +23,7 @@
 import org.apache.paimon.catalog.CatalogUtils;
 import org.apache.paimon.data.InternalRow;
 import org.apache.paimon.flink.FlinkCatalogFactory;
+import org.apache.paimon.flink.action.cdc.TypeMapping;
 import org.apache.paimon.flink.util.AbstractTestBase;
 import org.apache.paimon.fs.FileIO;
 import org.apache.paimon.fs.Path;
@@ -172,6 +173,7 @@ private void innerTestRandomCdcEvents(Supplier<Integer> bucket, boolean unawareB
                 // each table can only get 2 slots
                 .withTableOptions(Collections.singletonMap(SINK_PARALLELISM.key(), "2"))
                 .withDatabase(DATABASE_NAME)
+                .withTypeMapping(TypeMapping.defaultMapping())
                 .withCatalogLoader(catalogLoader)
                 .build();
 
diff --git a/paimon-flink/paimon-flink-cdc/src/test/resources/kafka/canal/table/typenochange/canal-data-3.txt b/paimon-flink/paimon-flink-cdc/src/test/resources/kafka/canal/table/typenochange/canal-data-3.txt
@@ -0,0 +1,19 @@
+/*
+ * 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.
+ */
+
+{"data":[{"k":"1","v":"1.2"}],"database":"paimon_sync_table","es":1683880554000,"id":2150,"isDdl":false,"mysqlType":{"k":"int","v":"decimal(10,2)"},"old":null,"pkNames":["k"],"sql":"","sqlType":{"k":4,"v":3},"table":"decimal_no_change","ts":1683880554351,"type":"INSERT"}
diff --git a/paimon-flink/paimon-flink-cdc/src/test/resources/kafka/canal/table/typenochange/canal-data-4.txt b/paimon-flink/paimon-flink-cdc/src/test/resources/kafka/canal/table/typenochange/canal-data-4.txt
@@ -0,0 +1,19 @@
+/*
+ * 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.
+ */
+
+{"data":[{"k":"2","v":"2.3"}],"database":"paimon_sync_table","es":1683880554000,"id":2150,"isDdl":false,"mysqlType":{"k":"int","v":"decimal(15,4)"},"old":null,"pkNames":["k"],"sql":"","sqlType":{"k":4,"v":3},"table":"decimal_no_change","ts":1683880554351,"type":"INSERT"}
