This is an automated email from the ASF dual-hosted git repository.
MartijnVisser pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-jdbc.git
The following commit(s) were added to refs/heads/main by this push:
new 2d1f830c [FLINK-40748][jdbc] Keep PreparedSplitterNumericParameters
inside maxVal
2d1f830c is described below
commit 2d1f830c058648b6b1ea993de1c3e77be4fc3ef8
Author: Martijn Visser <[email protected]>
AuthorDate: Tue Sep 22 10:40:49 2026 +0200
[FLINK-40748][jdbc] Keep PreparedSplitterNumericParameters inside maxVal
- The last split of withBatchSize ends at maxVal, as
JdbcNumericBetweenParametersProvider does since FLINK-38046
- minVal > maxVal is rejected in the constructor
Generated-by: Claude Code (Claude Fable 5.1, Claude Opus 5.5)
---
.../PreparedSplitterNumericParameters.java | 5 +-
.../PreparedSplitterNumericParametersTest.java | 60 ++++++++++++++++++++++
2 files changed, 64 insertions(+), 1 deletion(-)
diff --git
a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/splitter/PreparedSplitterNumericParameters.java
b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/splitter/PreparedSplitterNumericParameters.java
index c48763c3..c24ed9a6 100644
---
a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/splitter/PreparedSplitterNumericParameters.java
+++
b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/splitter/PreparedSplitterNumericParameters.java
@@ -37,6 +37,7 @@ public class PreparedSplitterNumericParameters implements
Serializable {
private int batchNum;
public PreparedSplitterNumericParameters(long minVal, long maxVal) {
+ Preconditions.checkArgument(minVal <= maxVal, "minVal must not be
larger than maxVal");
this.minVal = minVal;
this.maxVal = maxVal;
this.batchNum = 0;
@@ -77,11 +78,13 @@ public class PreparedSplitterNumericParameters implements
Serializable {
Serializable[][] parameters = new Serializable[batchNum][2];
long start = minVal;
- for (int i = 0; i < batchNum; i++) {
+ for (int i = 0; i < batchNum - 1; i++) {
long end = start + batchSize - 1 - (i >= bigBatchNum ? 1 : 0);
parameters[i] = new Long[] {start, end};
start = end + 1;
}
+ // the last split always ends exactly at maxVal
+ parameters[batchNum - 1] = new Long[] {start, maxVal};
return parameters;
}
}
diff --git
a/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/splitter/PreparedSplitterNumericParametersTest.java
b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/splitter/PreparedSplitterNumericParametersTest.java
index 0b21d40c..9e3aa616 100644
---
a/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/splitter/PreparedSplitterNumericParametersTest.java
+++
b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/splitter/PreparedSplitterNumericParametersTest.java
@@ -21,8 +21,10 @@ package
org.apache.flink.connector.jdbc.core.datastream.source.enumerator.splitt
import org.junit.jupiter.api.Test;
import java.io.Serializable;
+import java.util.Arrays;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
class PreparedSplitterNumericParametersTest {
@@ -123,6 +125,64 @@ class PreparedSplitterNumericParametersTest {
check(expected, parameters);
}
+ @Test
+ void testBatchSizeNeverReadsPastMaxVal() {
+ // 5 values in batches of 4: two splits, and the second must stop at
9, not at 10
+ Serializable[][] parameters =
+ new PreparedSplitterNumericParameters(5,
9).withBatchSize(4).getParameterValues();
+
+ long[][] expected = {new long[] {5, 7}, new long[] {8, 9}};
+ check(expected, parameters);
+ }
+
+ @Test
+ void testSplitsCoverTheRangeExactly() {
+ // every batch size and batch number for every range up to 40 values:
contiguous splits
+ // from minVal to maxVal
+ for (long minVal = -3; minVal <= 3; minVal += 3) {
+ for (int count = 1; count <= 40; count++) {
+ long maxVal = minVal + count - 1;
+ for (int k = 1; k <= 45; k++) {
+ assertCoversRange(
+ minVal,
+ maxVal,
+ new PreparedSplitterNumericParameters(minVal,
maxVal)
+ .withBatchSize(k)
+ .getParameterValues());
+ assertCoversRange(
+ minVal,
+ maxVal,
+ new PreparedSplitterNumericParameters(minVal,
maxVal)
+ .withBatchNum(k)
+ .getParameterValues());
+ }
+ }
+ }
+ }
+
+ @Test
+ void testMinValMustNotExceedMaxVal() {
+ assertThatThrownBy(() -> new PreparedSplitterNumericParameters(10, 9))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("minVal must not be larger than maxVal");
+ }
+
+ private static void assertCoversRange(long minVal, long maxVal,
Serializable[][] splits) {
+ long next = minVal;
+ for (Serializable[] split : splits) {
+ long start = (Long) split[0];
+ long end = (Long) split[1];
+ assertThat(start)
+ .as("start of %s for [%s, %s]", Arrays.toString(split),
minVal, maxVal)
+ .isEqualTo(next);
+ assertThat(end)
+ .as("end of %s for [%s, %s]", Arrays.toString(split),
minVal, maxVal)
+ .isGreaterThanOrEqualTo(start);
+ next = end + 1;
+ }
+ assertThat(next - 1).as("last end for [%s, %s]", minVal,
maxVal).isEqualTo(maxVal);
+ }
+
private void check(long[][] expected, Serializable[][] actual) {
assertThat(actual).hasDimensions(expected.length, expected[0].length);
for (int i = 0; i < expected.length; i++) {
