From 17303930fb96e9bb37c4370709d3da13d800dd0f Mon Sep 17 00:00:00 2001
From: hutiefang <hutiefang@qq.com>
Date: Tue, 30 Jun 2026 07:34:43 +0800
Subject: [PATCH] fix(cdc): normalize zero length varchar metadata

---
 .../dinky/cdc/convert/DataTypeConverter.java  |  9 +++++++-
 .../cdc/convert/DataTypeConverterTest.java    | 22 +++++++++++++++++++
 2 files changed, 30 insertions(+), 1 deletion(-)

diff --git a/dinky-cdc/dinky-cdc-core/src/main/java/org/dinky/cdc/convert/DataTypeConverter.java b/dinky-cdc/dinky-cdc-core/src/main/java/org/dinky/cdc/convert/DataTypeConverter.java
index 852dd4f3ec..6b169f2b97 100644
--- a/dinky-cdc/dinky-cdc-core/src/main/java/org/dinky/cdc/convert/DataTypeConverter.java
+++ b/dinky-cdc/dinky-cdc-core/src/main/java/org/dinky/cdc/convert/DataTypeConverter.java
@@ -100,10 +100,17 @@ public static LogicalType getLogicalType(Column column) {
                 return new VarBinaryType(Integer.MAX_VALUE);
             case STRING:
             default:
-                return new VarCharType(Asserts.isNull(column.getLength()) ? Integer.MAX_VALUE : column.getLength());
+                return new VarCharType(normalizeVarCharLength(column.getLength()));
         }
     }
 
+    private static int normalizeVarCharLength(Integer length) {
+        if (Asserts.isNull(length) || length < VarCharType.MIN_LENGTH) {
+            return VarCharType.MAX_LENGTH;
+        }
+        return length;
+    }
+
     public static Object convertToRow(Object value, LogicalType logicalType, ZoneId timeZone) {
         if (Asserts.isNull(value)) {
             return null;
diff --git a/dinky-cdc/dinky-cdc-core/src/test/java/org/dinky/cdc/convert/DataTypeConverterTest.java b/dinky-cdc/dinky-cdc-core/src/test/java/org/dinky/cdc/convert/DataTypeConverterTest.java
index 6eac6d75e5..ed2d9ae40d 100644
--- a/dinky-cdc/dinky-cdc-core/src/test/java/org/dinky/cdc/convert/DataTypeConverterTest.java
+++ b/dinky-cdc/dinky-cdc-core/src/test/java/org/dinky/cdc/convert/DataTypeConverterTest.java
@@ -19,14 +19,36 @@
 
 package org.dinky.cdc.convert;
 
+import org.dinky.data.model.Column;
+import org.dinky.data.types.DataTypes;
+
+import org.apache.flink.table.types.logical.LogicalType;
 import org.apache.flink.table.types.logical.TimestampType;
+import org.apache.flink.table.types.logical.VarCharType;
 
 import java.time.ZoneId;
 
 import org.junit.Assert;
+import org.junit.Test;
 
 public class DataTypeConverterTest {
 
+    @Test
+    public void testGetLogicalTypeWithZeroLengthString() {
+        Column column = Column.builder()
+                .name("enum_column")
+                .type("VARCHAR")
+                .isNullable(true)
+                .length(0)
+                .dataType(DataTypes.VARCHAR.toColumnType(true, 0))
+                .build();
+
+        LogicalType logicalType = DataTypeConverter.getLogicalType(column);
+
+        Assert.assertTrue(logicalType instanceof VarCharType);
+        Assert.assertEquals(VarCharType.MAX_LENGTH, ((VarCharType) logicalType).getLength());
+    }
+
     // todo: check the time zone of Timestamp
     //    @Test
     public void testConvertToRowWithTimestamp() {