[feature][connector][cdc] add SeaTunnelRowDebeziumDeserializeSchema (#3499)

* [feature][connector][cdc] add SeaTunnel row Deserialization Converters

* [feature][connector][mysql-cdc] add createDebeziumDeserializationSchema

* [chore] fix checkstyle

* [feature] fill config factory

Author: ashulin

seatunnel-api/src/main/java/org/apache/seatunnel/api/table/catalog/TableSchema.java

@@ -18,6 +18,7 @@
 package org.apache.seatunnel.api.table.catalog;
 
 import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
+import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
 
 import java.io.Serializable;
 import java.util.ArrayList;
@@ -49,6 +50,18 @@ public List<Column> getColumns() {
         return columns;
     }
 
+    public SeaTunnelRowType toPhysicalRowDataType() {
+        SeaTunnelDataType<?>[] fieldTypes = columns.stream()
+            .filter(Column::isPhysical)
+            .map(Column::getDataType)
+            .toArray(SeaTunnelDataType[]::new);
+        String[] fields = columns.stream()
+            .filter(Column::isPhysical)
+            .map(Column::getName)
+            .toArray(String[]::new);
+        return new SeaTunnelRowType(fields, fieldTypes);
+    }
+
     public static final class Builder {
         private final List<Column> columns = new ArrayList<>();
 

seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/config/JdbcSourceConfigFactory.java

@@ -17,11 +17,13 @@
 
 package org.apache.seatunnel.connectors.cdc.base.config;
 
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
 import org.apache.seatunnel.connectors.cdc.base.option.JdbcSourceOptions;
 import org.apache.seatunnel.connectors.cdc.base.option.SourceOptions;
 
 import java.time.Duration;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.List;
 import java.util.Properties;
 
@@ -177,6 +179,26 @@ public JdbcSourceConfigFactory startupOptions(StartupConfig startupConfig) {
         return this;
     }
 
+    public JdbcSourceConfigFactory fromReadonlyConfig(ReadonlyConfig config) {
+        this.port = config.get(JdbcSourceOptions.PORT);
+        this.hostname = config.get(JdbcSourceOptions.HOSTNAME);
+        this.password = config.get(JdbcSourceOptions.PASSWORD);
+        // TODO: support multi-table
+        this.databaseList = Collections.singletonList(config.get(JdbcSourceOptions.DATABASE_NAME));
+        this.tableList = Collections.singletonList(config.get(JdbcSourceOptions.TABLE_NAME));
+        this.distributionFactorUpper = config.get(JdbcSourceOptions.CHUNK_KEY_EVEN_DISTRIBUTION_FACTOR_UPPER_BOUND);
+        this.distributionFactorLower = config.get(JdbcSourceOptions.CHUNK_KEY_EVEN_DISTRIBUTION_FACTOR_LOWER_BOUND);
+        this.splitSize = config.get(SourceOptions.SNAPSHOT_SPLIT_SIZE);
+        this.fetchSize = config.get(SourceOptions.SNAPSHOT_FETCH_SIZE);
+        this.serverTimeZone = config.get(JdbcSourceOptions.SERVER_TIME_ZONE);
+        this.connectTimeout = config.get(JdbcSourceOptions.CONNECT_TIMEOUT);
+        this.connectMaxRetries = config.get(JdbcSourceOptions.CONNECT_MAX_RETRIES);
+        this.connectionPoolSize = config.get(JdbcSourceOptions.CONNECTION_POOL_SIZE);
+        this.dbzProperties = new Properties();
+        config.getOptional(SourceOptions.DEBEZIUM_PROPERTIES).ifPresent(map -> dbzProperties.putAll(map));
+        return this;
+    }
+
     @Override
     public abstract JdbcSourceConfig create(int subtask);
 }

seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/option/JdbcSourceOptions.java

@@ -103,4 +103,26 @@ public class JdbcSourceOptions extends SourceOptions {
                     .defaultValue(3)
                     .withDescription(
                             "The max retry times that the connector should retry to build database server connection.");
+
+    public static final Option<Double> CHUNK_KEY_EVEN_DISTRIBUTION_FACTOR_UPPER_BOUND =
+        Options.key("chunk-key.even-distribution.factor.upper-bound")
+            .doubleType()
+            .defaultValue(1000.0d)
+            .withDescription(
+                "The upper bound of chunk key distribution factor. The distribution factor is used to determine whether the"
+                    + " table is evenly distribution or not."
+                    + " The table chunks would use evenly calculation optimization when the data distribution is even,"
+                    + " and the query for splitting would happen when it is uneven."
+                    + " The distribution factor could be calculated by (MAX(id) - MIN(id) + 1) / rowCount.");
+
+    public static final Option<Double> CHUNK_KEY_EVEN_DISTRIBUTION_FACTOR_LOWER_BOUND =
+        Options.key("chunk-key.even-distribution.factor.lower-bound")
+            .doubleType()
+            .defaultValue(0.05d)
+            .withDescription(
+                "The lower bound of chunk key distribution factor. The distribution factor is used to determine whether the"
+                    + " table is evenly distribution or not."
+                    + " The table chunks would use evenly calculation optimization when the data distribution is even,"
+                    + " and the query for splitting would happen when it is uneven."
+                    + " The distribution factor could be calculated by (MAX(id) - MIN(id) + 1) / rowCount.");
 }

seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/option/SourceOptions.java

@@ -21,6 +21,8 @@
 import org.apache.seatunnel.api.configuration.Options;
 import org.apache.seatunnel.api.configuration.util.OptionRule;
 
+import java.util.Map;
+
 @SuppressWarnings("MagicNumber")
 public class SourceOptions {
 
@@ -83,10 +85,16 @@ public class SourceOptions {
         .noDefaultValue()
         .withDescription("Optional offsets used in case of \"specific\" stop mode");
 
+    public static final Option<Map<String, String>> DEBEZIUM_PROPERTIES = Options.key("debezium")
+        .mapType()
+        .noDefaultValue()
+        .withDescription("Decides if the table options contains Debezium client properties that start with prefix 'debezium'.");
+
     public static final OptionRule.Builder BASE_RULE = OptionRule.builder()
         .optional(SNAPSHOT_SPLIT_SIZE, SNAPSHOT_FETCH_SIZE)
         .optional(INCREMENTAL_PARALLELISM)
         .optional(STARTUP_MODE, STOP_MODE)
+        .optional(DEBEZIUM_PROPERTIES)
         .conditional(STARTUP_MODE, StartupMode.TIMESTAMP, STARTUP_TIMESTAMP)
         .conditional(STARTUP_MODE, StartupMode.SPECIFIC, STARTUP_SPECIFIC_OFFSET_FILE, STARTUP_SPECIFIC_OFFSET_POS)
         .conditional(STOP_MODE, StopMode.TIMESTAMP, STOP_TIMESTAMP)

seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/debezium/DebeziumDeserializationConverter.java

@@ -0,0 +1,30 @@
+/*
+ * 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.seatunnel.connectors.cdc.debezium;
+
+import org.apache.kafka.connect.data.Schema;
+
+import java.io.Serializable;
+
+/**
+ * Runtime converter that converts objects of Debezium into objects of internal data structures.
+ */
+@FunctionalInterface
+public interface DebeziumDeserializationConverter extends Serializable {
+    Object convert(Object dbzObj, Schema schema) throws Exception;
+}

seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/debezium/DebeziumDeserializationConverterFactory.java

@@ -0,0 +1,48 @@
+/*
+ * 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.seatunnel.connectors.cdc.debezium;
+
+import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
+
+import java.io.Serializable;
+import java.time.ZoneId;
+import java.util.Optional;
+
+/**
+ * Factory to create {@link DebeziumDeserializationConverter} according to {@link SeaTunnelDataType}. It's
+ * usually used to create a user-defined {@link DebeziumDeserializationConverter} which has a higher
+ * resolve order than default converter.
+ */
+public interface DebeziumDeserializationConverterFactory extends Serializable {
+
+    /**
+     * A user-defined converter factory which always fallback to default converters.
+     */
+    DebeziumDeserializationConverterFactory DEFAULT =
+        (logicalType, serverTimeZone) -> Optional.empty();
+
+    /**
+     * Returns an optional {@link DebeziumDeserializationConverter}. Returns {@link Optional#empty()}
+     * if fallback to default converter.
+     *
+     * @param type the SeaTunnel datatype to be converted from objects of Debezium
+     * @param serverTimeZone TimeZone used to convert data with timestamp type
+     */
+    Optional<DebeziumDeserializationConverter> createUserDefinedConverter(
+        SeaTunnelDataType<?> type, ZoneId serverTimeZone);
+}

seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/debezium/MetadataConverter.java

@@ -0,0 +1,30 @@
+/*
+ * 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.seatunnel.connectors.cdc.debezium;
+
+import org.apache.kafka.connect.source.SourceRecord;
+
+import java.io.Serializable;
+
+/**
+ * {@link SourceRecord} metadata info converter.
+ */
+@FunctionalInterface
+public interface MetadataConverter extends Serializable {
+    Object read(SourceRecord record);
+}