Skip to content

Commit c4e8376

Browse files
eskabetxeeskabetxe
committed
[FLINK-38733] Add new SplitterEnumerator on JdbcSource
1 parent 6256816 commit c4e8376

7 files changed

Lines changed: 253 additions & 36 deletions

File tree

‎flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/JdbcSourceBuilder.java‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -305,6 +305,13 @@ public JdbcSource<OUT> build() {
305305
Preconditions.checkNotNull(resultExtractor, "'resultExtractor' mustn't be null.");
306306
Preconditions.checkNotNull(typeInformation, "'typeInformation' mustn't be null.");
307307

308+
Preconditions.checkArgument(
309+
!(this.splitterEnumerator == null && sql == null),
310+
"No splitter or query defined, one must be defined");
311+
Preconditions.checkArgument(
312+
!(this.splitterEnumerator != null && sql != null),
313+
"Splitter and query defined, use only one of them");
314+
308315
if (this.splitterEnumerator == null) {
309316
Preconditions.checkState(
310317
!StringUtils.isNullOrWhitespaceOnly(sql), "'sql' mustn't be null or empty.");

‎flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/splitter/PreparedSplitterEnumerator.java‎

Lines changed: 0 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -24,9 +24,6 @@
2424
import org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider;
2525
import org.apache.flink.util.Preconditions;
2626

27-
import org.slf4j.Logger;
28-
import org.slf4j.LoggerFactory;
29-
3027
import javax.annotation.Nullable;
3128

3229
import java.io.Serializable;
@@ -37,7 +34,6 @@
3734
/** A split enumerator based on sql-parameters grains. */
3835
@PublicEvolving
3936
public class PreparedSplitterEnumerator extends SqlSplitterEnumerator {
40-
public static final Logger LOG = LoggerFactory.getLogger(PreparedSplitterEnumerator.class);
4137

4238
private final Serializable[][] sqlParameters;
4339
private boolean finished;

‎flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/splitter/PreparedSplitterNumericParameters.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -63,7 +63,7 @@ public PreparedSplitterNumericParameters withBatchNum(int batchNum) {
6363
batchNum = (int) maxElemCount;
6464
}
6565
this.batchNum = batchNum;
66-
this.batchSize = new Double(Math.ceil((double) maxElemCount / batchNum)).longValue();
66+
this.batchSize = (maxElemCount + batchNum - 1) / batchNum;
6767
return this;
6868
}
6969

‎flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/splitter/SlideTimingSplitterEnumerator.java‎

Lines changed: 20 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -23,9 +23,6 @@
2323
import org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider;
2424
import org.apache.flink.util.Preconditions;
2525

26-
import org.slf4j.Logger;
27-
import org.slf4j.LoggerFactory;
28-
2926
import javax.annotation.Nonnull;
3027
import javax.annotation.Nullable;
3128

@@ -37,24 +34,23 @@
3734
/** A split enumerator based on sql-parameters grains. */
3835
@PublicEvolving
3936
public class SlideTimingSplitterEnumerator extends SqlSplitterEnumerator {
40-
public static final Logger LOG = LoggerFactory.getLogger(SlideTimingSplitterEnumerator.class);
4137

42-
private final long slideStepMills;
43-
private final long slideSpanMills;
38+
private final long slideStepMillis;
39+
private final long slideSpanMillis;
4440
private final long splitGenerateDelayMillis;
4541

46-
private @Nonnull Long startMills;
42+
private @Nonnull Long startMillis;
4743

4844
protected SlideTimingSplitterEnumerator(
4945
String sqlTemplate,
5046
Long startMills,
51-
long slideSpanMills,
52-
long slideStepMills,
47+
long slideSpanMillis,
48+
long slideStepMillis,
5349
long splitGenerateDelayMillis) {
5450
super(sqlTemplate);
55-
this.startMills = Preconditions.checkNotNull(startMills);
56-
this.slideStepMills = slideStepMills;
57-
this.slideSpanMills = slideSpanMills;
51+
this.startMillis = Preconditions.checkNotNull(startMills);
52+
this.slideStepMillis = slideStepMillis;
53+
this.slideSpanMillis = slideSpanMillis;
5854
this.splitGenerateDelayMillis = splitGenerateDelayMillis;
5955
}
6056

@@ -72,31 +68,31 @@ public boolean isAllSplitsFinished() {
7268
@VisibleForTesting
7369
public Serializable[][] getSqlParameters() {
7470
List<Serializable[]> tmpList = new ArrayList<>();
75-
while (nextSplitAvailable(startMills)) {
76-
Serializable[] params = new Serializable[] {startMills, startMills + slideSpanMills};
71+
while (nextSplitAvailable(startMillis)) {
72+
Serializable[] params = new Serializable[] {startMillis, startMillis + slideSpanMillis};
7773
tmpList.add(params);
78-
startMills += slideStepMills;
74+
startMillis += slideStepMillis;
7975
}
8076
return tmpList.toArray(new Serializable[0][]);
8177
}
8278

8379
@Override
8480
public @Nullable Serializable serializableState() {
85-
return this.startMills;
81+
return this.startMillis;
8682
}
8783

8884
@Override
8985
public SlideTimingSplitterEnumerator restoreState(Serializable state) {
9086
Preconditions.checkArgument((Long) state > 0L);
91-
this.startMills = (Long) state;
87+
this.startMillis = (Long) state;
9288
return this;
9389
}
9490

9591
private boolean nextSplitAvailable(Long nextSpanStartMillis) {
9692
final long delayedNextSpanStartMillis = nextSpanStartMillis + splitGenerateDelayMillis;
9793
final long currentAvailableMillis = currentAvailableMillis();
9894
return currentAvailableMillis >= delayedNextSpanStartMillis
99-
&& (currentAvailableMillis - delayedNextSpanStartMillis >= slideSpanMills);
95+
&& (currentAvailableMillis - delayedNextSpanStartMillis >= slideSpanMillis);
10096
}
10197

10298
private Long currentAvailableMillis() {
@@ -112,20 +108,20 @@ public boolean equals(Object o) {
112108
return false;
113109
}
114110
SlideTimingSplitterEnumerator that = (SlideTimingSplitterEnumerator) o;
115-
return slideStepMills == that.slideStepMills
116-
&& slideSpanMills == that.slideSpanMills
111+
return slideStepMillis == that.slideStepMillis
112+
&& slideSpanMillis == that.slideSpanMillis
117113
&& splitGenerateDelayMillis == that.splitGenerateDelayMillis
118-
&& Objects.equals(startMills, that.startMills);
114+
&& Objects.equals(startMillis, that.startMillis);
119115
}
120116

121117
@Override
122118
public int hashCode() {
123119
return Objects.hash(
124120
super.hashCode(),
125-
slideStepMills,
126-
slideSpanMills,
121+
slideStepMillis,
122+
slideSpanMillis,
127123
splitGenerateDelayMillis,
128-
startMills);
124+
startMillis);
129125
}
130126

131127
public static SlideTimingSplitterEnumeratorBuilder builder() {

‎flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/splitter/SqlSplitterEnumerator.java‎

Lines changed: 0 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -23,9 +23,6 @@
2323
import org.apache.flink.connector.jdbc.core.datastream.source.split.JdbcSourceSplit;
2424
import org.apache.flink.util.Preconditions;
2525

26-
import org.slf4j.Logger;
27-
import org.slf4j.LoggerFactory;
28-
2926
import java.io.Serializable;
3027
import java.util.ArrayList;
3128
import java.util.Collections;
@@ -34,7 +31,6 @@
3431

3532
/** A sql base template split enumerator. */
3633
public abstract class SqlSplitterEnumerator implements SplitterEnumerator {
37-
public static final Logger LOG = LoggerFactory.getLogger(SqlSplitterEnumerator.class);
3834

3935
private final char[] currentId = "0000000000".toCharArray();
4036

‎flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/splitter/PreparedSplitterNumericParametersTest.java‎

Lines changed: 21 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,21 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing, software
13+
* distributed under the License is distributed on an "AS IS" BASIS,
14+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15+
* See the License for the specific language governing permissions and
16+
* limitations under the License.
17+
*/
18+
119
package org.apache.flink.connector.jdbc.core.datastream.source.enumerator.splitter;
220

321
import org.junit.jupiter.api.Test;
@@ -98,9 +116,9 @@ void testBatchMaxMinTooLarge() {
98116
.getParameterValues();
99117

100118
long[][] expected = {
101-
new long[] {2260418954055131340L, 2798685988449068491L},
102-
new long[] {2798685988449068492L, 3336953022843005643L},
103-
new long[] {3336953022843005644L, 3875220057236942795L}
119+
new long[] {2260418954055131340L, 2798685988449068510L},
120+
new long[] {2798685988449068511L, 3336953022843005680L},
121+
new long[] {3336953022843005681L, 3875220057236942850L}
104122
};
105123
check(expected, parameters);
106124
}

0 commit comments

Comments
 (0)