Fix: Adjust VARCHAR byte length calculation to fully support UTF-8 (utf8mb4) (#620)

The current logic in `TypeConverter.java` uses a multiplier of `3` to calculate the required byte length for the Doris `VARCHAR` type:

```java
// Current implementation
return length * 3 > 65533
        ? DorisType.STRING
        : String.format("%s(%s)", DorisType.VARCHAR, length * 3);
```

This assumes a maximum of 3 bytes per character, which is insufficient for the widely used utf8mb4 character set (common in MySQL/MariaDB and other sources). The utf8mb4 encoding supports the full range of Unicode characters (including emojis), requiring up to 4 bytes per character.

If a source column contains 4-byte characters, the calculated byte length may underestimate the required size, leading to:

Data truncation or corruption during the synchronization process.

Load failures with errors such as "data length exceeded" or "row size too large" when Doris enforces the byte limit.

Proposed Solution
This change updates the byte multiplier from 3 to 4 to safely accommodate the full utf8mb4 character set, ensuring the calculated byte length is always sufficient for the defined character length, thus guaranteeing data integrity and preventing sync failures.

Author: Kane

flink-doris-connector/src/main/java/org/apache/doris/flink/catalog/DorisTypeMapper.java

@@ -155,7 +155,7 @@ private static class LogicalTypeVisitor extends LogicalTypeDefaultVisitor<String
 
         @Override
         public String visit(CharType charType) {
-            long length = charType.getLength() * 3L;
+            long length = charType.getLength() * 4L;
             if (length <= MAX_CHAR_SIZE) {
                 return String.format("%s(%s)", DorisType.CHAR, length);
             } else {
@@ -166,7 +166,7 @@ public String visit(CharType charType) {
         @Override
         public String visit(VarCharType varCharType) {
             // Flink varchar length max value is int, it may overflow after multiplying by 3
-            long length = varCharType.getLength() * 3L;
+            long length = varCharType.getLength() * 4L;
             return length >= MAX_VARCHAR_SIZE ? STRING : String.format("%s(%s)", VARCHAR, length);
         }
 

flink-doris-connector/src/main/java/org/apache/doris/flink/sink/writer/serializer/jsondebezium/JsonDebeziumSchemaChangeImpl.java

@@ -148,12 +148,12 @@ private String handleType(String type) {
         if (type == null || "".equals(type)) {
             return "";
         }
-        // varchar len * 3
+        // varchar len * 4
         Pattern pattern = Pattern.compile("varchar\\(([1-9][0-9]*)\\)", Pattern.CASE_INSENSITIVE);
         Matcher matcher = pattern.matcher(type);
         if (matcher.find()) {
             String len = matcher.group(1);
-            return String.format("varchar(%d)", Math.min(Integer.parseInt(len) * 3, 65533));
+            return String.format("varchar(%d)", Math.min(Integer.parseInt(len) * 4, 65533));
         }
 
         return type;

flink-doris-connector/src/main/java/org/apache/doris/flink/tools/cdc/db2/Db2Type.java

@@ -76,9 +76,9 @@ public static String toDorisType(String db2Type, Integer precision, Integer scal
             case VARCHAR:
             case LONG_VARCHAR:
                 Preconditions.checkNotNull(precision);
-                return precision * 3 > 65533
+                return precision * 4 > 65533
                         ? DorisType.STRING
-                        : String.format("%s(%s)", DorisType.VARCHAR, precision * 3);
+                        : String.format("%s(%s)", DorisType.VARCHAR, precision * 4);
             case TIMESTAMP:
                 return String.format(
                         "%s(%s)", DorisType.DATETIME_V2, Math.min(scale == null ? 0 : scale, 6));

flink-doris-connector/src/main/java/org/apache/doris/flink/tools/cdc/mysql/MysqlType.java

@@ -192,9 +192,9 @@ public static String toDorisType(String type, Integer length, Integer scale) {
             case CHAR:
             case VARCHAR:
                 Preconditions.checkNotNull(length);
-                return length * 3 > 65533
+                return length * 4 > 65533
                         ? DorisType.STRING
-                        : String.format("%s(%s)", DorisType.VARCHAR, length * 3);
+                        : String.format("%s(%s)", DorisType.VARCHAR, length * 4);
             case TINYTEXT:
             case TEXT:
             case MEDIUMTEXT:

flink-doris-connector/src/main/java/org/apache/doris/flink/tools/cdc/oracle/OracleType.java

@@ -91,9 +91,9 @@ public static String toDorisType(String oracleType, Integer precision, Integer s
             case NCHAR:
             case NVARCHAR2:
                 Preconditions.checkNotNull(precision);
-                return precision * 3 > 65533
+                return precision * 4 > 65533
                         ? DorisType.STRING
-                        : String.format("%s(%s)", DorisType.VARCHAR, precision * 3);
+                        : String.format("%s(%s)", DorisType.VARCHAR, precision * 4);
             case LONG:
             case RAW:
             case LONG_RAW:

flink-doris-connector/src/main/java/org/apache/doris/flink/tools/cdc/postgres/PostgresType.java

@@ -107,9 +107,9 @@ public static String toDorisType(String postgresType, Integer precision, Integer
             case BPCHAR:
             case VARCHAR:
                 Preconditions.checkNotNull(precision);
-                return precision * 3 > 65533
+                return precision * 4 > 65533
                         ? DorisType.STRING
-                        : String.format("%s(%s)", DorisType.VARCHAR, precision * 3);
+                        : String.format("%s(%s)", DorisType.VARCHAR, precision * 4);
             case POINT:
             case LINE:
             case LSEG:

flink-doris-connector/src/main/java/org/apache/doris/flink/tools/cdc/sqlserver/SqlServerType.java

@@ -94,9 +94,9 @@ public static String toDorisType(String originSqlServerType, Integer precision,
             case VARCHAR:
             case NCHAR:
             case NVARCHAR:
-                return precision * 3 > 65533
+                return precision * 4 > 65533
                         ? DorisType.STRING
-                        : String.format("%s(%s)", DorisType.VARCHAR, precision * 3);
+                        : String.format("%s(%s)", DorisType.VARCHAR, precision * 4);
             case TEXT:
             case NTEXT:
             case TIME:

flink-doris-connector/src/test/java/org/apache/doris/flink/catalog/DorisTypeMapperTest.java

@@ -109,9 +109,9 @@ public void testCharType() {
         DataType result = DorisTypeMapper.toFlinkType("col", "CHAR", 10, 0);
         assertEquals(DataTypes.CHAR(10), result);
         String dorisType = DorisTypeMapper.toDorisType(DataTypes.CHAR(10));
-        assertEquals("CHAR(30)", dorisType);
+        assertEquals("CHAR(40)", dorisType);
         dorisType = DorisTypeMapper.toDorisType(DataTypes.CHAR(100));
-        assertEquals("VARCHAR(300)", dorisType);
+        assertEquals("VARCHAR(400)", dorisType);
     }
 
     @Test