This is an automated email from the ASF dual-hosted git repository.
davidzollo pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new edfc1c3447 [Improve][Connector-V2][JDBC] Add string split strategy
(#10861)
edfc1c3447 is described below
commit edfc1c3447049537b667db0004c591018063575f
Author: Jast <[email protected]>
AuthorDate: Wed May 13 20:52:10 2026 +0800
[Improve][Connector-V2][JDBC] Add string split strategy (#10861)
---
docs/en/connectors/source/Jdbc.md | 1 +
docs/zh/connectors/source/Jdbc.md | 5 +-
.../seatunnel/jdbc/config/JdbcSourceConfig.java | 5 +
.../seatunnel/jdbc/config/JdbcSourceOptions.java | 8 +
.../jdbc/internal/dialect/JdbcDialect.java | 18 +++
.../jdbc/internal/dialect/mysql/MysqlDialect.java | 119 +++++++++++++++
.../jdbc/source/AsciiStringRangeSplitter.java | 141 +++++++++++++++++
.../seatunnel/jdbc/source/ChunkSplitter.java | 45 ++++++
.../jdbc/source/DynamicChunkSplitter.java | 166 +++++++++++++++++++++
.../seatunnel/jdbc/source/FixedChunkSplitter.java | 140 ++++++++++++++++-
.../seatunnel/jdbc/source/JdbcSourceFactory.java | 1 +
.../jdbc/source/StringRangeSplitDecision.java | 40 +++++
.../seatunnel/jdbc/source/StringSplitStrategy.java | 40 +++++
.../jdbc/config/JdbcSourceConfigTest.java | 57 +++++++
.../internal/dialect/mysql/MysqlDialectTest.java | 105 +++++++++++++
.../jdbc/source/AsciiStringRangeSplitterTest.java | 51 +++++++
.../jdbc/source/FixedChunkSplitterTest.java | 118 +++++++++++++++
.../seatunnel/jdbc/JdbcMysqlSplitIT.java | 45 ++++++
18 files changed, 1095 insertions(+), 10 deletions(-)
diff --git a/docs/en/connectors/source/Jdbc.md
b/docs/en/connectors/source/Jdbc.md
index 6f696f170c..88284fc874 100644
--- a/docs/en/connectors/source/Jdbc.md
+++ b/docs/en/connectors/source/Jdbc.md
@@ -74,6 +74,7 @@ supports query SQL and can achieve projection effect.
| split.allow-sampling | Boolean | No | true
| Whether to allow sampling-based sharding strategy. When set to false,
the system will fall back to unevenly-sized chunk splitting (iterative query
approach) regardless of the shard count.
|
| common-options | | No | -
| Source plugin common parameters, please refer to [Source Common
Options](../common-options/source-common-options.md) for details.
[...]
| split.string_split_mode | String | No | sample
| Supports different string splitting algorithms. By default, `sample` is
used to determine the split by sampling the string value. You can switch to
`charset_based` to enable charset-based string splitting algorithm. When set to
`charset_based`, the algorithm assumes characters of partition_column are
within ASCII range 32-126, which covers most character-based splitting
scenarios. [...]
+| split.string-strategy | String | No | -
| Controls how String partition columns are split. Available values are
`none`, `hash`, `range`, and `auto`. `range` and `auto` currently require MySQL
binary collation and fixed-length printable ASCII key values. Other JDBC
dialects reject `range` and `auto` until their range split support is
explicitly validated. `auto` tries range splitting first and falls back to hash
splitting when range splitting i [...]
| split.string_split_mode_collate | String | No | -
| Specifies the collation to use when string_split_mode is set to
`charset_based` and the table has a special collation. If not specified, the
database's default collation will be used.
[...]
### Table Matching
diff --git a/docs/zh/connectors/source/Jdbc.md
b/docs/zh/connectors/source/Jdbc.md
index b109dcbb2c..e8fdf2e48b 100644
--- a/docs/zh/connectors/source/Jdbc.md
+++ b/docs/zh/connectors/source/Jdbc.md
@@ -71,8 +71,9 @@ import ChangeLog from '../changelog/connector-jdbc.md';
| split.even-distribution.factor.upper-bound | Double | 否 | 100
| 不建议修改。<br/> 分片键分布因子的上界。该因子用于判断表数据是否均匀分布。如果计算出的分布因子小于或等于此上界（即 (MAX(id) -
MIN(id) + 1) / 行数），则将优化为均匀分布的分片方式。否则，如果分布因子更大，则表将被视为非均匀分布，当预估的分片数超过
`sample-sharding.threshold` 指定的值时，将使用基于采样的分片策略。默认值为 100.0。 |
| split.sample-sharding.threshold | Int | 否 | 1000
| 此配置指定触发采样分片策略的预估分片数阈值。当分布因子超出 `split.even-distribution.factor.upper-bound` 和
`split.even-distribution.factor.lower-bound` 指定的范围，且预估分片数（计算方式为近似行数 /
分片大小）超过此阈值时，将使用采样分片策略。这有助于更高效地处理大数据集。默认值为 1000。
|
| split.inverse-sampling.rate | Int | 否 | 1000
| 采样分片策略中使用的采样率的倒数。例如，如果此值设置为 1000，则表示在采样过程中应用 1/1000
的采样率。此选项提供了控制采样粒度的灵活性，从而影响最终的分片数量。在处理大数据集时尤为有用，此时较低的采样率可能更合适。默认值为 1000。
|
-| split.allow-sampling | Boolean | 否 | true |
是否启用基于采样的分片策略。当设置为 false 时，无论预估分片数是否超过阈值，系统都将回退到非均匀分片方式（迭代查询方式）。默认值为 true。
|
+| split.allow-sampling | Boolean | 否 | true
| 是否启用基于采样的分片策略。当设置为 false 时，无论预估分片数是否超过阈值，系统都将回退到非均匀分片方式（迭代查询方式）。默认值为 true。
|
| split.string_split_mode | String | 否 | sample
| 支持不同的字符串分割算法。默认使用 `sample`，通过采样字符串值来确定分割。可以切换为 `charset_based`
启用基于字符集的字符串分割算法。设置为 `charset_based` 时，算法假定 partition_column 的字符在 ASCII 32-126
范围内，覆盖了大多数基于字符的分割场景。
|
+| split.string-strategy | String | 否 | -
| 控制 String 分区列的分片方式，可选值为 `none`、`hash`、`range`、`auto`。`range` 和 `auto` 当前要求
MySQL binary collation，且键值为固定长度的可打印 ASCII 字符串。其他 JDBC 方言在显式验证 range 分片支持前会拒绝
`range` 和 `auto`。`auto` 会优先尝试 range 分片，range 不安全时回退为 hash 分片。未设置该参数时，SeaTunnel
保持现有 `split.string_split_mode` 行为。
[...]
| split.string_split_mode_collate | String | 否 | -
| 当 string_split_mode 设置为 `charset_based`
且表具有特殊排序规则时，指定要使用的排序规则。如果未指定，将使用数据库的默认排序规则。
|
| common-options | | 否 | - |
源插件通用参数，请参考 [源通用选项](../common-options/source-common-options.md) 详见。
[...]
@@ -338,5 +339,3 @@ source {
## 变更日志
<ChangeLog />
-
-
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/config/JdbcSourceConfig.java
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/config/JdbcSourceConfig.java
index 4f63baf207..093c66d4df 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/config/JdbcSourceConfig.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/config/JdbcSourceConfig.java
@@ -19,6 +19,7 @@ package org.apache.seatunnel.connectors.seatunnel.jdbc.config;
import org.apache.seatunnel.api.configuration.ReadonlyConfig;
import org.apache.seatunnel.connectors.seatunnel.jdbc.source.StringSplitMode;
+import
org.apache.seatunnel.connectors.seatunnel.jdbc.source.StringSplitStrategy;
import lombok.Builder;
import lombok.Data;
@@ -49,6 +50,8 @@ public class JdbcSourceConfig implements Serializable {
private StringSplitMode stringSplitMode;
+ private StringSplitStrategy stringSplitStrategy;
+
private String stringSplitModeCollate;
public static JdbcSourceConfig of(ReadonlyConfig config) {
@@ -63,6 +66,8 @@ public class JdbcSourceConfig implements Serializable {
&&
config.getOptional(JdbcSourceOptions.PARTITION_COLUMN).isPresent();
builder.useDynamicSplitter(!isOldVersion);
builder.stringSplitMode(config.get(JdbcSourceOptions.STRING_SPLIT_MODE));
+ config.getOptional(JdbcSourceOptions.STRING_SPLIT_STRATEGY)
+ .ifPresent(builder::stringSplitStrategy);
builder.stringSplitModeCollate(config.get(JdbcSourceOptions.STRING_SPLIT_MODE_COLLATE));
builder.splitSize(config.get(JdbcSourceOptions.SPLIT_SIZE));
builder.splitEvenDistributionFactorUpperBound(
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/config/JdbcSourceOptions.java
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/config/JdbcSourceOptions.java
index d2252e2997..b34a36b430 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/config/JdbcSourceOptions.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/config/JdbcSourceOptions.java
@@ -20,6 +20,7 @@ package org.apache.seatunnel.connectors.seatunnel.jdbc.config;
import org.apache.seatunnel.api.configuration.Option;
import org.apache.seatunnel.api.configuration.Options;
import org.apache.seatunnel.connectors.seatunnel.jdbc.source.StringSplitMode;
+import
org.apache.seatunnel.connectors.seatunnel.jdbc.source.StringSplitStrategy;
import java.util.List;
@@ -162,6 +163,13 @@ public class JdbcSourceOptions extends JdbcCommonOptions {
.withDescription(
"Supports different string splitting algorithms.
By default, `sample` is used to determine the split by sampling the string
value. You can switch to `charset_based` to enable charset-based string
splitting algorithm. When set to `charset_based`, the algorithm assumes
characters of partition_column are within ASCII range 32-126, which covers most
character-based splitting scenarios.");
+ public static final Option<StringSplitStrategy> STRING_SPLIT_STRATEGY =
+ Options.key("split.string-strategy")
+ .enumType(StringSplitStrategy.class)
+ .noDefaultValue()
+ .withDescription(
+ "Controls how String partition columns are split.
`range` uses range-based splitting, `hash` uses hash-based splitting, `none`
disables String splitting, and `auto` prefers range splitting then falls back
to hash when needed. `range` and `auto` currently require MySQL binary
collation and fixed-length printable ASCII key values. When this option is not
set, SeaTunnel keeps the existing split.string_split_mode behavior.");
+
public static final Option<String> STRING_SPLIT_MODE_COLLATE =
Options.key("split.string_split_mode_collate")
.stringType()
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/JdbcDialect.java
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/JdbcDialect.java
index 3f133606dc..17f9d5d49e 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/JdbcDialect.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/JdbcDialect.java
@@ -37,6 +37,7 @@ import
org.apache.seatunnel.connectors.seatunnel.jdbc.internal.connection.Simple
import
org.apache.seatunnel.connectors.seatunnel.jdbc.internal.converter.JdbcRowConverter;
import
org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.dialectenum.FieldIdeEnum;
import org.apache.seatunnel.connectors.seatunnel.jdbc.source.JdbcSourceTable;
+import
org.apache.seatunnel.connectors.seatunnel.jdbc.source.StringRangeSplitDecision;
import org.apache.seatunnel.connectors.seatunnel.jdbc.utils.DefaultValueUtils;
import org.slf4j.Logger;
@@ -335,6 +336,23 @@ public interface JdbcDialect extends Serializable {
return SQLUtils.countForTable(connection,
tableIdentifier(table.getTablePath()));
}
+ /** Dialects must opt in after verifying collation and key-shape
constraints. */
+ default StringRangeSplitDecision validateStringRangeSplit(
+ Connection connection, JdbcSourceTable table, String columnName,
int sampleSize)
+ throws SQLException {
+ return StringRangeSplitDecision.unsafe(
+ "string range split is not validated for this JDBC dialect");
+ }
+
+ /** Returns whether this dialect has validated string range split support.
*/
+ default boolean supportStringRangeSplit() {
+ return false;
+ }
+
+ default boolean supportHashSplitter() {
+ return true;
+ }
+
/**
* Performs a sampling operation on the specified column of a table in a
JDBC-connected
* database.
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/mysql/MysqlDialect.java
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/mysql/MysqlDialect.java
index 2a6450b2cb..c63cbc4097 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/mysql/MysqlDialect.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/mysql/MysqlDialect.java
@@ -29,6 +29,7 @@ import
org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.JdbcDiale
import
org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.SQLUtils;
import
org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.dialectenum.FieldIdeEnum;
import org.apache.seatunnel.connectors.seatunnel.jdbc.source.JdbcSourceTable;
+import
org.apache.seatunnel.connectors.seatunnel.jdbc.source.StringRangeSplitDecision;
import com.mysql.cj.MysqlType;
import lombok.extern.slf4j.Slf4j;
@@ -42,6 +43,7 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
+import java.util.Locale;
import java.util.Map;
import java.util.Optional;
import java.util.stream.Collectors;
@@ -238,6 +240,123 @@ public class MysqlDialect implements JdbcDialect {
return SQLUtils.countForSubquery(connection, table.getQuery());
}
+ @Override
+ public StringRangeSplitDecision validateStringRangeSplit(
+ Connection connection, JdbcSourceTable table, String columnName,
int sampleSize)
+ throws SQLException {
+ if (table.getTablePath() == null
+ ||
TablePath.DEFAULT.getFullName().equals(table.getTablePath().getFullName())) {
+ return StringRangeSplitDecision.unsafe(
+ "missing physical table path for MySQL string range split
validation");
+ }
+
+ String collation = queryColumnCollation(connection, table, columnName);
+ if (StringUtils.isBlank(collation)) {
+ return StringRangeSplitDecision.unsafe(
+ String.format(
+ "column collation is unavailable for %s.%s",
+ table.getTablePath(), columnName));
+ }
+ if (!collation.toLowerCase(Locale.ROOT).endsWith("_bin")) {
+ return StringRangeSplitDecision.unsafe(
+ String.format("collation %s is not binary", collation));
+ }
+
+ List<String> samples = sampleStringValues(connection, table,
columnName, sampleSize);
+ if (samples.isEmpty()) {
+ return StringRangeSplitDecision.unsafe("no non-null sample values
found");
+ }
+ Integer sampleLength = null;
+ for (String sample : samples) {
+ if (!isPrintableAscii(sample)) {
+ return StringRangeSplitDecision.unsafe(
+ String.format("sample value contains non-ASCII
characters: [%s]", sample));
+ }
+ if (sampleLength == null) {
+ sampleLength = sample.length();
+ } else if (sample.length() != sampleLength) {
+ return StringRangeSplitDecision.unsafe(
+ "sample values have variable lengths and cannot
preserve string range order");
+ }
+ }
+ return StringRangeSplitDecision.safe(
+ String.format(
+ "collation %s is binary and %s sampled values are
fixed-length printable ASCII",
+ collation, samples.size()));
+ }
+
+ @Override
+ public boolean supportStringRangeSplit() {
+ return true;
+ }
+
+ private String queryColumnCollation(
+ Connection connection, JdbcSourceTable table, String columnName)
throws SQLException {
+ TablePath tablePath = table.getTablePath();
+ String sql =
+ "SELECT COLLATION_NAME "
+ + "FROM information_schema.COLUMNS "
+ + "WHERE TABLE_SCHEMA = ? AND TABLE_NAME = ? AND
COLUMN_NAME = ?";
+ try (PreparedStatement ps = connection.prepareStatement(sql)) {
+ ps.setString(1, tablePath.getDatabaseName());
+ ps.setString(2, tablePath.getTableName());
+ ps.setString(3, columnName);
+ try (ResultSet rs = ps.executeQuery()) {
+ if (rs.next()) {
+ return rs.getString(1);
+ }
+ return null;
+ }
+ }
+ }
+
+ private List<String> sampleStringValues(
+ Connection connection, JdbcSourceTable table, String columnName,
int sampleSize)
+ throws SQLException {
+ String quotedColumn = quoteIdentifier(columnName);
+ String sql;
+ if (StringUtils.isNotBlank(table.getQuery())) {
+ sql =
+ String.format(
+ "SELECT %s FROM (%s) tmp WHERE %s IS NOT NULL
ORDER BY %s ASC LIMIT %s",
+ quotedColumn, table.getQuery(), quotedColumn,
quotedColumn, sampleSize);
+ } else {
+ sql =
+ String.format(
+ "SELECT %s FROM %s WHERE %s IS NOT NULL ORDER BY
%s ASC LIMIT %s",
+ quotedColumn,
+ tableIdentifier(table.getTablePath()),
+ quotedColumn,
+ quotedColumn,
+ sampleSize);
+ }
+ List<String> samples = new ArrayList<>(sampleSize);
+ try (Statement stmt =
+ connection.createStatement(
+ ResultSet.TYPE_FORWARD_ONLY,
ResultSet.CONCUR_READ_ONLY)) {
+ stmt.setFetchSize(Integer.MIN_VALUE);
+ try (ResultSet rs = stmt.executeQuery(sql)) {
+ while (rs.next()) {
+ String value = rs.getString(1);
+ if (value != null) {
+ samples.add(value);
+ }
+ }
+ }
+ }
+ return samples;
+ }
+
+ private boolean isPrintableAscii(String value) {
+ for (int i = 0; i < value.length(); i++) {
+ char ch = value.charAt(i);
+ if (ch < 32 || ch > 126) {
+ return false;
+ }
+ }
+ return true;
+ }
+
@Override
public boolean supportDefaultValue(BasicTypeDefine typeBasicTypeDefine) {
MysqlType nativeType = (MysqlType) typeBasicTypeDefine.getNativeType();
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/AsciiStringRangeSplitter.java
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/AsciiStringRangeSplitter.java
new file mode 100644
index 0000000000..958844447a
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/AsciiStringRangeSplitter.java
@@ -0,0 +1,141 @@
+/*
+ * 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.seatunnel.jdbc.source;
+
+import java.math.BigInteger;
+
+final class AsciiStringRangeSplitter {
+
+ private static final int FIRST_PRINTABLE_ASCII = 32;
+ private static final int LAST_PRINTABLE_ASCII = 126;
+ private static final int RADIX = LAST_PRINTABLE_ASCII -
FIRST_PRINTABLE_ASCII + 1;
+
+ private AsciiStringRangeSplitter() {}
+
+ // This mapping is order-preserving only for fixed-length printable ASCII
strings under a
+ // binary collation. Other shapes must be rejected and handled by hash or
single splitting.
+ static String[] split(String left, String right, int expectSliceNumber) {
+ if (left == null || right == null) {
+ throw new IllegalArgumentException("String range boundary cannot
be null");
+ }
+ validateRangeBoundary(left, "left");
+ validateRangeBoundary(right, "right");
+ if (left.length() != right.length()) {
+ throw new IllegalArgumentException(
+ "String range split requires fixed-length ASCII
boundaries");
+ }
+ if (left.equals(right)) {
+ return new String[] {left, right};
+ }
+ if (left.compareTo(right) > 0) {
+ throw new IllegalArgumentException(
+ String.format(
+ "String range left boundary [%s] must not be
greater than right boundary [%s]",
+ left, right));
+ }
+
+ BigInteger[] splitValues =
+ splitBigInteger(
+ stringToBigInteger(left), stringToBigInteger(right),
expectSliceNumber);
+ String[] result = new String[splitValues.length];
+ result[0] = left;
+ result[splitValues.length - 1] = right;
+ for (int i = 1; i < splitValues.length - 1; i++) {
+ result[i] = bigIntegerToString(splitValues[i], left.length());
+ validateRangeBoundary(result[i], "generated");
+ }
+ validateLexicalOrder(result);
+ return result;
+ }
+
+ private static BigInteger[] splitBigInteger(
+ BigInteger left, BigInteger right, int expectSliceNumber) {
+ if (expectSliceNumber < 1) {
+ throw new IllegalArgumentException("expectSliceNumber must be
greater than 0");
+ }
+ BigInteger gap = right.subtract(left);
+ BigInteger step = gap.divide(BigInteger.valueOf(expectSliceNumber));
+ BigInteger remainder =
gap.remainder(BigInteger.valueOf(expectSliceNumber));
+ if (step.compareTo(BigInteger.ZERO) == 0) {
+ expectSliceNumber = Math.max(remainder.intValue(), 1);
+ }
+
+ BigInteger[] result = new BigInteger[expectSliceNumber + 1];
+ result[0] = left;
+ result[expectSliceNumber] = right;
+
+ BigInteger upperBound = left;
+ for (int i = 1; i < expectSliceNumber; i++) {
+ upperBound = upperBound.add(step);
+ if (remainder.compareTo(BigInteger.valueOf(i)) >= 0) {
+ upperBound = upperBound.add(BigInteger.ONE);
+ }
+ result[i] = upperBound;
+ }
+ return result;
+ }
+
+ private static BigInteger stringToBigInteger(String value) {
+ BigInteger result = BigInteger.ZERO;
+ BigInteger radix = BigInteger.valueOf(RADIX);
+ for (int i = value.length() - 1, index = 0; i >= 0; i--, index++) {
+ result =
+ result.add(
+ BigInteger.valueOf(value.charAt(i) -
FIRST_PRINTABLE_ASCII)
+ .multiply(radix.pow(index)));
+ }
+ return result;
+ }
+
+ private static String bigIntegerToString(BigInteger value, int length) {
+ char[] result = new char[length];
+ BigInteger current = value;
+ BigInteger radix = BigInteger.valueOf(RADIX);
+ for (int i = length - 1; i >= 0; i--) {
+ result[i] = (char) (current.mod(radix).intValue() +
FIRST_PRINTABLE_ASCII);
+ current = current.divide(radix);
+ }
+ if (current.compareTo(BigInteger.ZERO) > 0) {
+ throw new IllegalArgumentException("Generated string boundary
exceeds fixed length");
+ }
+ return new String(result);
+ }
+
+ private static void validateRangeBoundary(String value, String name) {
+ for (int i = 0; i < value.length(); i++) {
+ char ch = value.charAt(i);
+ if (ch < FIRST_PRINTABLE_ASCII || ch > LAST_PRINTABLE_ASCII) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Only printable ASCII strings are supported
for string range split, %s=[%s]",
+ name, value));
+ }
+ }
+ }
+
+ private static void validateLexicalOrder(String[] values) {
+ for (int i = 1; i < values.length; i++) {
+ if (values[i - 1].compareTo(values[i]) >= 0) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Generated string range boundary [%s] is not
greater than previous boundary [%s]",
+ values[i], values[i - 1]));
+ }
+ }
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/ChunkSplitter.java
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/ChunkSplitter.java
index 03d86a38c1..67750cd70b 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/ChunkSplitter.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/ChunkSplitter.java
@@ -178,6 +178,51 @@ public abstract class ChunkSplitter implements
AutoCloseable, Serializable {
null);
}
+ protected StringSplitStrategy resolveStringSplitStrategy(
+ JdbcSourceTable table, String splitKeyName) throws SQLException {
+ StringSplitStrategy requestedStrategy =
config.getStringSplitStrategy();
+ if (requestedStrategy == null
+ || (requestedStrategy != StringSplitStrategy.RANGE
+ && requestedStrategy != StringSplitStrategy.AUTO)) {
+ return requestedStrategy;
+ }
+
+ if (!jdbcDialect.supportStringRangeSplit()) {
+ throw new JdbcConnectorException(
+ CommonErrorCodeDeprecated.ILLEGAL_ARGUMENT,
+ String.format(
+ "String split strategy %s requires validated range
split support, but dialect %s does not support range/auto string split
strategy.",
+ requestedStrategy, jdbcDialect.dialectName()));
+ }
+
+ StringRangeSplitDecision decision =
+ jdbcDialect.validateStringRangeSplit(
+ getOrEstablishConnection(), table, splitKeyName, 256);
+ if (decision.isSafe()) {
+ return StringSplitStrategy.RANGE;
+ }
+ if (requestedStrategy == StringSplitStrategy.RANGE) {
+ throw new JdbcConnectorException(
+ CommonErrorCodeDeprecated.ILLEGAL_ARGUMENT,
+ String.format(
+ "String range split is unsafe for table %s, split
column %s: %s",
+ table.getTablePath(), splitKeyName,
decision.getReason()));
+ }
+
+ StringSplitStrategy fallback =
+ jdbcDialect.supportHashSplitter()
+ ? StringSplitStrategy.HASH
+ : StringSplitStrategy.NONE;
+ log.warn(
+ "String range split is unsafe for table {}, split column {}.
Requested strategy: {}, fallback strategy: {}, reason: {}",
+ table.getTablePath(),
+ splitKeyName,
+ requestedStrategy,
+ fallback,
+ decision.getReason());
+ return fallback;
+ }
+
protected PreparedStatement createSingleSplitStatement(JdbcSourceSplit
split)
throws SQLException {
String splitQuery = split.getSplitQuery();
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/DynamicChunkSplitter.java
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/DynamicChunkSplitter.java
index cc03b5d471..a6c10b8a5e 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/DynamicChunkSplitter.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/DynamicChunkSplitter.java
@@ -26,8 +26,11 @@ import org.apache.seatunnel.api.table.catalog.TablePath;
import org.apache.seatunnel.api.table.catalog.TableSchema;
import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+import org.apache.seatunnel.api.table.type.SqlType;
import org.apache.seatunnel.common.exception.CommonError;
+import org.apache.seatunnel.common.exception.CommonErrorCodeDeprecated;
import org.apache.seatunnel.connectors.seatunnel.jdbc.config.JdbcSourceConfig;
+import
org.apache.seatunnel.connectors.seatunnel.jdbc.exception.JdbcConnectorException;
import org.apache.seatunnel.connectors.seatunnel.jdbc.utils.ObjectUtils;
import lombok.Data;
@@ -79,6 +82,11 @@ public class DynamicChunkSplitter extends ChunkSplitter {
JdbcSourceTable table, SeaTunnelRowType splitKey) throws Exception
{
String splitKeyName = splitKey.getFieldNames()[0];
SeaTunnelDataType splitKeyType = splitKey.getFieldType(0);
+ if (SqlType.STRING.equals(splitKeyType.getSqlType())
+ && config.getStringSplitStrategy() != null) {
+ return createStringStrategySplits(table, splitKeyName,
splitKeyType);
+ }
+
List<ChunkRange> chunks = splitTableIntoChunks(table, splitKeyName,
splitKeyType);
List<JdbcSourceSplit> splits = new ArrayList<>();
@@ -100,12 +108,170 @@ public class DynamicChunkSplitter extends ChunkSplitter {
private PreparedStatement createDynamicSplitStatement(JdbcSourceSplit
split, TableSchema schema)
throws SQLException {
+ if (isHashStringSplit(split)) {
+ return createStringColumnSplitStatement(split);
+ }
String splitQuery = createDynamicSplitQuerySQL(split, schema);
PreparedStatement statement = createPreparedStatement(splitQuery);
prepareDynamicSplitStatement(statement, split);
return statement;
}
+ private boolean isHashStringSplit(JdbcSourceSplit split) {
+ return SqlType.STRING.equals(split.getSplitKeyType().getSqlType())
+ && split.getSplitStart() instanceof Integer
+ && split.getSplitEnd() == null;
+ }
+
+ private PreparedStatement createStringColumnSplitStatement(JdbcSourceSplit
split)
+ throws SQLException {
+ PreparedStatement statement =
createPreparedStatement(split.getSplitQuery());
+ statement.setInt(1, (Integer) split.getSplitStart());
+ return statement;
+ }
+
+ private Collection<JdbcSourceSplit> createStringStrategySplits(
+ JdbcSourceTable table, String splitKeyName, SeaTunnelDataType
splitKeyType)
+ throws Exception {
+ StringSplitStrategy strategy = resolveStringSplitStrategy(table,
splitKeyName);
+ switch (strategy) {
+ case NONE:
+ return Collections.singletonList(
+ createSingleStringSplit(table, splitKeyName,
splitKeyType));
+ case HASH:
+ if (jdbcDialect.supportHashSplitter()) {
+ return createStringColumnSplits(
+ table, splitKeyName, splitKeyType,
config.getSplitSize());
+ }
+ return Collections.singletonList(
+ createSingleStringSplit(table, splitKeyName,
splitKeyType));
+ case RANGE:
+ if (config.getStringSplitStrategy() ==
StringSplitStrategy.AUTO) {
+ try {
+ return createStringRangeSplits(table, splitKeyName,
splitKeyType);
+ } catch (Exception e) {
+ log.warn(
+ "Range string split failed for table {},
fallback to hash split",
+ table.getTablePath(),
+ e);
+ if (jdbcDialect.supportHashSplitter()) {
+ return createStringColumnSplits(
+ table, splitKeyName, splitKeyType,
config.getSplitSize());
+ }
+ return Collections.singletonList(
+ createSingleStringSplit(table, splitKeyName,
splitKeyType));
+ }
+ }
+ return createStringRangeSplits(table, splitKeyName,
splitKeyType);
+ default:
+ throw new JdbcConnectorException(
+ CommonErrorCodeDeprecated.ILLEGAL_ARGUMENT,
+ "Unsupported string split strategy: " +
config.getStringSplitStrategy());
+ }
+ }
+
+ private JdbcSourceSplit createSingleStringSplit(
+ JdbcSourceTable table, String splitKeyName, SeaTunnelDataType
splitKeyType) {
+ return new JdbcSourceSplit(
+ table.getTablePath(),
+ createSplitId(table.getTablePath(), 0),
+ table.getQuery(),
+ splitKeyName,
+ splitKeyType,
+ null,
+ null);
+ }
+
+ @VisibleForTesting
+ Collection<JdbcSourceSplit> createStringColumnSplits(
+ JdbcSourceTable table,
+ String splitKeyName,
+ SeaTunnelDataType splitKeyType,
+ int chunkSize)
+ throws SQLException {
+ log.info("Use string hash chunks for table {}", table.getTablePath());
+ long approximateRowCnt = queryApproximateRowCnt(table);
+ int shardCount = Math.max((int) (approximateRowCnt /
Math.max(chunkSize, 1)) + 1, 1);
+ List<JdbcSourceSplit> splits = new ArrayList<>(shardCount);
+ Column column =
+ table.getCatalogTable().getTableSchema().getColumns().stream()
+ .filter(c -> c.getName().equals(splitKeyName))
+ .findAny()
+ .get();
+ for (int i = 0; i < shardCount; i++) {
+ String splitQuery;
+ if (StringUtils.isNotBlank(table.getQuery())) {
+ splitQuery =
+ String.format(
+ "SELECT * FROM (%s) st_jdbc_splitter WHERE %s
= ?",
+ applyUserWhereCondition(table.getQuery()),
+ jdbcDialect.hashModForField(
+ column.getSourceType(), splitKeyName,
shardCount));
+ } else if
(StringUtils.isNotBlank(config.getWhereConditionClause())) {
+ String userQuery =
+ String.format(
+ "SELECT * FROM %s",
+
jdbcDialect.tableIdentifier(table.getTablePath()));
+ splitQuery =
+ String.format(
+ "SELECT * FROM (%s) st_jdbc_splitter WHERE %s
= ?",
+ applyUserWhereCondition(userQuery),
+ jdbcDialect.hashModForField(
+ column.getSourceType(), splitKeyName,
shardCount));
+ } else {
+ splitQuery =
+ String.format(
+ "SELECT * FROM %s WHERE %s = ?",
+
jdbcDialect.tableIdentifier(table.getTablePath()),
+ jdbcDialect.hashModForField(
+ column.getSourceType(), splitKeyName,
shardCount));
+ }
+
+ splits.add(
+ new JdbcSourceSplit(
+ table.getTablePath(),
+ createSplitId(table.getTablePath(), i),
+ splitQuery,
+ splitKeyName,
+ splitKeyType,
+ i,
+ null));
+ }
+ return splits;
+ }
+
+ private Collection<JdbcSourceSplit> createStringRangeSplits(
+ JdbcSourceTable table, String splitKeyName, SeaTunnelDataType
splitKeyType)
+ throws SQLException {
+ Pair<Object, Object> splitColumnRange = queryMinMax(table,
splitKeyName);
+ String min =
+ splitColumnRange.getLeft() == null ? null :
splitColumnRange.getLeft().toString();
+ String max =
+ splitColumnRange.getRight() == null ? null :
splitColumnRange.getRight().toString();
+ if (min == null || max == null || min.equals(max)) {
+ return Collections.singletonList(
+ createSingleStringSplit(table, splitKeyName,
splitKeyType));
+ }
+
+ long approximateRowCnt = queryApproximateRowCnt(table);
+ int shardCount =
+ Math.max((int) (approximateRowCnt /
Math.max(config.getSplitSize(), 1)) + 1, 1);
+ String[] rangeResult = AsciiStringRangeSplitter.split(min, max,
shardCount);
+ List<JdbcSourceSplit> splits = new ArrayList<>(rangeResult.length - 1);
+ for (int i = 0; i < rangeResult.length - 1; i++) {
+ splits.add(
+ new JdbcSourceSplit(
+ table.getTablePath(),
+ createSplitId(table.getTablePath(), i),
+ table.getQuery(),
+ splitKeyName,
+ splitKeyType,
+ i == 0 ? null : rangeResult[i],
+ i == rangeResult.length - 2 ? null : rangeResult[i
+ 1]));
+ }
+ return splits;
+ }
+
private List<ChunkRange> splitTableIntoChunks(
JdbcSourceTable table, String splitColumnName, SeaTunnelDataType
splitColumnType)
throws Exception {
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/FixedChunkSplitter.java
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/FixedChunkSplitter.java
index 92a26490ee..309b6c82bd 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/FixedChunkSplitter.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/FixedChunkSplitter.java
@@ -76,11 +76,41 @@ public class FixedChunkSplitter extends ChunkSplitter {
}
}
if (SqlType.STRING.equals(splitKeyType.getSqlType())) {
- log.info("useNewStringSplitter is {}",
useCharsetBasedStringSplitter);
- if (useCharsetBasedStringSplitter) {
- return getJdbcSourceStringSplits(table, splitKeyName,
splitKeyType);
- } else {
- return createStringColumnSplits(table, splitKeyName,
splitKeyType);
+ if (config.getStringSplitStrategy() == null) {
+ log.info("useNewStringSplitter is {}",
useCharsetBasedStringSplitter);
+ if (useCharsetBasedStringSplitter) {
+ return getJdbcSourceStringSplits(table, splitKeyName,
splitKeyType);
+ } else {
+ return createStringColumnSplits(table, splitKeyName,
splitKeyType);
+ }
+ }
+ switch (resolveStringSplitStrategy(table, splitKeyName)) {
+ case NONE:
+ return Collections.singletonList(createSingleSplit(table));
+ case HASH:
+ return createStringColumnSplits(table, splitKeyName,
splitKeyType);
+ case RANGE:
+ if (config.getStringSplitStrategy() ==
StringSplitStrategy.AUTO) {
+ try {
+ return createStringRangeSplits(table,
splitKeyName, splitKeyType);
+ } catch (Exception e) {
+ log.warn(
+ "Range string split failed for table {},
fallback to hash split",
+ table.getTablePath(),
+ e);
+ return createStringColumnSplits(table,
splitKeyName, splitKeyType);
+ }
+ }
+ return createStringRangeSplits(table, splitKeyName,
splitKeyType);
+ case AUTO:
+ throw new JdbcConnectorException(
+ CommonErrorCodeDeprecated.ILLEGAL_ARGUMENT,
+ "Unexpected string split strategy: " +
config.getStringSplitStrategy());
+ default:
+ throw new JdbcConnectorException(
+ CommonErrorCodeDeprecated.ILLEGAL_ARGUMENT,
+ "Unsupported string split strategy: "
+ + config.getStringSplitStrategy());
}
}
return getJdbcSourceSplits(table, splitKeyName, splitKeyType);
@@ -194,17 +224,32 @@ public class FixedChunkSplitter extends ChunkSplitter {
@Override
protected PreparedStatement createSplitStatement(JdbcSourceSplit split,
TableSchema schema)
throws SQLException {
- if (SqlType.STRING.equals(split.getSplitKeyType().getSqlType())
- && !useCharsetBasedStringSplitter) {
+ if (isHashStringSplit(split)) {
return createStringColumnSplitStatement(split);
}
if (split.getSplitStart() == null && split.getSplitEnd() == null) {
return createSingleSplitStatement(split);
}
+ if (isStringRangeSplit(split)) {
+ return createStringRangeSplitStatement(split);
+ }
return createNumberColumnSplitStatement(split);
}
+ private boolean isHashStringSplit(JdbcSourceSplit split) {
+ return SqlType.STRING.equals(split.getSplitKeyType().getSqlType())
+ && split.getSplitStart() instanceof Integer
+ && split.getSplitEnd() == null;
+ }
+
+ private boolean isStringRangeSplit(JdbcSourceSplit split) {
+ return config.getStringSplitStrategy() != null
+ && SqlType.STRING.equals(split.getSplitKeyType().getSqlType())
+ && (split.getSplitStart() instanceof String
+ || split.getSplitEnd() instanceof String);
+ }
+
private Collection<JdbcSourceSplit> createStringColumnSplits(
JdbcSourceTable table, String splitKeyName, SeaTunnelDataType
splitKeyType) {
List<JdbcSourceSplit> splits = new
ArrayList<>(table.getPartitionNumber());
@@ -264,6 +309,34 @@ public class FixedChunkSplitter extends ChunkSplitter {
return splits;
}
+ private Collection<JdbcSourceSplit> createStringRangeSplits(
+ JdbcSourceTable table, String splitKeyName, SeaTunnelDataType
splitKeyType)
+ throws SQLException {
+ Pair<Object, Object> splitColumnRange = queryMinMax(table,
splitKeyName);
+ String min =
+ splitColumnRange.getLeft() == null ? null :
splitColumnRange.getLeft().toString();
+ String max =
+ splitColumnRange.getRight() == null ? null :
splitColumnRange.getRight().toString();
+ if (min == null || max == null || min.equals(max)) {
+ return Collections.singletonList(createSingleSplit(table));
+ }
+
+ String[] rangeResult = AsciiStringRangeSplitter.split(min, max,
table.getPartitionNumber());
+ List<JdbcSourceSplit> splits = new ArrayList<>(rangeResult.length - 1);
+ for (int i = 0; i < rangeResult.length - 1; i++) {
+ splits.add(
+ new JdbcSourceSplit(
+ table.getTablePath(),
+ createSplitId(table.getTablePath(), i),
+ table.getQuery(),
+ splitKeyName,
+ splitKeyType,
+ i == 0 ? null : rangeResult[i],
+ i == rangeResult.length - 2 ? null : rangeResult[i
+ 1]));
+ }
+ return splits;
+ }
+
private PreparedStatement createStringColumnSplitStatement(JdbcSourceSplit
split)
throws SQLException {
PreparedStatement statement =
createPreparedStatement(split.getSplitQuery());
@@ -271,6 +344,59 @@ public class FixedChunkSplitter extends ChunkSplitter {
return statement;
}
+ private PreparedStatement createStringRangeSplitStatement(JdbcSourceSplit
split)
+ throws SQLException {
+ String splitKeyName =
jdbcDialect.quoteIdentifier(split.getSplitKeyName());
+ boolean isFirstSplit = split.getSplitStart() == null;
+ boolean isLastSplit = split.getSplitEnd() == null;
+
+ String condition;
+ if (isFirstSplit) {
+ condition = String.format("%s <= ? AND NOT (%s = ?)",
splitKeyName, splitKeyName);
+ } else if (isLastSplit) {
+ condition = String.format("%s >= ?", splitKeyName);
+ } else {
+ condition =
+ String.format(
+ "%s >= ? AND NOT (%s = ?) AND %s <= ?",
+ splitKeyName, splitKeyName, splitKeyName);
+ }
+
+ String splitQuery;
+ if (StringUtils.isNotBlank(split.getSplitQuery())) {
+ splitQuery =
+ String.format(
+ "SELECT * FROM (%s) st_jdbc_splitter WHERE %s",
+ applyUserWhereCondition(split.getSplitQuery()),
condition);
+ } else if (StringUtils.isNotBlank(config.getWhereConditionClause())) {
+ String userQuery =
+ String.format(
+ "SELECT * FROM %s",
jdbcDialect.tableIdentifier(split.getTablePath()));
+ splitQuery =
+ String.format(
+ "SELECT * FROM (%s) st_jdbc_splitter WHERE %s",
+ applyUserWhereCondition(userQuery), condition);
+ } else {
+ splitQuery =
+ String.format(
+ "SELECT * FROM %s WHERE %s",
+ jdbcDialect.tableIdentifier(split.getTablePath()),
condition);
+ }
+
+ PreparedStatement statement = createPreparedStatement(splitQuery);
+ if (isFirstSplit) {
+ statement.setString(1, (String) split.getSplitEnd());
+ statement.setString(2, (String) split.getSplitEnd());
+ } else if (isLastSplit) {
+ statement.setString(1, (String) split.getSplitStart());
+ } else {
+ statement.setString(1, (String) split.getSplitStart());
+ statement.setString(2, (String) split.getSplitEnd());
+ statement.setString(3, (String) split.getSplitEnd());
+ }
+ return statement;
+ }
+
private Collection<JdbcSourceSplit> createNumberColumnSplits(
JdbcSourceTable table,
String splitKeyName,
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/JdbcSourceFactory.java
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/JdbcSourceFactory.java
index c22d4555dc..de2da4aa41 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/JdbcSourceFactory.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/JdbcSourceFactory.java
@@ -74,6 +74,7 @@ public class JdbcSourceFactory implements TableSourceFactory {
JdbcSourceOptions.PARTITION_NUM,
JdbcSourceOptions.COMPATIBLE_MODE,
JdbcSourceOptions.STRING_SPLIT_MODE,
+ JdbcSourceOptions.STRING_SPLIT_STRATEGY,
JdbcSourceOptions.STRING_SPLIT_MODE_COLLATE,
JdbcSourceOptions.PROPERTIES,
JdbcSourceOptions.QUERY,
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/StringRangeSplitDecision.java
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/StringRangeSplitDecision.java
new file mode 100644
index 0000000000..f3c60d43ea
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/StringRangeSplitDecision.java
@@ -0,0 +1,40 @@
+/*
+ * 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.seatunnel.jdbc.source;
+
+import lombok.Getter;
+
+@Getter
+public class StringRangeSplitDecision {
+
+ private final boolean safe;
+ private final String reason;
+
+ private StringRangeSplitDecision(boolean safe, String reason) {
+ this.safe = safe;
+ this.reason = reason;
+ }
+
+ public static StringRangeSplitDecision safe(String reason) {
+ return new StringRangeSplitDecision(true, reason);
+ }
+
+ public static StringRangeSplitDecision unsafe(String reason) {
+ return new StringRangeSplitDecision(false, reason);
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/StringSplitStrategy.java
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/StringSplitStrategy.java
new file mode 100644
index 0000000000..07722320fb
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/StringSplitStrategy.java
@@ -0,0 +1,40 @@
+/*
+ * 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.seatunnel.jdbc.source;
+
+public enum StringSplitStrategy {
+ AUTO("auto"),
+ RANGE("range"),
+ HASH("hash"),
+ NONE("none");
+
+ private final String strategy;
+
+ StringSplitStrategy(String strategy) {
+ this.strategy = strategy;
+ }
+
+ public String getStrategy() {
+ return strategy;
+ }
+
+ @Override
+ public String toString() {
+ return strategy;
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/config/JdbcSourceConfigTest.java
b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/config/JdbcSourceConfigTest.java
new file mode 100644
index 0000000000..8dc074264a
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/config/JdbcSourceConfigTest.java
@@ -0,0 +1,57 @@
+/*
+ * 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.seatunnel.jdbc.config;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import
org.apache.seatunnel.connectors.seatunnel.jdbc.source.StringSplitStrategy;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+public class JdbcSourceConfigTest {
+
+ @Test
+ public void testStringSplitStrategyIsUnsetByDefault() {
+ JdbcSourceConfig sourceConfig =
JdbcSourceConfig.of(ReadonlyConfig.fromMap(baseConfig()));
+
+ assertNull(sourceConfig.getStringSplitStrategy());
+ }
+
+ @Test
+ public void testResolveExplicitStringSplitStrategy() {
+ Map<String, Object> configMap = baseConfig();
+ configMap.put("split.string-strategy", "range");
+
+ JdbcSourceConfig sourceConfig =
JdbcSourceConfig.of(ReadonlyConfig.fromMap(configMap));
+
+ assertEquals(StringSplitStrategy.RANGE,
sourceConfig.getStringSplitStrategy());
+ }
+
+ private Map<String, Object> baseConfig() {
+ Map<String, Object> configMap = new HashMap<>();
+ configMap.put("url", "jdbc:postgresql://localhost:5432/test");
+ configMap.put("driver", "org.postgresql.Driver");
+ configMap.put("table_path", "public.test");
+ return configMap;
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/mysql/MysqlDialectTest.java
b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/mysql/MysqlDialectTest.java
index 51fe7d57b6..122cec9e28 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/mysql/MysqlDialectTest.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/mysql/MysqlDialectTest.java
@@ -17,14 +17,24 @@
package org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.mysql;
+import org.apache.seatunnel.api.table.catalog.TablePath;
+import org.apache.seatunnel.connectors.seatunnel.jdbc.source.JdbcSourceTable;
+import
org.apache.seatunnel.connectors.seatunnel.jdbc.source.StringRangeSplitDecision;
+
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import lombok.extern.slf4j.Slf4j;
import java.security.MessageDigest;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.Statement;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.HashMap;
+import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.zip.CRC32;
@@ -32,6 +42,22 @@ import java.util.zip.CRC32;
@Slf4j
public class MysqlDialectTest {
+ @Test
+ public void testValidateStringRangeSplitAcceptsPrintableAsciiPunctuation()
throws Exception {
+ MysqlDialect dialect = new MysqlDialect();
+ JdbcSourceTable table =
+ JdbcSourceTable.builder().tablePath(TablePath.of("test_db",
"test_table")).build();
+
+ StringRangeSplitDecision decision =
+ dialect.validateStringRangeSplit(
+ mockConnection("utf8mb4_bin", Arrays.asList("AA/01",
"AB.CD", "ID~99")),
+ table,
+ "id",
+ 3);
+
+ Assertions.assertTrue(decision.isSafe());
+ }
+
@Test
public void testHashDistributionMD5vsCRC32WithSnowflakeIds() {
int totalRecords = 1_100_000;
@@ -188,4 +214,83 @@ public class MysqlDialectTest {
return Math.sqrt(sumSquaredDiff / partitions);
}
+
+ private Connection mockConnection(String collation, List<String> samples) {
+ return (Connection)
+ java.lang.reflect.Proxy.newProxyInstance(
+ Connection.class.getClassLoader(),
+ new Class<?>[] {Connection.class},
+ (proxy, method, args) -> {
+ switch (method.getName()) {
+ case "prepareStatement":
+ return mockPreparedStatement(collation);
+ case "createStatement":
+ return mockStatement(samples);
+ default:
+ return
defaultValue(method.getReturnType());
+ }
+ });
+ }
+
+ private PreparedStatement mockPreparedStatement(String collation) {
+ return (PreparedStatement)
+ java.lang.reflect.Proxy.newProxyInstance(
+ PreparedStatement.class.getClassLoader(),
+ new Class<?>[] {PreparedStatement.class},
+ (proxy, method, args) -> {
+ if ("executeQuery".equals(method.getName())) {
+ return mockResultSet(Arrays.asList(collation));
+ }
+ return defaultValue(method.getReturnType());
+ });
+ }
+
+ private Statement mockStatement(List<String> samples) {
+ return (Statement)
+ java.lang.reflect.Proxy.newProxyInstance(
+ Statement.class.getClassLoader(),
+ new Class<?>[] {Statement.class},
+ (proxy, method, args) -> {
+ if ("executeQuery".equals(method.getName())) {
+ return mockResultSet(samples);
+ }
+ return defaultValue(method.getReturnType());
+ });
+ }
+
+ private ResultSet mockResultSet(List<String> values) {
+ Iterator<String> iterator = values.iterator();
+ String[] current = new String[1];
+ return (ResultSet)
+ java.lang.reflect.Proxy.newProxyInstance(
+ ResultSet.class.getClassLoader(),
+ new Class<?>[] {ResultSet.class},
+ (proxy, method, args) -> {
+ switch (method.getName()) {
+ case "next":
+ if (iterator.hasNext()) {
+ current[0] = iterator.next();
+ return true;
+ }
+ return false;
+ case "getString":
+ return current[0];
+ default:
+ return
defaultValue(method.getReturnType());
+ }
+ });
+ }
+
+ private Object defaultValue(Class<?> returnType) {
+ if (!returnType.isPrimitive()) {
+ return null;
+ }
+ if (boolean.class.equals(returnType)) {
+ return false;
+ }
+ if (void.class.equals(returnType)) {
+ return null;
+ }
+ return 0;
+ }
}
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/AsciiStringRangeSplitterTest.java
b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/AsciiStringRangeSplitterTest.java
new file mode 100644
index 0000000000..b8254ba108
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/AsciiStringRangeSplitterTest.java
@@ -0,0 +1,51 @@
+/*
+ * 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.seatunnel.jdbc.source;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+public class AsciiStringRangeSplitterTest {
+
+ @Test
+ public void testSplitFixedLengthPrintableAsciiRange() {
+ assertArrayEquals(
+ new String[] {"aa", "ff", "kk"},
AsciiStringRangeSplitter.split("aa", "kk", 2));
+ }
+
+ @Test
+ public void testSplitNumericSuffixKeepsGeneratedBoundariesPrintable() {
+ assertDoesNotThrow(() -> AsciiStringRangeSplitter.split("key00000",
"key00099", 11));
+ }
+
+ @Test
+ public void testRejectVariableLengthRange() {
+ assertThrows(
+ IllegalArgumentException.class, () ->
AsciiStringRangeSplitter.split("aa", "z", 2));
+ }
+
+ @Test
+ public void testRejectNonPrintableAsciiRange() {
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> AsciiStringRangeSplitter.split("aa", "中中", 2));
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/FixedChunkSplitterTest.java
b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/FixedChunkSplitterTest.java
index d36e8aaa1f..cdd4bb8e99 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/FixedChunkSplitterTest.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/FixedChunkSplitterTest.java
@@ -17,22 +17,103 @@
package org.apache.seatunnel.connectors.seatunnel.jdbc.source;
+import org.apache.seatunnel.api.table.catalog.TablePath;
+import org.apache.seatunnel.api.table.catalog.TableSchema;
+import org.apache.seatunnel.api.table.type.BasicType;
+import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
+import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
import
org.apache.seatunnel.connectors.seatunnel.jdbc.config.JdbcConnectionConfig;
import org.apache.seatunnel.connectors.seatunnel.jdbc.config.JdbcSourceConfig;
+import
org.apache.seatunnel.connectors.seatunnel.jdbc.exception.JdbcConnectorException;
import org.junit.jupiter.api.Test;
import lombok.extern.slf4j.Slf4j;
+import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
+import java.lang.reflect.Proxy;
import java.math.BigDecimal;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.util.HashMap;
+import java.util.Map;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
@Slf4j
public class FixedChunkSplitterTest {
+ @Test
+ public void testCreateFirstStringRangeSplitStatement() throws SQLException
{
+ CapturingFixedChunkSplitter splitter = new
CapturingFixedChunkSplitter(mysqlConfig());
+ JdbcSourceSplit split =
+ new JdbcSourceSplit(
+ TablePath.of("db", "tbl"),
+ "split-0",
+ null,
+ "id",
+ BasicType.STRING_TYPE,
+ null,
+ "mm");
+
+ splitter.generateSplitStatement(split, TableSchema.builder().build());
+
+ assertEquals("SELECT * FROM `db`.`tbl` WHERE `id` <= ? AND NOT (`id` =
?)", splitter.sql);
+ assertEquals("mm", splitter.stringParameters.get(1));
+ assertEquals("mm", splitter.stringParameters.get(2));
+ }
+
+ @Test
+ public void testCreateLastStringRangeSplitStatement() throws SQLException {
+ CapturingFixedChunkSplitter splitter = new
CapturingFixedChunkSplitter(mysqlConfig());
+ JdbcSourceSplit split =
+ new JdbcSourceSplit(
+ TablePath.of("db", "tbl"),
+ "split-1",
+ null,
+ "id",
+ BasicType.STRING_TYPE,
+ "mm",
+ null);
+
+ splitter.generateSplitStatement(split, TableSchema.builder().build());
+
+ assertEquals("SELECT * FROM `db`.`tbl` WHERE `id` >= ?", splitter.sql);
+ assertEquals("mm", splitter.stringParameters.get(1));
+ }
+
+ @Test
+ public void testRejectAutoStringRangeSplitForNonMysqlDialect() {
+ JdbcSourceConfig config =
+ JdbcSourceConfig.builder()
+ .jdbcConnectionConfig(
+ JdbcConnectionConfig.builder()
+
.url("jdbc:postgresql://localhost:5432/test")
+ .driverName("org.postgresql.Driver")
+ .build())
+ .stringSplitStrategy(StringSplitStrategy.AUTO)
+ .build();
+ CapturingFixedChunkSplitter splitter = new
CapturingFixedChunkSplitter(config);
+ JdbcSourceTable table =
+ JdbcSourceTable.builder().tablePath(TablePath.of("public",
"tbl")).build();
+
+ JdbcConnectorException exception =
+ assertThrows(
+ JdbcConnectorException.class,
+ () ->
+ splitter.createSplits(
+ table,
+ new SeaTunnelRowType(
+ new String[] {"id"},
+ new SeaTunnelDataType[]
{BasicType.STRING_TYPE})));
+
+ assertTrue(exception.getMessage().contains("does not support
range/auto"));
+ }
+
@Test
public void testConvertFloat() throws Exception {
JdbcSourceConfig config =
@@ -83,4 +164,41 @@ public class FixedChunkSplitterTest {
BigDecimal oldPrecisionWay = BigDecimal.valueOf(precisionTestFloat);
assertNotEquals(new BigDecimal("0.1"), oldPrecisionWay);
}
+
+ private static JdbcSourceConfig mysqlConfig() {
+ return JdbcSourceConfig.builder()
+ .jdbcConnectionConfig(
+ JdbcConnectionConfig.builder()
+ .url("jdbc:mysql://localhost:3306/test")
+ .driverName("com.mysql.cj.jdbc.Driver")
+ .build())
+ .stringSplitStrategy(StringSplitStrategy.RANGE)
+ .build();
+ }
+
+ private static class CapturingFixedChunkSplitter extends
FixedChunkSplitter {
+ private String sql;
+ private final Map<Integer, String> stringParameters = new HashMap<>();
+
+ private CapturingFixedChunkSplitter(JdbcSourceConfig config) {
+ super(config);
+ }
+
+ @Override
+ protected PreparedStatement createPreparedStatement(String sql) {
+ this.sql = sql;
+ InvocationHandler handler =
+ (proxy, method, args) -> {
+ if ("setString".equals(method.getName())) {
+ stringParameters.put((Integer) args[0], (String)
args[1]);
+ }
+ return null;
+ };
+ return (PreparedStatement)
+ Proxy.newProxyInstance(
+ PreparedStatement.class.getClassLoader(),
+ new Class<?>[] {PreparedStatement.class},
+ handler);
+ }
+ }
}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-7/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/JdbcMysqlSplitIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-7/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/JdbcMysqlSplitIT.java
index 040a4563d9..8a43578b3d 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-7/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/JdbcMysqlSplitIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-7/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/JdbcMysqlSplitIT.java
@@ -123,6 +123,7 @@ public class JdbcMysqlSplitIT extends TestSuiteBase
implements TestResource {
+ " `c_mediumtext` mediumtext,\n"
+ " `c_text` text,\n"
+ " `c_varchar` varchar(255)
DEFAULT NULL,\n"
+ + " `c_varchar_bin` varchar(8) CHARACTER SET
utf8mb4 COLLATE utf8mb4_bin DEFAULT NULL,\n"
+ " `c_json` json
DEFAULT NULL,\n"
+ " `c_longtext` longtext,\n"
+ " `c_date` date
DEFAULT NULL,\n"
@@ -267,6 +268,7 @@ public class JdbcMysqlSplitIT extends TestSuiteBase
implements TestResource {
"c_mediumtext",
"c_text",
"c_varchar",
+ "c_varchar_bin",
"c_json",
"c_longtext",
"c_date",
@@ -328,6 +330,7 @@ public class JdbcMysqlSplitIT extends TestSuiteBase
implements TestResource {
String.format("f1_%s", i),
String.format("f1_%s", i),
String.format("f1_%s", i),
+ String.format("key%05d", i),
String.format("{\"aa\":\"bb_%s\"}", i),
String.format("f1_%s", i),
Date.valueOf(currentDate),
@@ -467,6 +470,22 @@ public class JdbcMysqlSplitIT extends TestSuiteBase
implements TestResource {
configMap.put("partition_column", "c_date");
assertDateSplit(splitArray);
+ configMap.put("split.string-strategy", "none");
+ splitArray = getCheckedSplitArray(configMap, table, "c_varchar", 1);
+ assertStringSingleSplit(splitArray);
+
+ configMap.put("split.string-strategy", "hash");
+ splitArray = getCheckedSplitArray(configMap, table, "c_varchar", 11);
+ assertStringHashSplit(splitArray);
+
+ configMap.put("split.string-strategy", "range");
+ splitArray = getCheckedSplitArray(configMap, table, "c_varchar_bin",
11);
+ assertStringRangeSplit(splitArray);
+
+ configMap.put("split.string-strategy", "auto");
+ splitArray = getCheckedSplitArray(configMap, table, "c_varchar_bin",
11);
+ assertStringRangeSplit(splitArray);
+
mySqlCatalog.close();
}
@@ -535,6 +554,32 @@ public class JdbcMysqlSplitIT extends TestSuiteBase
implements TestResource {
}
}
+ private void assertStringSingleSplit(JdbcSourceSplit[] splitArray) {
+ Assertions.assertEquals(1, splitArray.length);
+ Assertions.assertNull(splitArray[0].getSplitStart());
+ Assertions.assertNull(splitArray[0].getSplitEnd());
+ }
+
+ private void assertStringHashSplit(JdbcSourceSplit[] splitArray) {
+ Assertions.assertTrue(splitArray.length > 1);
+ for (JdbcSourceSplit split : splitArray) {
+ Assertions.assertTrue(split.getSplitStart() instanceof Integer);
+ Assertions.assertNull(split.getSplitEnd());
+ }
+ }
+
+ private void assertStringRangeSplit(JdbcSourceSplit[] splitArray) {
+ Assertions.assertTrue(splitArray.length > 1);
+ Assertions.assertNull(splitArray[0].getSplitStart());
+ for (int i = 0; i < splitArray.length; i++) {
+ if (i == splitArray.length - 1) {
+ Assertions.assertNull(splitArray[i].getSplitEnd());
+ continue;
+ }
+ Assertions.assertTrue(splitArray[i].getSplitEnd() instanceof
String);
+ }
+ }
+
@NotNull private DynamicChunkSplitter getDynamicChunkSplitter(Map<String,
Object> configMap) {
ReadonlyConfig readonlyConfig = ReadonlyConfig.fromMap(configMap);
JdbcSourceConfig sourceConfig = JdbcSourceConfig.of(readonlyConfig);
