[Improve](cdc)(MongoDB) optimize type inference to avoid unnecessary decimal conversion (#596)

Author: kwonder0926

flink-doris-connector/src/main/java/org/apache/doris/flink/tools/cdc/mongodb/MongoDBSchema.java

@@ -106,6 +106,13 @@ private String determineDorisType(String fieldName, Object value) {
                         && existingField.getTypeString().equals(DorisType.STRING))) {
             return DorisType.STRING;
         }
+        if (existingField != null) {
+            String existingType = existingField.getTypeString();
+            if (isDowngrade(existingType, dorisType)) {
+                return existingType;
+            }
+        }
+
         // Check and process for decimal types
         DecimalJudgement decimalJudgement = judgeDecimalField(fieldName, dorisType);
         if (DecimalJudgement.needProcessing(decimalJudgement)) {
@@ -118,6 +125,27 @@ private String determineDorisType(String fieldName, Object value) {
         return dorisType;
     }
 
+    /**
+     * Check whether the newType is a downgrade from the existingType. Currently only considers
+     * numeric types: TINYINT < SMALLINT < INT < BIGINT.
+     */
+    private boolean isDowngrade(String existingType, String newType) {
+        List<String> typeHierarchy =
+                Arrays.asList(
+                        DorisType.TINYINT, DorisType.SMALLINT, DorisType.INT, DorisType.BIGINT);
+
+        int existingIndex = typeHierarchy.indexOf(existingType);
+        int newIndex = typeHierarchy.indexOf(newType);
+
+        return newIndex != -1 && existingIndex != -1 && newIndex < existingIndex;
+    }
+
+    /**
+     * Determine whether the field should be treated as a decimal type: - If the existing type is
+     * already decimal and current value is also decimal, return CERTAIN_DECIMAL. - If the field's
+     * type is convertible (e.g., INT, BIGINT) and the current value is decimal or double, return
+     * CONVERT_TO_DECIMAL. - Otherwise, no decimal processing is needed.
+     */
     private DecimalJudgement judgeDecimalField(String fieldName, String dorisType) {
         FieldSchema existingField = fields.get(fieldName);
         if (existingField == null) {
@@ -127,7 +155,7 @@ private DecimalJudgement judgeDecimalField(String fieldName, String dorisType) {
         boolean isDecimal = dorisType.startsWith(DorisType.DECIMAL);
         if (existDecimal && isDecimal) {
             return DecimalJudgement.CERTAIN_DECIMAL;
-        } else if (CONVERT_TYPE.contains(dorisType)) {
+        } else if (existDecimal && CONVERT_TYPE.contains(dorisType)) {
             return DecimalJudgement.CONVERT_TO_DECIMAL;
         }
         return DecimalJudgement.NOT_DECIMAL;

flink-doris-connector/src/test/java/org/apache/doris/flink/tools/cdc/mongodb/MongoDBSchemaTest.java

@@ -189,4 +189,49 @@ public void replaceDecimalTypeIfNeededTest6() throws Exception {
             }
         }
     }
+
+    @Test
+    public void testIntFieldNotConvertedToDecimal() throws Exception {
+        ArrayList<Document> documents = new ArrayList<>();
+        documents.add(new Document("fields1", Integer.MAX_VALUE));
+        documents.add(new Document("fields1", 1234567));
+        documents.add(new Document("fields1", Integer.MIN_VALUE));
+
+        MongoDBSchema mongoDBSchema = new MongoDBSchema(documents, "db_TEST", "test_table", "");
+        FieldSchema fieldSchema = mongoDBSchema.getFields().get("fields1");
+
+        assertEquals("INT", fieldSchema.getTypeString());
+    }
+
+    @Test
+    public void testBigIntFieldNotConvertedToDecimal() throws Exception {
+        ArrayList<Document> documents = new ArrayList<>();
+        documents.add(new Document("fields1", Long.MAX_VALUE));
+        documents.add(new Document("fields1", 12233720368541346L));
+        documents.add(new Document("fields1", Long.MIN_VALUE));
+
+        MongoDBSchema mongoDBSchema = new MongoDBSchema(documents, "db_TEST", "test_table", "");
+        Map<String, FieldSchema> fields = mongoDBSchema.getFields();
+
+        for (Map.Entry<String, FieldSchema> entry : fields.entrySet()) {
+            String fieldName = entry.getKey();
+            FieldSchema fieldSchema = entry.getValue();
+            if (fieldName.equals("fields1")) {
+                assertEquals("BIGINT", fieldSchema.getTypeString());
+            }
+        }
+    }
+
+    @Test
+    public void testMixedIntAndBigIntFieldsShouldConvertToBigInt() throws Exception {
+        ArrayList<Document> documents = new ArrayList<>();
+        documents.add(new Document("fields1", 2147483647));
+        documents.add(new Document("fields1", 9223372036854775807L));
+        documents.add(new Document("fields1", 12234));
+
+        MongoDBSchema mongoDBSchema = new MongoDBSchema(documents, "db_TEST", "test_table", "");
+        FieldSchema fieldSchema = mongoDBSchema.getFields().get("fields1");
+
+        assertEquals("BIGINT", fieldSchema.getTypeString());
+    }
 }