[FLINK-34638][cdc-common] Support column with default value

flink-cdc-common/src/main/java/org/apache/flink/cdc/common/schema/Column.java

@@ -38,9 +38,16 @@ public abstract class Column implements Serializable {
 
     private static final long serialVersionUID = 1L;
 
-    protected static final String FIELD_FORMAT_WITH_DESCRIPTION = "%s %s '%s'";
+    protected static final String FIELD_FORMAT_WITH_DESCRIPTION_NO_DEFAULT_VALUE_EXPRESSION =
+            "%s %s '%s'";
 
-    protected static final String FIELD_FORMAT_NO_DESCRIPTION = "%s %s";
+    protected static final String FIELD_FORMAT_NO_DESCRIPTION_WITH_DEFAULT_VALUE_EXPRESSION =
+            "%s %s '%s'";
+
+    protected static final String FIELD_FORMAT_WITH_DESCRIPTION_WITH_DEFAULT_VALUE_EXPRESSION =
+            "%s %s '%s' '%s'";
+
+    protected static final String FIELD_FORMAT_NO_DESCRIPTION_NO_DEFAULT_VALUE_EXPRESSION = "%s %s";
 
     protected final String name;
 
@@ -48,10 +55,29 @@ public abstract class Column implements Serializable {
 
     protected final @Nullable String comment;
 
+    /**
+     * Save the literal value of the column's default value, For uncertain functions such as UUID(),
+     * the value is null, For the current time function such as CURRENT_TIMESTAMP(), the value is
+     * Unix Epoch time(1970-01-01 00:00:00).
+     */
+    protected final @Nullable String defaultValueExpression;
+
     protected Column(String name, DataType type, @Nullable String comment) {
         this.name = name;
         this.type = type;
         this.comment = comment;
+        this.defaultValueExpression = null;
+    }
+
+    protected Column(
+            String name,
+            DataType type,
+            @Nullable String comment,
+            @Nullable String defaultValueExpression) {
+        this.name = name;
+        this.type = type;
+        this.comment = comment;
+        this.defaultValueExpression = defaultValueExpression;
     }
 
     /** Returns the name of this column. */
@@ -69,17 +95,41 @@ public abstract class Column implements Serializable {
         return comment;
     }
 
+    @Nullable
+    public String getDefaultValueExpression() {
+        return defaultValueExpression;
+    }
+
     /** Returns a string that summarizes this column for printing to a console. */
     public String asSummaryString() {
         if (comment == null) {
-            return String.format(
+            if (defaultValueExpression == null) {
-                    FIELD_FORMAT_NO_DESCRIPTION, escapeIdentifier(name), type.asSummaryString());
+                return String.format(
+                        FIELD_FORMAT_NO_DESCRIPTION_NO_DEFAULT_VALUE_EXPRESSION,
+                        escapeIdentifier(name),
+                        type.asSummaryString());
+            } else {
+                return String.format(
+                        FIELD_FORMAT_NO_DESCRIPTION_WITH_DEFAULT_VALUE_EXPRESSION,
+                        escapeIdentifier(name),
+                        type.asSummaryString(),
+                        defaultValueExpression);
+            }
         } else {
-            return String.format(
+            if (defaultValueExpression == null) {
-                    FIELD_FORMAT_WITH_DESCRIPTION,
+                return String.format(
-                    escapeIdentifier(name),
+                        FIELD_FORMAT_WITH_DESCRIPTION_NO_DEFAULT_VALUE_EXPRESSION,
-                    type.asSummaryString(),
+                        escapeIdentifier(name),
-                    escapeSingleQuotes(comment));
+                        type.asSummaryString(),
+                        escapeSingleQuotes(comment));
+            } else {
+                return String.format(
+                        FIELD_FORMAT_WITH_DESCRIPTION_WITH_DEFAULT_VALUE_EXPRESSION,
+                        escapeIdentifier(name),
+                        type.asSummaryString(),
+                        escapeSingleQuotes(comment),
+                        defaultValueExpression);
+            }
         }
     }
 
@@ -103,12 +153,13 @@ public abstract class Column implements Serializable {
         Column column = (Column) o;
         return name.equals(column.name)
                 && type.equals(column.type)
-                && Objects.equals(comment, column.comment);
+                && Objects.equals(comment, column.comment)
+                && Objects.equals(defaultValueExpression, column.defaultValueExpression);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(name, type, comment);
+        return Objects.hash(name, type, comment, defaultValueExpression);
     }
 
     @Override
@@ -116,6 +167,15 @@ public abstract class Column implements Serializable {
         return asSummaryString();
     }
 
+    /** Creates a physical column. */
+    public static PhysicalColumn physicalColumn(
+            String name,
+            DataType type,
+            @Nullable String comment,
+            @Nullable String defaultValueExpression) {
+        return new PhysicalColumn(name, type, comment, defaultValueExpression);
+    }
+
     /** Creates a physical column. */
     public static PhysicalColumn physicalColumn(
             String name, DataType type, @Nullable String comment) {

flink-cdc-common/src/main/java/org/apache/flink/cdc/common/schema/PhysicalColumn.java

@@ -32,6 +32,11 @@ public class PhysicalColumn extends Column {
         super(name, type, comment);
     }
 
+    public PhysicalColumn(
+            String name, DataType type, @Nullable String comment, @Nullable String defaultValue) {
+        super(name, type, comment, defaultValue);
+    }
+
     @Override
     public boolean isPhysical() {
         return true;
@@ -39,11 +44,11 @@ public class PhysicalColumn extends Column {
 
     @Override
     public Column copy(DataType newType) {
-        return new PhysicalColumn(name, newType, comment);
+        return new PhysicalColumn(name, newType, comment, defaultValueExpression);
     }
 
     @Override
     public Column copy(String newName) {
-        return new PhysicalColumn(newName, type, comment);
+        return new PhysicalColumn(newName, type, comment, defaultValueExpression);
     }
 }

flink-cdc-common/src/main/java/org/apache/flink/cdc/common/schema/Schema.java

@@ -316,6 +316,21 @@ public class Schema implements Serializable {
             return this;
         }
 
+        /**
+         * Declares a physical column that is appended to this schema.
+         *
+         * @param columnName column name
+         * @param type data type of the column
+         * @param comment description of the column
+         * @param defaultValue default value of the column
+         */
+        public Builder physicalColumn(
+                String columnName, DataType type, String comment, String defaultValue) {
+            checkColumn(columnName, type);
+            columns.add(Column.physicalColumn(columnName, type, comment, defaultValue));
+            return this;
+        }
+
         /**
          * Declares a metadata column that is appended to this schema.
          *

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-doris/src/main/java/org/apache/flink/cdc/connectors/doris/sink/DorisMetadataApplier.java

@@ -134,7 +134,11 @@ public class DorisMetadataApplier implements MetadataApplier {
             }
             fieldSchemaMap.put(
                     column.getName(),
-                    new FieldSchema(column.getName(), typeString, column.getComment()));
+                    new FieldSchema(
+                            column.getName(),
+                            typeString,
+                            column.getDefaultValueExpression(),
+                            column.getComment()));
         }
         return fieldSchemaMap;
     }
@@ -170,6 +174,7 @@ public class DorisMetadataApplier implements MetadataApplier {
                     new FieldSchema(
                             column.getName(),
                             buildTypeString(column.getType()),
+                            column.getDefaultValueExpression(),
                             column.getComment());
             schemaChangeManager.addColumn(
                     tableId.getSchemaName(), tableId.getTableName(), addFieldSchema);

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/parser/CustomAlterTableParserListener.java

@@ -253,7 +253,10 @@ public class CustomAlterTableParserListener extends MySqlParserBaseListener {
 
     private org.apache.flink.cdc.common.schema.Column toCdcColumn(Column dbzColumn) {
         return org.apache.flink.cdc.common.schema.Column.physicalColumn(
-                dbzColumn.name(), fromDbzColumn(dbzColumn), dbzColumn.comment());
+                dbzColumn.name(),
+                fromDbzColumn(dbzColumn),
+                dbzColumn.comment(),
+                dbzColumn.defaultValueExpression().orElse(null));
     }
 
     private org.apache.flink.cdc.common.event.TableId toCdcTableId(TableId dbzTableId) {

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/reader/MySqlPipelineRecordEmitter.java

@@ -205,7 +205,11 @@ public class MySqlPipelineRecordEmitter extends MySqlRecordEmitter<Event> {
             if (!column.isOptional()) {
                 dataType = dataType.notNull();
             }
-            tableBuilder.physicalColumn(colName, dataType, column.comment());
+            tableBuilder.physicalColumn(
+                    colName,
+                    dataType,
+                    column.comment(),
+                    column.defaultValueExpression().orElse(null));
         }
 
         List<String> primaryKey = table.primaryKeyColumnNames();

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlPipelineITCase.java

@@ -355,7 +355,7 @@ public class MySqlPipelineITCase extends MySqlSourceTestBase {
                 tableId,
                 Schema.newBuilder()
                         .physicalColumn("id", DataTypes.INT().notNull())
-                        .physicalColumn("name", DataTypes.VARCHAR(255).notNull())
+                        .physicalColumn("name", DataTypes.VARCHAR(255).notNull(), null, "flink")
                         .physicalColumn("description", DataTypes.VARCHAR(512))
                         .physicalColumn("weight", DataTypes.FLOAT())
                         .primaryKey(Collections.singletonList("id"))

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/main/java/org/apache/flink/cdc/connectors/starrocks/sink/StarRocksMetadataApplier.java

@@ -37,6 +37,8 @@ import org.slf4j.LoggerFactory;
 import java.util.ArrayList;
 import java.util.List;
 
+import static org.apache.flink.cdc.connectors.starrocks.sink.StarRocksUtils.toStarRocksDataType;
+
 /** A {@code MetadataApplier} that applies metadata changes to StarRocks. */
 public class StarRocksMetadataApplier implements MetadataApplier {
 
@@ -117,8 +119,9 @@ public class StarRocksMetadataApplier implements MetadataApplier {
                     new StarRocksColumn.Builder()
                             .setColumnName(column.getName())
                             .setOrdinalPosition(-1)
-                            .setColumnComment(column.getComment());
+                            .setColumnComment(column.getComment())
-            StarRocksUtils.toStarRocksDataType(column, false, builder);
+                            .setDefaultValue(column.getDefaultValueExpression());
+            toStarRocksDataType(column, false, builder);
             addColumns.add(builder.build());
         }
 

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/main/java/org/apache/flink/cdc/connectors/starrocks/sink/StarRocksUtils.java

@@ -84,7 +84,8 @@ public class StarRocksUtils {
                     new StarRocksColumn.Builder()
                             .setColumnName(column.getName())
                             .setOrdinalPosition(i)
-                            .setColumnComment(column.getComment());
+                            .setColumnComment(column.getComment())
+                            .setDefaultValue(column.getDefaultValueExpression());
             toStarRocksDataType(column, i < primaryKeyCount, builder);
             starRocksColumns.add(builder.build());
         }

flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/coordinator/SchemaManager.java

@@ -172,7 +172,11 @@ public class SchemaManager {
     /** Serializer for {@link SchemaManager}. */
     public static class Serializer implements SimpleVersionedSerializer<SchemaManager> {
 
-        public static final int CURRENT_VERSION = 1;
+        /**
+         * Update history: from Version 3.0.0, set to 0, from version 3.1.1, updated to 1, from
+         * version 3.2.0, updated to 2.
+         */
+        public static final int CURRENT_VERSION = 2;
 
         @Override
         public int getVersion() {
@@ -214,6 +218,7 @@ public class SchemaManager {
             switch (version) {
                 case 0:
                 case 1:
+                case 2:
                     TableIdSerializer tableIdSerializer = TableIdSerializer.INSTANCE;
                     SchemaSerializer schemaSerializer = SchemaSerializer.INSTANCE;
                     try (ByteArrayInputStream bais = new ByteArrayInputStream(serialized);

flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/coordinator/SchemaRegistry.java

@@ -214,6 +214,7 @@ public class SchemaRegistry implements OperatorCoordinator, CoordinationRequestH
                         break;
                     }
                 case 1:
+                case 2:
                     {
                         int length = in.readInt();
                         byte[] serializedSchemaManager = new byte[length];

flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TableChangeInfo.java

@@ -109,7 +109,7 @@ public class TableChangeInfo {
     /** Serializer for {@link TableChangeInfo}. */
     public static class Serializer implements SimpleVersionedSerializer<TableChangeInfo> {
 
-        public static final int CURRENT_VERSION = 1;
+        public static final int CURRENT_VERSION = 2;
 
         @Override
         public int getVersion() {

flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/schema/ColumnSerializer.java

@@ -46,6 +46,13 @@ public class ColumnSerializer extends TypeSerializerSingleton<Column> {
     private final MetadataColumnSerializer metadataColumnSerializer =
             MetadataColumnSerializer.INSTANCE;
 
+    private static int currentVersion = 2;
+
+    /** Update {@link #currentVersion} as We did not directly include this version in the file. */
+    public static void updateVersion(int version) {
+        currentVersion = version;
+    }
+
     @Override
     public boolean isImmutableType() {
         return false;
@@ -92,12 +99,16 @@ public class ColumnSerializer extends TypeSerializerSingleton<Column> {
 
     @Override
     public Column deserialize(DataInputView source) throws IOException {
+        return deserialize(currentVersion, source);
+    }
+
+    public Column deserialize(int version, DataInputView source) throws IOException {
         ColumnType columnType = enumSerializer.deserialize(source);
         switch (columnType) {
             case METADATA:
                 return metadataColumnSerializer.deserialize(source);
             case PHYSICAL:
-                return physicalColumnSerializer.deserialize(source);
+                return physicalColumnSerializer.deserialize(version, source);
             default:
                 throw new IOException("Unknown column type: " + columnType);
         }

flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/schema/PhysicalColumnSerializer.java

@@ -42,6 +42,8 @@ public class PhysicalColumnSerializer extends TypeSerializerSingleton<PhysicalCo
     private final DataTypeSerializer dataTypeSerializer = new DataTypeSerializer();
     private final StringSerializer stringSerializer = StringSerializer.INSTANCE;
 
+    private static final int CURRENT_VERSION = 2;
+
     @Override
     public boolean isImmutableType() {
         return false;
@@ -57,7 +59,8 @@ public class PhysicalColumnSerializer extends TypeSerializerSingleton<PhysicalCo
         return Column.physicalColumn(
                 stringSerializer.copy(from.getName()),
                 dataTypeSerializer.copy(from.getType()),
-                stringSerializer.copy(from.getComment()));
+                stringSerializer.copy(from.getComment()),
+                stringSerializer.copy(from.getDefaultValueExpression()));
     }
 
     @Override
@@ -75,14 +78,37 @@ public class PhysicalColumnSerializer extends TypeSerializerSingleton<PhysicalCo
         stringSerializer.serialize(record.getName(), target);
         dataTypeSerializer.serialize(record.getType(), target);
         stringSerializer.serialize(record.getComment(), target);
+        stringSerializer.serialize(record.getDefaultValueExpression(), target);
     }
 
     @Override
     public PhysicalColumn deserialize(DataInputView source) throws IOException {
-        String name = stringSerializer.deserialize(source);
+        return deserialize(CURRENT_VERSION, source);
-        DataType dataType = dataTypeSerializer.deserialize(source);
+    }
-        String comment = stringSerializer.deserialize(source);
+
-        return Column.physicalColumn(name, dataType, comment);
+    public PhysicalColumn deserialize(int version, DataInputView source) throws IOException {
+        switch (version) {
+            case 0:
+            case 1:
+                {
+                    String name = stringSerializer.deserialize(source);
+                    DataType dataType = dataTypeSerializer.deserialize(source);
+                    String comment = stringSerializer.deserialize(source);
+                    return Column.physicalColumn(name, dataType, comment);
+                }
+            case 2:
+                {
+                    String name = stringSerializer.deserialize(source);
+                    DataType dataType = dataTypeSerializer.deserialize(source);
+                    String comment = stringSerializer.deserialize(source);
+                    String defaultValue = stringSerializer.deserialize(source);
+                    return Column.physicalColumn(name, dataType, comment, defaultValue);
+                }
+            default:
+                {
+                    throw new IOException("Unrecognized serialization version " + version);
+                }
+        }
     }
 
     @Override

flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/schema/SchemaSerializer.java

@@ -89,7 +89,7 @@ public class SchemaSerializer extends TypeSerializerSingleton<Schema> {
         stringSerializer.serialize(record.comment(), target);
     }
 
-    private static final int CURRENT_VERSION = 1;
+    private static final int CURRENT_VERSION = 2;
 
     @Override
     public Schema deserialize(DataInputView source) throws IOException {
@@ -97,6 +97,9 @@ public class SchemaSerializer extends TypeSerializerSingleton<Schema> {
     }
 
     public Schema deserialize(int version, DataInputView source) throws IOException {
+        // Manually updating versions because column deserialization is wrapped by
+        // ListSerializer.
+        ColumnSerializer.updateVersion(version);
         switch (version) {
             case 0:
                 return Schema.newBuilder()
@@ -106,6 +109,7 @@ public class SchemaSerializer extends TypeSerializerSingleton<Schema> {
                         .comment(stringSerializer.deserialize(source))
                         .build();
             case 1:
+            case 2:
                 return Schema.newBuilder()
                         .setColumns(columnsSerializer.deserialize(source))
                         .primaryKey(primaryKeysSerializer.deserialize(source))

flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/serializer/schema/PhysicalColumnSerializerTest.java

@@ -44,7 +44,8 @@ public class PhysicalColumnSerializerTest extends SerializerTestBase<PhysicalCol
     protected PhysicalColumn[] getTestData() {
         return new PhysicalColumn[] {
             Column.physicalColumn("col1", DataTypes.BIGINT()),
-            Column.physicalColumn("col1", DataTypes.BIGINT(), "comment")
+            Column.physicalColumn("col1", DataTypes.BIGINT(), "comment"),
+            Column.physicalColumn("col1", DataTypes.BIGINT(), "comment", "default value")
         };
     }
 }