[Feature][Connector-V2]Jdbc chunk split add “snapshot.split.column” params #7794

docs/en/connector-v2/source/MySQL-CDC.md

@@ -169,7 +169,7 @@ When an initial consistent snapshot is made for large databases, your establishe
 
 ## Source Options
 
-|                      Name                      |   Type   | Required | Default | Description                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                          |
+| Name                                           | Type     | Required | Default | Description                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                          |
 |------------------------------------------------|----------|----------|---------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
 | base-url                                       | String   | Yes      | -       | The URL of the JDBC connection. Refer to a case: `jdbc:mysql://localhost:3306:3306/test`.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                            |
 | username                                       | String   | Yes      | -       | Name of the database to use when connecting to the database server.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                  |
@@ -184,6 +184,7 @@ When an initial consistent snapshot is made for large databases, your establishe
 | stop.specific-offset.file                      | String   | No       | -       | Stop from the specified binlog file name. **Note, This option is required when the `stop.mode` option used `specific`.**                                                                                                                                                                                                                                                                                                                                                                                                                                                                                             |
 | stop.specific-offset.pos                       | Long     | No       | -       | Stop from the specified binlog file position. **Note, This option is required when the `stop.mode` option used `specific`.**                                                                                                                                                                                                                                                                                                                                                                                                                                                                                         |
 | snapshot.split.size                            | Integer  | No       | 8096    | The split size (number of rows) of table snapshot, captured tables are split into multiple splits when read the snapshot of table.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                   |
+| snapshot.split.column                          | Config   | No       | -       | The table name and split column (must be unique key) of table snapshot, captured tables are split into multiple splits when read the snapshot of table.                                                                                                                                                                                                                                                                                                                                                                                                                                                              |
 | snapshot.fetch.size                            | Integer  | No       | 1024    | The maximum fetch size for per poll when read table snapshot.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                        |
 | server-id                                      | String   | No       | -       | A numeric ID or a numeric ID range of this database client, The numeric ID syntax is like `5400`, the numeric ID range syntax is like '5400-5408'. <br/> Every ID must be unique across all currently-running database processes in the MySQL cluster. This connector joins the <br/> MySQL cluster as another server (with this unique ID) so it can read the binlog. <br/> By default, a random number is generated between 6500 and 2,148,492,146, though we recommend setting an explicit value.                                                                                                                 |
 | server-time-zone                               | String   | No       | UTC     | The session time zone in database server. If not set, then ZoneId.systemDefault() is used to determine the server time zone.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                         |

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

@@ -34,6 +34,7 @@ public abstract class BaseSourceConfig implements SourceConfig {
     @Getter protected final StopConfig stopConfig;
 
     @Getter protected final int splitSize;
+    @Getter protected final Properties splitColumn;
 
     @Getter protected final double distributionFactorUpper;
     @Getter protected final double distributionFactorLower;
@@ -50,6 +51,7 @@ public BaseSourceConfig(
             StartupConfig startupConfig,
             StopConfig stopConfig,
             int splitSize,
+            Properties splitColumn,
             double distributionFactorUpper,
             double distributionFactorLower,
             int sampleShardingThreshold,
@@ -59,6 +61,7 @@ public BaseSourceConfig(
         this.startupConfig = startupConfig;
         this.stopConfig = stopConfig;
         this.splitSize = splitSize;
+        this.splitColumn = splitColumn;
         this.distributionFactorUpper = distributionFactorUpper;
         this.distributionFactorLower = distributionFactorLower;
         this.sampleShardingThreshold = sampleShardingThreshold;

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

@@ -49,6 +49,7 @@ public JdbcSourceConfig(
             List<String> databaseList,
             List<String> tableList,
             int splitSize,
+            Properties splitColumn,
             double distributionFactorUpper,
             double distributionFactorLower,
             int sampleShardingThreshold,
@@ -70,6 +71,7 @@ public JdbcSourceConfig(
                 startupConfig,
                 stopConfig,
                 splitSize,
+                splitColumn,
                 distributionFactorUpper,
                 distributionFactorLower,
                 sampleShardingThreshold,

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

@@ -51,6 +51,7 @@ public abstract class JdbcSourceConfigFactory implements SourceConfig.Factory<Jd
             JdbcSourceOptions.SAMPLE_SHARDING_THRESHOLD.defaultValue();
     protected int inverseSamplingRate = JdbcSourceOptions.INVERSE_SAMPLING_RATE.defaultValue();
     protected int splitSize = SourceOptions.SNAPSHOT_SPLIT_SIZE.defaultValue();
+    protected Properties splitColumn;
     protected int fetchSize = SourceOptions.SNAPSHOT_FETCH_SIZE.defaultValue();
     protected String serverTimeZone = JdbcSourceOptions.SERVER_TIME_ZONE.defaultValue();
     protected long connectTimeoutMillis = JdbcSourceOptions.CONNECT_TIMEOUT_MS.defaultValue();
@@ -65,6 +66,11 @@ public JdbcSourceConfigFactory hostname(String hostname) {
         return this;
     }
 
+    public JdbcSourceConfigFactory splitColumn(Properties splitColumn) {
+        this.splitColumn = splitColumn;
+        return this;
+    }
+
     /** Integer port number of the database server. */
     public JdbcSourceConfigFactory port(int port) {
         this.port = port;
@@ -239,6 +245,9 @@ public JdbcSourceConfigFactory fromReadonlyConfig(ReadonlyConfig config) {
         this.sampleShardingThreshold = config.get(JdbcSourceOptions.SAMPLE_SHARDING_THRESHOLD);
         this.inverseSamplingRate = config.get(JdbcSourceOptions.INVERSE_SAMPLING_RATE);
         this.splitSize = config.get(SourceOptions.SNAPSHOT_SPLIT_SIZE);
+        this.splitColumn = new Properties();
+        config.getOptional(SourceOptions.SNAPSHOT_SPLIT_COLUMN)
+                .ifPresent(map -> splitColumn.putAll(map));
         this.fetchSize = config.get(SourceOptions.SNAPSHOT_FETCH_SIZE);
         this.serverTimeZone = config.get(JdbcSourceOptions.SERVER_TIME_ZONE);
         this.connectTimeoutMillis = config.get(JdbcSourceOptions.CONNECT_TIMEOUT_MS);

seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/dialect/JdbcDataSourceDialect.java

@@ -119,6 +119,26 @@ default List<ConstraintKey> getUniqueKeys(JdbcConnection jdbcConnection, TableId
                 .collect(Collectors.toList());
     }
 
+    default Boolean isUniqueKey(JdbcConnection jdbcConnection, TableId tableId, String columnName)
+            throws SQLException {
+        boolean isUnique = false;
+        if (null != columnName) {
+            DatabaseMetaData metaData = jdbcConnection.connection().getMetaData();
+            ResultSet resultSet =
+                    metaData.getIndexInfo(
+                            tableId.catalog(), tableId.schema(), tableId.table(), false, false);
+
+            while (resultSet.next()) {
+                if (columnName.equalsIgnoreCase(resultSet.getString("COLUMN_NAME"))
+                        && resultSet.getBoolean("NON_UNIQUE") == false) {
+                    isUnique = true;
+                    break;
+                }
+            }
+        }
+        return isUnique;
+    }
+
     default List<ConstraintKey> getConstraintKeys(JdbcConnection jdbcConnection, TableId tableId)
             throws SQLException {
         DatabaseMetaData metaData = jdbcConnection.connection().getMetaData();

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

@@ -37,6 +37,13 @@ public class SourceOptions {
                     .withDescription(
                             "The split size (number of rows) of table snapshot, captured tables are split into multiple splits when read the snapshot of table.");
 
+    public static final Option<Map<String, String>> SNAPSHOT_SPLIT_COLUMN =
+            Options.key("snapshot.split.column")
+                    .mapType()
+                    .noDefaultValue()
+                    .withDescription(
+                            "The split column of table snapshot, captured tables are split into multiple splits when read the snapshot of table.");
+
     public static final Option<Integer> SNAPSHOT_FETCH_SIZE =
             Options.key("snapshot.fetch.size")
                     .intType()
@@ -111,7 +118,7 @@ public class SourceOptions {
     public static OptionRule.Builder getBaseRule() {
         return OptionRule.builder()
                 .optional(FORMAT)
-                .optional(SNAPSHOT_SPLIT_SIZE, SNAPSHOT_FETCH_SIZE)
+                .optional(SNAPSHOT_SPLIT_SIZE, SNAPSHOT_FETCH_SIZE, SNAPSHOT_SPLIT_COLUMN)
                 .optional(INCREMENTAL_PARALLELISM)
                 .optional(DEBEZIUM_PROPERTIES);
     }

seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/splitter/AbstractJdbcSourceChunkSplitter.java

@@ -39,6 +39,7 @@
 import java.util.List;
 import java.util.Objects;
 import java.util.Optional;
+import java.util.Properties;
 
 import static java.math.BigDecimal.ROUND_CEILING;
 import static org.apache.seatunnel.connectors.cdc.base.utils.ObjectUtils.doubleCompare;
@@ -379,12 +380,38 @@ protected SnapshotSplit createSnapshotSplit(
     protected Column getSplitColumn(
             JdbcConnection jdbc, JdbcDataSourceDialect dialect, TableId tableId)
             throws SQLException {
-        Optional<PrimaryKey> primaryKey = dialect.getPrimaryKey(jdbc, tableId);
         Column splitColumn = null;
+        Table table = dialect.queryTableSchema(jdbc, tableId).getTable();
+
+        // first , compare user defined split column is in the primary key or unique key
+        Properties splitColumnProperties = new Properties();
+        try {
+            splitColumnProperties = sourceConfig.getSplitColumn();
+        } catch (Exception e) {
+            log.error("Config snapshot.split.column get exception in {}:{}", tableId, e);
+        }
+        String tableSc =
+                (String) splitColumnProperties.get(tableId.catalog() + "." + tableId.table());
+        Boolean isUniqueKey = dialect.isUniqueKey(jdbc, tableId, tableSc);
+        if (isUniqueKey) {
+            Column column = table.columnWithName(tableSc);
+            if (isEvenlySplitColumn(column)) {
+                return column;
+            } else {
+                log.warn(
+                        "Config snapshot.split.column type in {} is not TINYINT、SMALLINT、INT、BIGINT、DECIMAL、STRING",
+                        tableId);
+            }
+        } else {
+            log.warn(
+                    "Config snapshot.split.column not exists or not unique key for table {}",
+                    tableId);
+        }
+
+        Optional<PrimaryKey> primaryKey = dialect.getPrimaryKey(jdbc, tableId);
         if (primaryKey.isPresent()) {
             List<String> pkColumns = primaryKey.get().getColumnNames();
 
-            Table table = dialect.queryTableSchema(jdbc, tableId).getTable();
             for (String pkColumn : pkColumns) {
                 Column column = table.columnWithName(pkColumn);
                 if (isEvenlySplitColumn(column)) {
@@ -400,7 +427,6 @@ protected Column getSplitColumn(
 
         List<ConstraintKey> uniqueKeys = dialect.getUniqueKeys(jdbc, tableId);
         if (!uniqueKeys.isEmpty()) {
-            Table table = dialect.queryTableSchema(jdbc, tableId).getTable();
             for (ConstraintKey uniqueKey : uniqueKeys) {
                 List<ConstraintKey.ConstraintKeyColumn> uniqueKeyColumns =
                         uniqueKey.getColumnNames();

seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/config/MySqlSourceConfig.java

@@ -41,6 +41,7 @@ public MySqlSourceConfig(
             List<String> databaseList,
             List<String> tableList,
             int splitSize,
+            Properties splitColumn,
             double distributionFactorUpper,
             double distributionFactorLower,
             int sampleShardingThreshold,
@@ -64,6 +65,7 @@ public MySqlSourceConfig(
                 databaseList,
                 tableList,
                 splitSize,
+                splitColumn,
                 distributionFactorUpper,
                 distributionFactorLower,
                 sampleShardingThreshold,

seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/config/MySqlSourceConfigFactory.java

@@ -116,6 +116,7 @@ public MySqlSourceConfig create(int subtaskId) {
                 databaseList,
                 tableList,
                 splitSize,
+                splitColumn,
                 distributionFactorUpper,
                 distributionFactorLower,
                 sampleShardingThreshold,

seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/config/OracleSourceConfig.java

@@ -49,6 +49,7 @@ public OracleSourceConfig(
             List<String> databaseList,
             List<String> tableList,
             int splitSize,
+            Properties splitColumn,
             double distributionFactorUpper,
             double distributionFactorLower,
             int sampleShardingThreshold,
@@ -72,6 +73,7 @@ public OracleSourceConfig(
                 databaseList,
                 tableList,
                 splitSize,
+                splitColumn,
                 distributionFactorUpper,
                 distributionFactorLower,
                 sampleShardingThreshold,

seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/config/OracleSourceConfigFactory.java

@@ -144,6 +144,7 @@ public OracleSourceConfig create(int subtask) {
                 databaseList,
                 tableList,
                 splitSize,
+                splitColumn,
                 distributionFactorUpper,
                 distributionFactorLower,
                 sampleShardingThreshold,

seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/config/PostgresSourceConfig.java

@@ -36,6 +36,7 @@ public PostgresSourceConfig(
             List<String> databaseList,
             List<String> tableList,
             int splitSize,
+            Properties splitColumn,
             double distributionFactorUpper,
             double distributionFactorLower,
             int sampleShardingThreshold,
@@ -59,6 +60,7 @@ public PostgresSourceConfig(
                 databaseList,
                 tableList,
                 splitSize,
+                splitColumn,
                 distributionFactorUpper,
                 distributionFactorLower,
                 sampleShardingThreshold,

seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/config/PostgresSourceConfigFactory.java

@@ -115,6 +115,7 @@ public PostgresSourceConfig create(int subtask) {
                 databaseList,
                 tableList,
                 splitSize,
+                splitColumn,
                 distributionFactorUpper,
                 distributionFactorLower,
                 sampleShardingThreshold,

seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/sqlserver/config/SqlServerSourceConfig.java

@@ -41,6 +41,7 @@ public SqlServerSourceConfig(
             List<String> databaseList,
             List<String> tableList,
             int splitSize,
+            Properties splitColumn,
             double distributionFactorUpper,
             double distributionFactorLower,
             int sampleShardingThreshold,
@@ -64,6 +65,7 @@ public SqlServerSourceConfig(
                 databaseList,
                 tableList,
                 splitSize,
+                splitColumn,
                 distributionFactorUpper,
                 distributionFactorLower,
                 sampleShardingThreshold,