add default value

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

Hide file tree

Showing 5 changed files with 29 additions and 18 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
@@ -246,12 +246,17 @@ private static void buildColumn(StringBuilder sql, FieldSchema field, boolean is
         if (isKey && DorisType.STRING.equals(fieldType)) {
             fieldType = String.format("%s(%s)", DorisType.VARCHAR, 65533);
         }
-        sql.append(identifier(field.getName()))
-                .append(" ")
-                .append(fieldType)
-                .append(" COMMENT '")
-                .append(quoteComment(field.getComment()))
-                .append("',");
+        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(" COMMENT '").append(quoteComment(field.getComment())).append("',");
     }
 
     public static String quoteComment(String comment) {

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
@@ -111,7 +111,11 @@ public static String buildAddColumnDDL(String tableIdentifier, FieldSchema field
                         DorisSystem.identifier(name),
                         type);
         if (!StringUtils.isNullOrWhitespaceOnly(defaultValue)) {
-            addDDL = addDDL + " DEFAULT '" + defaultValue + "'";
+            // DEFAULT current_timestamp not need quote
+            if (!defaultValue.equalsIgnoreCase("current_timestamp")) {
+                defaultValue = "'" + defaultValue + "'";
+            }
+            addDDL = addDDL + " DEFAULT " + defaultValue;
         }
         if (!StringUtils.isNullOrWhitespaceOnly(comment)) {
             addDDL = addDDL + " COMMENT '" + DorisSystem.quoteComment(comment) + "'";

diff --git a/...pache/doris/flink/sink/writer/serializer/jsondebezium/JsonDebeziumSchemaChangeImplV2.java b/...pache/doris/flink/sink/writer/serializer/jsondebezium/JsonDebeziumSchemaChangeImplV2.java
@@ -369,17 +369,11 @@ private String handleDefaultValue(String defaultValue) {
         if (StringUtils.isNullOrWhitespaceOnly(defaultValue)) {
             return null;
         }
-        // Due to historical reasons, doris needs to add quotes to
-        // the default value of the new column
-        // For example in mysql: alter table add column c1 int default 100
-        // In Doris: alter table add column c1 int default '100'
-        if (Pattern.matches("['\"].*?['\"]", defaultValue)) {
-            return defaultValue;
-        } else if (defaultValue.equals("1970-01-01 00:00:00")) {
+        if (defaultValue.equals("1970-01-01 00:00:00")) {
             // TODO: The default value of setting the current time in CDC is 1970-01-01 00:00:00
             return "current_timestamp";
         }
-        return "'" + defaultValue + "'";
+        return defaultValue;
     }
 
     @VisibleForTesting

diff --git a/flink-doris-connector/src/main/java/org/apache/doris/flink/tools/cdc/SourceSchema.java b/flink-doris-connector/src/main/java/org/apache/doris/flink/tools/cdc/SourceSchema.java
@@ -59,17 +59,19 @@ public SourceSchema(
                 String fieldName = rs.getString("COLUMN_NAME");
                 String comment = rs.getString("REMARKS");
                 String fieldType = rs.getString("TYPE_NAME");
+                String defaultValue = rs.getString("COLUMN_DEF");
                 Integer precision = rs.getInt("COLUMN_SIZE");
-
                 if (rs.wasNull()) {
                     precision = null;
                 }
+
                 Integer scale = rs.getInt("DECIMAL_DIGITS");
                 if (rs.wasNull()) {
                     scale = null;
                 }
                 String dorisTypeStr = convertToDorisType(fieldType, precision, scale);
-                fields.put(fieldName, new FieldSchema(fieldName, dorisTypeStr, comment));
+                fields.put(
+                        fieldName, new FieldSchema(fieldName, dorisTypeStr, defaultValue, comment));
             }
         }
 

diff --git a/...k-doris-connector/src/test/java/org/apache/doris/flink/sink/schema/SchemaManagerTest.java b/...k-doris-connector/src/test/java/org/apache/doris/flink/sink/schema/SchemaManagerTest.java
@@ -114,11 +114,17 @@ public void testAddColumn() {
                 "ALTER TABLE `test`.`test_flink` ADD COLUMN `col` int COMMENT 'comment \"\\'sdf\\''",
                 addColumnDDL);
 
-        field = new FieldSchema("col", "int", "10","comment \"'sdf'");
+        field = new FieldSchema("col", "int", "10", "comment \"'sdf'");
         addColumnDDL = SchemaChangeHelper.buildAddColumnDDL("test.test_flink", field);
         Assert.assertEquals(
                 "ALTER TABLE `test`.`test_flink` ADD COLUMN `col` int DEFAULT '10' COMMENT 'comment \"\\'sdf\\''",
                 addColumnDDL);
+
+        field = new FieldSchema("col", "int", "current_timestamp", "comment \"'sdf'");
+        addColumnDDL = SchemaChangeHelper.buildAddColumnDDL("test.test_flink", field);
+        Assert.assertEquals(
+                "ALTER TABLE `test`.`test_flink` ADD COLUMN `col` int DEFAULT current_timestamp COMMENT 'comment \"\\'sdf\\''",
+                addColumnDDL);
     }
 
     @Test