This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] 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 e649b2a319 [Fix][Connector-V2] Preserve numeric JDBC partition bounds
(#12584)
e649b2a319 is described below
commit e649b2a319c9234ecf8a47bae219f7e65e2c26c2
Author: hutiefang76 <[email protected]>
AuthorDate: Wed Oct 28 04:06:28 2026 +0000
[Fix][Connector-V2] Preserve numeric JDBC partition bounds (#12584)
Co-authored-by: zhiweiniu <[email protected]>
---
docs/en/connectors/source/Jdbc.md | 2 +
docs/zh/connectors/source/Jdbc.md | 2 +
.../JdbcNumericBetweenParametersProvider.java | 54 +++---
.../JdbcNumericBetweenParametersProviderTest.java | 184 +++++++++++++++++++++
.../jdbc/source/FixedChunkSplitterTest.java | 168 +++++++++++++++++++
.../seatunnel/jdbc/JdbcMysqlSplitIT.java | 82 +++++++++
6 files changed, 463 insertions(+), 29 deletions(-)
diff --git a/docs/en/connectors/source/Jdbc.md
b/docs/en/connectors/source/Jdbc.md
index 4ee50242d8..f4c20f3150 100644
--- a/docs/en/connectors/source/Jdbc.md
+++ b/docs/en/connectors/source/Jdbc.md
@@ -321,6 +321,8 @@ For a full-table snapshot, configure `table_path` and
normally leave split-key d
Only the top-level combination of `query` and `partition_column` selects the
legacy fixed splitter; `partition_num` then controls the number of splits. The
partition column must be present in the query result. Optional lower and upper
bounds avoid extra `MIN`/`MAX` queries, but incorrect bounds can omit source
rows, so use them only when the full data range is known.
+For fixed splitting of integer keys, including BIGINT and zero-scale DECIMAL,
the partitions cover the numeric range without gaps or overlaps. An uneven
division by `partition_num` preserves the remaining values, and each interval
stays within the lower and upper bounds.
+
Entries inside `table_list` continue to use dynamic splitting even when they
include `query` or partition settings. Do not expect `split.size` to affect the
top-level fixed partition mode.
### Query and primary-key caution
diff --git a/docs/zh/connectors/source/Jdbc.md
b/docs/zh/connectors/source/Jdbc.md
index 8f6fa0c37b..7d06080bb5 100644
--- a/docs/zh/connectors/source/Jdbc.md
+++ b/docs/zh/connectors/source/Jdbc.md
@@ -317,6 +317,8 @@ numeric_special_value_as_null = true
只有顶层同时配置 `query` 和 `partition_column` 时才会使用旧版 fixed splitter，再通过
`partition_num` 控制分片数。分区列必须包含在 query 结果中。配置上下界可以避免额外执行 `MIN`/`MAX`
查询，但错误的边界会漏读源数据，因此只有确认完整数据范围时才应设置。
+对 BIGINT、零标度 DECIMAL 等整数键使用固定分片时，各分片完整覆盖数值范围，不产生间隙或重叠。范围不能被 `partition_num`
整除时，剩余值仍会保留，各区间均位于上下界之内。
+
`table_list` 中的条目即使配置了 `query` 或分区参数，仍然使用动态分片。`split.size` 不影响顶层 fixed
partition 模式。
### Query 与主键注意事项
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/split/JdbcNumericBetweenParametersProvider.java
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/split/JdbcNumericBetweenParametersProvider.java
index 4bf8834f6d..c514450032 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/split/JdbcNumericBetweenParametersProvider.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/split/JdbcNumericBetweenParametersProvider.java
@@ -44,8 +44,9 @@ public class JdbcNumericBetweenParametersProvider implements
JdbcParameterValues
private final BigDecimal minVal;
private final BigDecimal maxVal;
- private long batchSize;
+ private BigDecimal batchSize;
private int batchNum;
+ private BigDecimal largeBatchCount;
/**
* NumericBetweenParametersProviderJdbc constructor.
@@ -78,16 +79,15 @@ public class JdbcNumericBetweenParametersProvider
implements JdbcParameterValues
checkArgument(batchSize > 0, "Batch size must be positive");
BigDecimal maxElemCount =
(maxVal.subtract(minVal)).add(BigDecimal.valueOf(1));
- if (BigDecimal.valueOf(batchSize).compareTo(maxElemCount) > 0) {
- batchSize = maxElemCount.longValue();
+ BigDecimal requestedBatchSize = BigDecimal.valueOf(batchSize);
+ if (requestedBatchSize.compareTo(maxElemCount) > 0) {
+ requestedBatchSize = maxElemCount;
}
- this.batchSize = batchSize;
- this.batchNum =
- new Double(
- Math.ceil(
-
(maxElemCount.divide(BigDecimal.valueOf(batchSize)))
- .doubleValue()))
- .intValue();
+ int numberOfBatches =
+ maxElemCount.divide(requestedBatchSize, 0,
RoundingMode.CEILING).intValueExact();
+ this.batchSize = requestedBatchSize;
+ this.batchNum = numberOfBatches;
+ this.largeBatchCount = BigDecimal.valueOf(this.batchNum);
return this;
}
@@ -99,36 +99,32 @@ public class JdbcNumericBetweenParametersProvider
implements JdbcParameterValues
batchNum = maxElemCount.intValue();
}
this.batchNum = batchNum;
- // For the presence of a decimal we take the integer up
- this.batchSize =
- (maxElemCount.divide(BigDecimal.valueOf(batchNum), 2,
RoundingMode.HALF_UP))
- .setScale(0, RoundingMode.CEILING)
- .longValue();
+ this.batchSize = maxElemCount.divide(BigDecimal.valueOf(batchNum), 0,
RoundingMode.CEILING);
+ // Distribute the remainder without rounding away keys or exceeding
the upper bound.
+ this.largeBatchCount =
+ maxElemCount.subtract(
+ this.batchSize
+ .subtract(BigDecimal.valueOf(1))
+ .multiply(BigDecimal.valueOf(batchNum)));
return this;
}
@Override
public Serializable[][] getParameterValues() {
checkState(
- batchSize > 0,
+ batchSize != null && batchSize.compareTo(BigDecimal.ZERO) > 0
&& batchNum > 0,
"Batch size and batch number must be positive. Have you called
`ofBatchSize` or `ofBatchNum`?");
- BigDecimal maxElemCount =
(maxVal.subtract(minVal)).add(BigDecimal.valueOf(1));
- BigDecimal bigBatchNum =
- maxElemCount
- .subtract(BigDecimal.valueOf(batchSize - 1))
- .multiply(BigDecimal.valueOf(batchNum));
-
Serializable[][] parameters = new Serializable[batchNum][2];
BigDecimal start = minVal;
for (int i = 0; i < batchNum; i++) {
- BigDecimal end =
- start.add(BigDecimal.valueOf(batchSize))
- .subtract(BigDecimal.valueOf(1))
- .subtract(
-
BigDecimal.valueOf(i).compareTo(bigBatchNum) >= 0
- ? BigDecimal.ONE
- : BigDecimal.ZERO);
+ BigDecimal end =
start.add(batchSize).subtract(BigDecimal.valueOf(1));
+ if (BigDecimal.valueOf(i).compareTo(largeBatchCount) >= 0) {
+ end = end.subtract(BigDecimal.valueOf(1));
+ }
+ if (end.compareTo(maxVal) > 0) {
+ end = maxVal;
+ }
parameters[i] = new BigDecimal[] {start, end};
start = end.add(BigDecimal.valueOf(1));
}
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/split/JdbcNumericBetweenParametersProviderTest.java
b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/split/JdbcNumericBetweenParametersProviderTest.java
new file mode 100644
index 0000000000..40161f94ee
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/split/JdbcNumericBetweenParametersProviderTest.java
@@ -0,0 +1,184 @@
+/*
+ * 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.internal.split;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.function.Executable;
+
+import java.io.Serializable;
+import java.math.BigDecimal;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class JdbcNumericBetweenParametersProviderTest {
+
+ private static final BigDecimal ONE = BigDecimal.ONE;
+ private static final BigDecimal DECIMAL_38_MAX =
+ new BigDecimal("99999999999999999999999999999999999999");
+
+ @Test
+ void splitsRangeByFetchSize() {
+ Serializable[][] actual =
+ new JdbcNumericBetweenParametersProvider(300, ONE,
BigDecimal.valueOf(1000))
+ .getParameterValues();
+ BigDecimal[][] expected = {
+ {ONE, BigDecimal.valueOf(300)},
+ {BigDecimal.valueOf(301), BigDecimal.valueOf(600)},
+ {BigDecimal.valueOf(601), BigDecimal.valueOf(900)},
+ {BigDecimal.valueOf(901), BigDecimal.valueOf(1000)}
+ };
+ assertPairs(expected, actual);
+ }
+
+ @Test
+ void clampsBatchSizeToSingletonRange() {
+ Serializable[][] actual =
+ new JdbcNumericBetweenParametersProvider(
+ BigDecimal.valueOf(5), BigDecimal.valueOf(5))
+ .ofBatchSize(3)
+ .getParameterValues();
+ BigDecimal[][] expected = {{BigDecimal.valueOf(5),
BigDecimal.valueOf(5)}};
+ assertPairs(expected, actual);
+ }
+
+ @Test
+ void clampsBatchNumToElementCount() {
+ Serializable[][] actual =
+ new JdbcNumericBetweenParametersProvider(
+ BigDecimal.valueOf(-2), BigDecimal.valueOf(2))
+ .ofBatchNum(10)
+ .getParameterValues();
+ BigDecimal[][] expected = {
+ {BigDecimal.valueOf(-2), BigDecimal.valueOf(-2)},
+ {BigDecimal.valueOf(-1), BigDecimal.valueOf(-1)},
+ {BigDecimal.ZERO, BigDecimal.ZERO},
+ {ONE, ONE},
+ {BigDecimal.valueOf(2), BigDecimal.valueOf(2)}
+ };
+ assertPairs(expected, actual);
+ }
+
+ @Test
+ void keepsExactBoundsForDecimal38Range() {
+ Serializable[][] actual =
+ new JdbcNumericBetweenParametersProvider(ONE, DECIMAL_38_MAX)
+ .ofBatchNum(3)
+ .getParameterValues();
+ assertEquals(3, actual.length);
+ assertEquals(ONE, actual[0][0]);
+ assertEquals(DECIMAL_38_MAX, actual[2][1]);
+ for (int i = 1; i < actual.length; i++) {
+ assertEquals(((BigDecimal) actual[i - 1][1]).add(ONE),
actual[i][0]);
+ assertTrue(((BigDecimal) actual[i][0]).compareTo((BigDecimal)
actual[i][1]) <= 0);
+ }
+ }
+
+ @Test
+ void rejectsNonPositiveBatchSize() {
+ assertMessage(
+ IllegalArgumentException.class,
+ "Batch size must be positive",
+ () -> new JdbcNumericBetweenParametersProvider(ONE,
BigDecimal.TEN).ofBatchSize(0));
+ }
+
+ @Test
+ void rejectsNonPositiveBatchCount() {
+ assertMessage(
+ IllegalArgumentException.class,
+ "Batch number must be positive",
+ () -> new JdbcNumericBetweenParametersProvider(ONE,
BigDecimal.TEN).ofBatchNum(0));
+ }
+
+ @Test
+ void rejectsReversedBounds() {
+ assertMessage(
+ IllegalArgumentException.class,
+ "minVal must not be larger than maxVal",
+ () -> new JdbcNumericBetweenParametersProvider(BigDecimal.TEN,
ONE));
+ }
+
+ @Test
+ void rejectsUnconfiguredState() {
+ assertMessage(
+ IllegalStateException.class,
+ "Batch size and batch number must be positive. Have you called
`ofBatchSize` or `ofBatchNum`?",
+ () ->
+ new JdbcNumericBetweenParametersProvider(ONE,
BigDecimal.TEN)
+ .getParameterValues());
+ }
+
+ @Test
+ void preservesFractionalBoundsWhenClampingToOneBatch() {
+ BigDecimal min = new BigDecimal("0.1");
+ BigDecimal max = new BigDecimal("0.2");
+ assertPairs(
+ new BigDecimal[][] {{min, max}},
+ new JdbcNumericBetweenParametersProvider(min, max)
+ .ofBatchNum(2)
+ .getParameterValues());
+ }
+
+ @Test
+ void replacesPreviousPartitionMode() {
+ JdbcNumericBetweenParametersProvider provider =
+ new JdbcNumericBetweenParametersProvider(ONE, BigDecimal.TEN);
+ provider.ofBatchNum(3).ofBatchSize(4);
+ assertPairs(
+ new BigDecimal[][] {
+ {ONE, BigDecimal.valueOf(4)},
+ {BigDecimal.valueOf(5), BigDecimal.valueOf(8)},
+ {BigDecimal.valueOf(9), BigDecimal.TEN}
+ },
+ provider.getParameterValues());
+ provider.ofBatchNum(3);
+ assertPairs(
+ new BigDecimal[][] {
+ {ONE, BigDecimal.valueOf(4)},
+ {BigDecimal.valueOf(5), BigDecimal.valueOf(7)},
+ {BigDecimal.valueOf(8), BigDecimal.TEN}
+ },
+ provider.getParameterValues());
+ }
+
+ @Test
+ void rejectsTooManyBatchesWithoutChangingConfiguration() {
+ BigDecimal min = BigDecimal.valueOf(Long.MIN_VALUE);
+ BigDecimal max = BigDecimal.valueOf(Long.MAX_VALUE);
+ JdbcNumericBetweenParametersProvider provider =
+ new JdbcNumericBetweenParametersProvider(min,
max).ofBatchNum(2);
+ assertMessage(ArithmeticException.class, "Overflow", () ->
provider.ofBatchSize(1));
+ assertPairs(
+ new BigDecimal[][] {{min, BigDecimal.valueOf(-1)},
{BigDecimal.ZERO, max}},
+ provider.getParameterValues());
+ }
+
+ private static void assertPairs(BigDecimal[][] expected, Serializable[][]
actual) {
+ assertEquals(expected.length, actual.length);
+ for (int i = 0; i < expected.length; i++) {
+ assertArrayEquals(expected[i], actual[i], "pair " + i);
+ }
+ }
+
+ private static void assertMessage(
+ Class<? extends Throwable> type, String message, Executable
executable) {
+ assertEquals(message, assertThrows(type, executable).getMessage());
+ }
+}
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 cdd4bb8e99..c1b230c2c9 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
@@ -26,6 +26,7 @@ import
org.apache.seatunnel.connectors.seatunnel.jdbc.config.JdbcConnectionConfi
import org.apache.seatunnel.connectors.seatunnel.jdbc.config.JdbcSourceConfig;
import
org.apache.seatunnel.connectors.seatunnel.jdbc.exception.JdbcConnectorException;
+import org.duckdb.DuckDBDriver;
import org.junit.jupiter.api.Test;
import lombok.extern.slf4j.Slf4j;
@@ -34,12 +35,27 @@ import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.math.BigDecimal;
+import java.sql.Connection;
import java.sql.PreparedStatement;
+import java.sql.ResultSet;
import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashSet;
+import java.util.List;
import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.LongStream;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -201,4 +217,156 @@ public class FixedChunkSplitterTest {
handler);
}
}
+
+ private static final String DUCKDB_URL = "jdbc:duckdb:";
+
+ @Test
+ public void testNumericSplitsReadEveryRowExactlyOnce() throws Exception {
+ try (Connection connection = new DuckDBDriver().connect(DUCKDB_URL,
new Properties());
+ DuckDbFixedChunkSplitter splitter =
+ new DuckDbFixedChunkSplitter(duckDbConfig(),
connection)) {
+ try (Statement statement = connection.createStatement()) {
+ statement.execute("CREATE TABLE t (id BIGINT)");
+ statement.execute("INSERT INTO t SELECT * FROM range(1,
2002)");
+ }
+
+ JdbcSourceTable table =
+ JdbcSourceTable.builder()
+ .tablePath(TablePath.of("main", "t"))
+ .query("SELECT id FROM t")
+ .partitionColumn("id")
+ .partitionStart("1")
+ .partitionEnd("2001")
+ .partitionNumber(1000)
+ .build();
+
+ Collection<JdbcSourceSplit> splits = splitter.createSplits(table,
splitKeyType());
+ List<Long> ids = readAllIds(splitter, splits);
+
+ assertEquals(2001, ids.size(), "every row must be read exactly
once");
+ List<Long> distinct = new ArrayList<>(new LinkedHashSet<>(ids));
+ assertEquals(2001, distinct.size(), "no row may be read twice");
+ Collections.sort(distinct);
+ assertEquals(
+ LongStream.rangeClosed(1,
2001).boxed().collect(Collectors.toList()), distinct);
+ }
+ }
+
+ @Test
+ public void testSignedBigIntBoundariesAreSplitWithoutNarrowing() throws
Exception {
+ try (Connection connection = new DuckDBDriver().connect(DUCKDB_URL,
new Properties());
+ DuckDbFixedChunkSplitter splitter =
+ new DuckDbFixedChunkSplitter(duckDbConfig(),
connection)) {
+ long[] boundaries = {Long.MIN_VALUE, -1L, 0L, Long.MAX_VALUE};
+ try (Statement statement = connection.createStatement()) {
+ statement.execute("CREATE TABLE t (id BIGINT)");
+ }
+ try (PreparedStatement insert =
+ connection.prepareStatement("INSERT INTO t (id) VALUES
(?)")) {
+ for (long value : boundaries) {
+ insert.setLong(1, value);
+ insert.addBatch();
+ }
+ insert.executeBatch();
+ }
+
+ JdbcSourceTable table =
+ JdbcSourceTable.builder()
+ .tablePath(TablePath.of("main", "t"))
+ .query("SELECT id FROM t")
+ .partitionColumn("id")
+ .partitionStart(String.valueOf(Long.MIN_VALUE))
+ .partitionEnd(String.valueOf(Long.MAX_VALUE))
+ .partitionNumber(2)
+ .build();
+
+ Collection<JdbcSourceSplit> splits = splitter.createSplits(table,
splitKeyType());
+ List<Long> ids = readAllIds(splitter, splits);
+
+ assertEquals(4, ids.size(), "each signed BIGINT boundary must be
read exactly once");
+ List<Long> sorted = new ArrayList<>(ids);
+ Collections.sort(sorted);
+ assertEquals(Arrays.asList(Long.MIN_VALUE, -1L, 0L,
Long.MAX_VALUE), sorted);
+ }
+ }
+
+ @Test
+ public void testConfiguredBoundsCoverRangeAndExcludeOutsiders() throws
Exception {
+ try (Connection connection = new DuckDBDriver().connect(DUCKDB_URL,
new Properties());
+ DuckDbFixedChunkSplitter splitter =
+ new DuckDbFixedChunkSplitter(duckDbConfig(),
connection)) {
+ try (Statement statement = connection.createStatement()) {
+ statement.execute("CREATE TABLE t (id BIGINT)");
+ statement.execute("INSERT INTO t SELECT * FROM range(0, 12)");
+ }
+
+ JdbcSourceTable table =
+ JdbcSourceTable.builder()
+ .tablePath(TablePath.of("main", "t"))
+ .query("SELECT id FROM t")
+ .partitionColumn("id")
+ .partitionStart("1")
+ .partitionEnd("10")
+ .partitionNumber(3)
+ .build();
+
+ Collection<JdbcSourceSplit> splits = splitter.createSplits(table,
splitKeyType());
+ List<Long> ids = readAllIds(splitter, splits);
+ assertEquals(10, ids.size());
+ Set<Long> distinct = new HashSet<>(ids);
+
+ assertEquals(
+ LongStream.rangeClosed(1,
10).boxed().collect(Collectors.toSet()),
+ distinct,
+ "the configured bounds must cover 1..10");
+ assertFalse(
+ distinct.contains(0L), "values below the configured start
must be excluded");
+ assertFalse(distinct.contains(11L), "values above the configured
end must be excluded");
+ }
+ }
+
+ private static List<Long> readAllIds(ChunkSplitter splitter,
Collection<JdbcSourceSplit> splits)
+ throws SQLException {
+ List<Long> ids = new ArrayList<>();
+ for (JdbcSourceSplit split : splits) {
+ try (PreparedStatement statement =
+ splitter.generateSplitStatement(split,
TableSchema.builder().build());
+ ResultSet resultSet = statement.executeQuery()) {
+ while (resultSet.next()) {
+ ids.add(resultSet.getLong(1));
+ }
+ }
+ }
+ return ids;
+ }
+
+ private static JdbcSourceConfig duckDbConfig() {
+ return JdbcSourceConfig.builder()
+ .jdbcConnectionConfig(
+ JdbcConnectionConfig.builder()
+ .url(DUCKDB_URL)
+ .driverName("org.duckdb.DuckDBDriver")
+ .build())
+ .build();
+ }
+
+ private static SeaTunnelRowType splitKeyType() {
+ return new SeaTunnelRowType(
+ new String[] {"id"}, new SeaTunnelDataType<?>[]
{BasicType.LONG_TYPE});
+ }
+
+ private static final class DuckDbFixedChunkSplitter extends
FixedChunkSplitter {
+
+ private final Connection connection;
+
+ private DuckDbFixedChunkSplitter(JdbcSourceConfig config, Connection
connection) {
+ super(config);
+ this.connection = connection;
+ }
+
+ @Override
+ protected PreparedStatement createPreparedStatement(String sql) throws
SQLException {
+ return connection.prepareStatement(sql);
+ }
+ }
}
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 e32dc8b593..fe850d5f9c 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
@@ -56,7 +56,9 @@ import java.sql.Connection;
import java.sql.Date;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
+import java.sql.ResultSet;
import java.sql.SQLException;
+import java.sql.Statement;
import java.sql.Timestamp;
import java.time.Duration;
import java.time.LocalDate;
@@ -65,8 +67,10 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@@ -791,4 +795,82 @@ public class JdbcMysqlSplitIT extends TestSuiteBase
implements TestResource {
}
}
}
+
+ @Test
+ public void testFixedNumericSplitsReadAllRows() throws Exception {
+ try (Connection connection = getJdbcConnection();
+ Statement statement = connection.createStatement()) {
+ statement.execute(
+ "CREATE TABLE "
+ + MYSQL_DATABASE
+ + ".fixed_numeric_uneven (id BIGINT PRIMARY KEY)");
+ try {
+ try (PreparedStatement insert =
+ connection.prepareStatement(
+ "INSERT INTO "
+ + MYSQL_DATABASE
+ + ".fixed_numeric_uneven (id) VALUES
(?)")) {
+ for (int id = 0; id <= 2002; id++) {
+ insert.setLong(1, id);
+ insert.addBatch();
+ }
+ insert.executeBatch();
+ }
+
+ Map<String, Object> configMap = new HashMap<>();
+ configMap.put("url", mysqlUrlInfo.getUrlWithDatabase().get());
+ configMap.put("driver", "com.mysql.cj.jdbc.Driver");
+ configMap.put("username", MYSQL_USERNAME);
+ configMap.put("password", MYSQL_PASSWORD);
+ configMap.put(
+ "query", "SELECT id FROM " + MYSQL_DATABASE +
".fixed_numeric_uneven");
+ configMap.put("partition_column", "id");
+
+ TablePath tablePath = TablePath.of(MYSQL_DATABASE,
"fixed_numeric_uneven");
+ try (MySqlCatalog mySqlCatalog =
+ new MySqlCatalog(
+ "mysql",
+ MYSQL_USERNAME,
+ MYSQL_PASSWORD,
+ mysqlUrlInfo,
+ null);
+ FixedChunkSplitter splitter =
getFixedChunkSplitter(configMap)) {
+ mySqlCatalog.open();
+ CatalogTable table = mySqlCatalog.getTable(tablePath);
+ JdbcSourceTable jdbcSourceTable =
+ JdbcSourceTable.builder()
+ .tablePath(tablePath)
+ .catalogTable(table)
+ .query(
+ "SELECT id FROM "
+ + MYSQL_DATABASE
+ + ".fixed_numeric_uneven")
+ .partitionColumn("id")
+ .partitionNumber(1000)
+ .partitionStart("1")
+ .partitionEnd("2001")
+ .build();
+
+ Set<Long> uniqueIds = new HashSet<>();
+ for (JdbcSourceSplit split :
splitter.generateSplits(jdbcSourceTable)) {
+ try (PreparedStatement splitStatement =
+ splitter.generateSplitStatement(
+ split, table.getTableSchema());
+ ResultSet resultSet =
splitStatement.executeQuery()) {
+ while (resultSet.next()) {
+ long id = resultSet.getLong(1);
+ Assertions.assertTrue(
+ id >= 1 && id <= 2001, "id out of
range: " + id);
+ Assertions.assertTrue(uniqueIds.add(id),
"duplicated id: " + id);
+ }
+ }
+ }
+ Assertions.assertEquals(2001, uniqueIds.size());
+ }
+ } finally {
+ statement.execute(
+ "DROP TABLE IF EXISTS " + MYSQL_DATABASE +
".fixed_numeric_uneven");
+ }
+ }
+ }
}
