improve

JNSimba · Jan 24, 2024 · 063604b · 063604b
1 parent 42b85ec
commit 063604b
Show file tree

Hide file tree

Showing 4 changed files with 20 additions and 21 deletions.
diff --git a/flink-doris-connector/src/main/java/org/apache/doris/flink/catalog/doris/DorisSystem.java b/flink-doris-connector/src/main/java/org/apache/doris/flink/catalog/doris/DorisSystem.java
@@ -249,16 +249,19 @@ private static void buildColumn(StringBuilder sql, FieldSchema field, boolean is
         sql.append(identifier(field.getName())).append(" ").append(fieldType);
 
         if (field.getDefaultValue() != null) {
-            String defaultValue = field.getDefaultValue();
-            // DEFAULT current_timestamp not need quote
-            if (!defaultValue.equalsIgnoreCase("current_timestamp")) {
-                defaultValue = "'" + defaultValue + "'";
-            }
-            sql.append(" DEFAULT " + defaultValue);
+            sql.append(" DEFAULT " + quoteDefaultValue(field.getDefaultValue()));
         }
         sql.append(" COMMENT '").append(quoteComment(field.getComment())).append("',");
     }
 
+    public static String quoteDefaultValue(String defaultValue) {
+        // DEFAULT current_timestamp not need quote
+        if (defaultValue.equalsIgnoreCase("current_timestamp")) {
+            return defaultValue;
+        }
+        return "'" + defaultValue + "'";
+    }
+
     public static String quoteComment(String comment) {
         if (comment == null) {
             return "";

diff --git a/...-doris-connector/src/main/java/org/apache/doris/flink/sink/schema/SchemaChangeHelper.java b/...-doris-connector/src/main/java/org/apache/doris/flink/sink/schema/SchemaChangeHelper.java
@@ -110,12 +110,8 @@ public static String buildAddColumnDDL(String tableIdentifier, FieldSchema field
                         DorisSystem.quoteTableIdentifier(tableIdentifier),
                         DorisSystem.identifier(name),
                         type);
-        if (!StringUtils.isNullOrWhitespaceOnly(defaultValue)) {
-            // DEFAULT current_timestamp not need quote
-            if (!defaultValue.equalsIgnoreCase("current_timestamp")) {
-                defaultValue = "'" + defaultValue + "'";
-            }
-            addDDL = addDDL + " DEFAULT " + defaultValue;
+        if (defaultValue != null) {
+            addDDL = addDDL + " DEFAULT " + DorisSystem.quoteDefaultValue(defaultValue);
         }
         if (!StringUtils.isNullOrWhitespaceOnly(comment)) {
             addDDL = addDDL + " COMMENT '" + DorisSystem.quoteComment(comment) + "'";

diff --git a/...is-connector/src/test/java/org/apache/doris/flink/sink/schema/SchemaChangeHelperTest.java b/...is-connector/src/test/java/org/apache/doris/flink/sink/schema/SchemaChangeHelperTest.java
@@ -33,15 +33,15 @@ public class SchemaChangeHelperTest {
 
     @Before
     public void setUp() {
-        originFieldSchemaMap.put("id", new FieldSchema("id", "INT", "", ""));
-        originFieldSchemaMap.put("c2", new FieldSchema("c2", "INT", "", ""));
-        originFieldSchemaMap.put("c3", new FieldSchema("c3", "VARCHAR(30)", "", ""));
+        originFieldSchemaMap.put("id", new FieldSchema("id", "INT", null, ""));
+        originFieldSchemaMap.put("c2", new FieldSchema("c2", "INT", null, ""));
+        originFieldSchemaMap.put("c3", new FieldSchema("c3", "VARCHAR(30)", null, ""));
 
-        updateFieldSchemaMap.put("id", new FieldSchema("id", "INT", "", ""));
-        updateFieldSchemaMap.put("c2", new FieldSchema("c2", "INT", "", ""));
-        updateFieldSchemaMap.put("c3", new FieldSchema("c3", "VARCHAR(30)", "", ""));
-        updateFieldSchemaMap.put("c4", new FieldSchema("c4", "BIGINT", "", ""));
-        updateFieldSchemaMap.put("c5", new FieldSchema("c5", "DATETIMEV2(0)", "", ""));
+        updateFieldSchemaMap.put("id", new FieldSchema("id", "INT", null, ""));
+        updateFieldSchemaMap.put("c2", new FieldSchema("c2", "INT", null, ""));
+        updateFieldSchemaMap.put("c3", new FieldSchema("c3", "VARCHAR(30)", null, ""));
+        updateFieldSchemaMap.put("c4", new FieldSchema("c4", "BIGINT", null, ""));
+        updateFieldSchemaMap.put("c5", new FieldSchema("c5", "DATETIMEV2(0)", null, ""));
     }
 
     @Test

diff --git a/...e/doris/flink/sink/writer/serializer/jsondebezium/TestJsonDebeziumSchemaChangeImplV2.java b/...e/doris/flink/sink/writer/serializer/jsondebezium/TestJsonDebeziumSchemaChangeImplV2.java
@@ -154,7 +154,7 @@ public void testFillOriginSchema() throws IOException {
         srcFiledSchemaMap.put("name", new FieldSchema("name", "VARCHAR(150)", null, null));
         srcFiledSchemaMap.put(
                 "test_time", new FieldSchema("test_time", "DATETIMEV2(0)", null, null));
-        srcFiledSchemaMap.put("c1", new FieldSchema("c1", "INT", "'100'", null));
+        srcFiledSchemaMap.put("c1", new FieldSchema("c1", "INT", "100", null));
 
         schemaChange.setSourceConnector("mysql");
         String columnsString =