[cdc] Update flink.cdc.version to 3.5.0 and add FlinkCDC to Paimon converter.

paimon-flink/paimon-flink-cdc/pom.xml

@@ -35,8 +35,8 @@ under the License.
 
     <properties>
         <flink.version>1.20.1</flink.version>
-        <flink.cdc.version>3.1.1</flink.cdc.version>
-        <flink.mongodb.cdc.version>3.1.1</flink.mongodb.cdc.version>
+        <flink.cdc.version>3.5.0</flink.cdc.version>
+        <flink.mongodb.cdc.version>3.5.0</flink.mongodb.cdc.version>
         <avro.version>1.11.4</avro.version>
         <geometry.version>2.2.0</geometry.version>
         <json-path.version>2.9.0</json-path.version>

paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/postgres/PostgresActionUtils.java

@@ -186,7 +186,7 @@ public static JdbcIncrementalSource<CdcSourceRecord> buildPostgresSource(
         customConverterConfigs.put(JsonConverterConfig.DECIMAL_FORMAT_CONFIG, "numeric");
         CdcDebeziumDeserializationSchema schema =
                 new CdcDebeziumDeserializationSchema(true, customConverterConfigs);
-        return sourceBuilder.deserializer(schema).includeSchemaChanges(true).build();
+        return sourceBuilder.deserializer(schema).build();
     }
 
     public static void registerJdbcDriver() {

paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/schema/SchemaChangeProvider.java

@@ -0,0 +1,152 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.pipeline.cdc.schema;
+
+import org.apache.paimon.flink.pipeline.cdc.util.FlinkCDCToPaimonTypeConverter;
+import org.apache.paimon.schema.SchemaChange;
+
+import org.apache.flink.cdc.common.event.AddColumnEvent;
+import org.apache.flink.cdc.common.schema.Column;
+import org.apache.flink.cdc.common.types.DataType;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+/**
+ * The SchemaChangeProvider class provides static methods to create SchemaChange objects that
+ * represent different types of schema modifications.
+ */
+public class SchemaChangeProvider {
+
+    /**
+     * Creates a SchemaChange object for adding a column without specifying its position.
+     *
+     * @param columnWithPosition The ColumnWithPosition object containing the column details and its
+     *     intended position within the schema.
+     * @return A SchemaChange object representing the addition of a column.
+     */
+    public static List<SchemaChange> add(AddColumnEvent.ColumnWithPosition columnWithPosition) {
+        List<SchemaChange> result = new ArrayList<>();
+        result.add(
+                SchemaChange.addColumn(
+                        columnWithPosition.getAddColumn().getName(),
+                        FlinkCDCToPaimonTypeConverter.convertFlinkCDCDataTypeToPaimonDataType(
+                                columnWithPosition.getAddColumn().getType()),
+                        columnWithPosition.getAddColumn().getComment()));
+        // if default value express exists, we need to set the default value to the table
+        // option
+        Column column = columnWithPosition.getAddColumn();
+        Optional.ofNullable(
+                        FlinkCDCToPaimonTypeConverter.convertFlinkCDCDefaultValueToValidValue(
+                                column.getDefaultValueExpression(), column.getType()))
+                .ifPresent(
+                        value -> {
+                            result.add(
+                                    SchemaChange.updateColumnDefaultValue(
+                                            new String[] {column.getName()}, value));
+                        });
+        return result;
+    }
+
+    /**
+     * Creates a SchemaChange object for adding a column with a specified position.
+     *
+     * @param columnWithPosition The ColumnWithPosition object containing the column details and its
+     *     intended position within the schema.
+     * @param move The move operation to indicate the column's new position.
+     * @return A SchemaChange object representing the addition of a column with position
+     *     information.
+     */
+    public static List<SchemaChange> add(
+            AddColumnEvent.ColumnWithPosition columnWithPosition, SchemaChange.Move move) {
+        List<SchemaChange> result = new ArrayList<>();
+        result.add(
+                SchemaChange.addColumn(
+                        columnWithPosition.getAddColumn().getName(),
+                        FlinkCDCToPaimonTypeConverter.convertFlinkCDCDataTypeToPaimonDataType(
+                                columnWithPosition.getAddColumn().getType()),
+                        columnWithPosition.getAddColumn().getComment(),
+                        move));
+        // if default value express exists, we need to set the default value to the table
+        // option
+        Column column = columnWithPosition.getAddColumn();
+        Optional.ofNullable(
+                        FlinkCDCToPaimonTypeConverter.convertFlinkCDCDefaultValueToValidValue(
+                                column.getDefaultValueExpression(), column.getType()))
+                .ifPresent(
+                        value -> {
+                            result.add(
+                                    SchemaChange.updateColumnDefaultValue(
+                                            new String[] {column.getName()}, value));
+                        });
+        return result;
+    }
+
+    /**
+     * Creates a SchemaChange object to update the data type of a column.
+     *
+     * @param oldColumnName The name of the column whose data type is to be updated.
+     * @param newType The new DataType for the column.
+     * @return A SchemaChange object representing the update of the column's data type.
+     */
+    public static SchemaChange updateColumnType(String oldColumnName, DataType newType) {
+        return SchemaChange.updateColumnType(
+                oldColumnName,
+                FlinkCDCToPaimonTypeConverter.convertFlinkCDCDataTypeToPaimonDataType(newType));
+    }
+
+    /**
+     * Creates a SchemaChange object for renaming a column.
+     *
+     * @param oldColumnName The current name of the column to be renamed.
+     * @param newColumnName The new name for the column.
+     * @return A SchemaChange object representing the renaming of a column.
+     */
+    public static List<SchemaChange> rename(
+            String oldColumnName, String newColumnName, Map<String, String> options) {
+        List<SchemaChange> result = new ArrayList<>();
+        result.add(SchemaChange.renameColumn(oldColumnName, newColumnName));
+        return result;
+    }
+
+    /**
+     * Creates a SchemaChange object for dropping a column.
+     *
+     * @param columnName The name of the column to be dropped.
+     * @return A SchemaChange object representing the deletion of a column.
+     */
+    public static List<SchemaChange> drop(String columnName) {
+        List<SchemaChange> result = new ArrayList<>();
+        result.add(SchemaChange.dropColumn(columnName));
+        return result;
+    }
+
+    /**
+     * Creates a SchemaChange object for setting an option.
+     *
+     * @param key The key of the option to be set.
+     * @param value The value of the option to be set.
+     * @return A SchemaChange object representing the setting of an option.
+     */
+    public static SchemaChange setOption(String key, String value) {
+        return SchemaChange.setOption(key, value);
+    }
+}