This is an automated email from the ASF dual-hosted git repository.
RocMarshal 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 af994651 [FLINK-38733] Add new SplitterEnumerator on JdbcSource (#180)
af994651 is described below
commit af99465109ebafdf77b4c65b9f4526989ad93cda
Author: Joao Boto <[email protected]>
AuthorDate: Thu Jun 18 07:10:45 2026 +0200
[FLINK-38733] Add new SplitterEnumerator on JdbcSource (#180)
---
.../jdbc/core/datastream/source/JdbcSource.java | 62 ++++---
.../core/datastream/source/JdbcSourceBuilder.java | 92 +++++++---
.../config/ContinuousUnBoundingSettings.java | 1 +
.../source/enumerator/JdbcSourceEnumerator.java | 168 +++++++++---------
.../enumerator/JdbcSqlSplitEnumeratorBase.java | 8 +
.../enumerator/SqlTemplateSplitEnumerator.java | 9 +-
.../splitter/JdbcSqlSplitterEnumerator.java | 141 ++++++++++++++++
.../splitter/PreparedSplitterEnumerator.java | 135 +++++++++++++++
.../PreparedSplitterNumericParameters.java} | 70 ++------
.../splitter/SlideTimingSplitterEnumerator.java | 187 +++++++++++++++++++++
.../enumerator/splitter/SplitterEnumerator.java | 73 ++++++++
.../enumerator/splitter/SqlSplitterEnumerator.java | 106 ++++++++++++
.../flink/connector/jdbc/lineage/LineageUtils.java | 8 +-
.../CompositeJdbcParameterValuesProvider.java | 1 +
.../split/JdbcGenericParameterValuesProvider.java | 5 +
.../JdbcNumericBetweenParametersProvider.java | 6 +
.../jdbc/split/JdbcParameterValuesProvider.java | 4 +
.../split/JdbcSlideTimingParameterProvider.java | 11 +-
.../datastream/source/JdbcSourceBuilderTest.java | 48 ++++--
.../source/JdbcSourceStreamRelatedITCase.java | 1 +
.../enumerator/JdbcSourceEnumeratorTest.java | 43 ++++-
.../PreparedSplitterNumericParametersTest.java | 134 +++++++++++++++
.../SlideTimingSplitterEnumeratorTest.java | 142 ++++++++++++++++
23 files changed, 1233 insertions(+), 222 deletions(-)
diff --git
a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/JdbcSource.java
b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/JdbcSource.java
index c861d676..427c9669 100644
---
a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/JdbcSource.java
+++
b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/JdbcSource.java
@@ -35,7 +35,8 @@ import
org.apache.flink.connector.jdbc.core.datastream.source.enumerator.JdbcSou
import
org.apache.flink.connector.jdbc.core.datastream.source.enumerator.JdbcSourceEnumerator;
import
org.apache.flink.connector.jdbc.core.datastream.source.enumerator.JdbcSourceEnumeratorState;
import
org.apache.flink.connector.jdbc.core.datastream.source.enumerator.JdbcSqlSplitEnumeratorBase;
-import
org.apache.flink.connector.jdbc.core.datastream.source.enumerator.SqlTemplateSplitEnumerator;
+import
org.apache.flink.connector.jdbc.core.datastream.source.enumerator.splitter.JdbcSqlSplitterEnumerator;
+import
org.apache.flink.connector.jdbc.core.datastream.source.enumerator.splitter.SplitterEnumerator;
import
org.apache.flink.connector.jdbc.core.datastream.source.reader.JdbcSourceReader;
import
org.apache.flink.connector.jdbc.core.datastream.source.reader.JdbcSourceSplitReader;
import
org.apache.flink.connector.jdbc.core.datastream.source.reader.extractor.ResultExtractor;
@@ -66,17 +67,16 @@ public class JdbcSource<OUT>
Source<OUT, JdbcSourceSplit, JdbcSourceEnumeratorState>,
ResultTypeQueryable<OUT> {
- private final Boundedness boundedness;
private final TypeInformation<OUT> typeInformation;
- private final @Nullable ContinuousUnBoundingSettings
continuousUnBoundingSettings;
private final Configuration configuration;
- private final JdbcSqlSplitEnumeratorBase.Provider<JdbcSourceSplit>
sqlSplitEnumeratorProvider;
+ private final SplitterEnumerator splitterEnumerator;
protected JdbcConnectionProvider connectionProvider;
private final ResultExtractor<OUT> resultExtractor;
private final DeliveryGuarantee deliveryGuarantee;
+ @Deprecated
JdbcSource(
Configuration configuration,
JdbcConnectionProvider connectionProvider,
@@ -85,23 +85,35 @@ public class JdbcSource<OUT>
TypeInformation<OUT> typeInformation,
@Nullable DeliveryGuarantee deliveryGuarantee,
@Nullable ContinuousUnBoundingSettings
continuousUnBoundingSettings) {
+ this(
+ configuration,
+ connectionProvider,
+ new JdbcSqlSplitterEnumerator(
+ sqlSplitEnumeratorProvider,
continuousUnBoundingSettings),
+ resultExtractor,
+ typeInformation,
+ deliveryGuarantee);
+ }
+
+ JdbcSource(
+ Configuration configuration,
+ JdbcConnectionProvider connectionProvider,
+ SplitterEnumerator splitterEnumerator,
+ ResultExtractor<OUT> resultExtractor,
+ TypeInformation<OUT> typeInformation,
+ @Nullable DeliveryGuarantee deliveryGuarantee) {
this.configuration = Preconditions.checkNotNull(configuration);
this.connectionProvider =
Preconditions.checkNotNull(connectionProvider);
- this.sqlSplitEnumeratorProvider =
Preconditions.checkNotNull(sqlSplitEnumeratorProvider);
+ this.splitterEnumerator =
Preconditions.checkNotNull(splitterEnumerator);
this.resultExtractor = Preconditions.checkNotNull(resultExtractor);
this.deliveryGuarantee =
Objects.isNull(deliveryGuarantee) ? DeliveryGuarantee.NONE :
deliveryGuarantee;
this.typeInformation = Preconditions.checkNotNull(typeInformation);
- this.continuousUnBoundingSettings = continuousUnBoundingSettings;
- this.boundedness =
- Objects.isNull(continuousUnBoundingSettings)
- ? Boundedness.BOUNDED
- : Boundedness.CONTINUOUS_UNBOUNDED;
}
@Override
public Boundedness getBoundedness() {
- return boundedness;
+ return splitterEnumerator.getBoundedness();
}
@Override
@@ -124,10 +136,7 @@ public class JdbcSource<OUT>
public SplitEnumerator<JdbcSourceSplit, JdbcSourceEnumeratorState>
createEnumerator(
SplitEnumeratorContext<JdbcSourceSplit> enumContext) throws
Exception {
return new JdbcSourceEnumerator(
- enumContext,
- sqlSplitEnumeratorProvider.create(),
- continuousUnBoundingSettings,
- new ArrayList<>());
+ enumContext, splitterEnumerator, connectionProvider, new
ArrayList<>());
}
@Override
@@ -139,8 +148,8 @@ public class JdbcSource<OUT>
checkpoint.getOptionalUserDefinedSplitEnumeratorState();
return new JdbcSourceEnumerator(
enumContext,
-
sqlSplitEnumeratorProvider.restore(optionalUserDefinedSplitEnumeratorState),
- continuousUnBoundingSettings,
+
splitterEnumerator.restoreState(optionalUserDefinedSplitEnumeratorState),
+ connectionProvider,
checkpoint.getRemainingSplits());
}
@@ -167,8 +176,8 @@ public class JdbcSource<OUT>
// ---- Visible for testing methods. ---
@VisibleForTesting
- public JdbcSqlSplitEnumeratorBase.Provider<JdbcSourceSplit>
getSqlSplitEnumeratorProvider() {
- return sqlSplitEnumeratorProvider;
+ public SplitterEnumerator getSplitterEnumerator() {
+ return splitterEnumerator;
}
@VisibleForTesting
@@ -201,27 +210,24 @@ public class JdbcSource<OUT>
return false;
}
JdbcSource<?> that = (JdbcSource<?>) o;
- return boundedness == that.boundedness
- && Objects.equals(typeInformation, that.typeInformation)
+ return Objects.equals(typeInformation, that.typeInformation)
&& Objects.equals(configuration, that.configuration)
- && Objects.equals(sqlSplitEnumeratorProvider,
that.sqlSplitEnumeratorProvider)
+ && Objects.equals(splitterEnumerator, that.splitterEnumerator)
&& Objects.equals(connectionProvider, that.connectionProvider)
&& Objects.equals(resultExtractor, that.resultExtractor)
- && deliveryGuarantee == that.deliveryGuarantee
- && Objects.equals(continuousUnBoundingSettings,
that.continuousUnBoundingSettings);
+ && deliveryGuarantee == that.deliveryGuarantee;
}
@Override
public LineageVertex getLineageVertex() {
DefaultTypeDatasetFacet defaultTypeDatasetFacet =
new DefaultTypeDatasetFacet(getTypeInformation());
- SqlTemplateSplitEnumerator enumerator =
- (SqlTemplateSplitEnumerator)
sqlSplitEnumeratorProvider.create();
- Optional<String> nameOpt =
LineageUtils.tableNameOf(enumerator.getSqlTemplate(), true);
+ Optional<String> nameOpt =
+ LineageUtils.tableNameOf(splitterEnumerator.lineageQueries(),
true);
String namespace = LineageUtils.namespaceOf(connectionProvider);
LineageDataset dataset =
LineageUtils.datasetOf(
nameOpt.orElse(""), namespace,
Arrays.asList(defaultTypeDatasetFacet));
- return LineageUtils.sourceLineageVertexOf(boundedness,
Collections.singleton(dataset));
+ return LineageUtils.sourceLineageVertexOf(getBoundedness(),
Collections.singleton(dataset));
}
}
diff --git
a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/JdbcSourceBuilder.java
b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/JdbcSourceBuilder.java
index 87b39f40..fbd171fe 100644
---
a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/JdbcSourceBuilder.java
+++
b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/JdbcSourceBuilder.java
@@ -24,7 +24,10 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import
org.apache.flink.connector.jdbc.core.datastream.source.config.ContinuousUnBoundingSettings;
+import
org.apache.flink.connector.jdbc.core.datastream.source.enumerator.JdbcSqlSplitEnumeratorBase;
import
org.apache.flink.connector.jdbc.core.datastream.source.enumerator.SqlTemplateSplitEnumerator;
+import
org.apache.flink.connector.jdbc.core.datastream.source.enumerator.splitter.JdbcSqlSplitterEnumerator;
+import
org.apache.flink.connector.jdbc.core.datastream.source.enumerator.splitter.SplitterEnumerator;
import
org.apache.flink.connector.jdbc.core.datastream.source.reader.extractor.ResultExtractor;
import
org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider;
import
org.apache.flink.connector.jdbc.datasource.connections.SimpleJdbcConnectionProvider;
@@ -51,7 +54,7 @@ import java.util.Objects;
*
* <pre><code>
* JdbcSource<Row> source = JdbcSource.<Row>builder()
- * .setSql(validSql)
+ * .setSplitter(PreparedSplitterEnumerator.of(validSql))
* .setResultExtractor(new RowResultExtractor())
* .setDBUrl(dbUrl)
* .setDriverName(driverName)
@@ -67,6 +70,7 @@ import java.util.Objects;
*
* <pre><code>
*
+ * String query = "select * from books WHERE author = ?"
* Serializable[][] queryParameters = new String[2][1];
* queryParameters[0] = new String[]{"Kumar"};
* queryParameters[1] = new String[]{"Tan Ah Teck"};
@@ -78,13 +82,12 @@ import java.util.Objects;
* .setUsername(username)
* .setDriverName("org.apache.derby.jdbc.EmbeddedDriver")
* .setDBUrl("jdbc:derby:memory:ebookshop")
- * .setSql("select * from books WHERE author = ?")
- * .setJdbcParameterValuesProvider(new
JdbcGenericParameterValuesProvider(queryParameters))
+ * .setSplitter(PreparedSplitterEnumerator.of(query, queryParameters))
* .build();
* </code></pre>
*
* @see Row
- * @see JdbcParameterValuesProvider
+ * @see SplitterEnumerator
* @see PreparedStatement
* @see DriverManager
* @see JdbcSource
@@ -118,7 +121,7 @@ public class JdbcSourceBuilder<OUT> {
private JdbcParameterValuesProvider jdbcParameterValuesProvider;
private @Nullable Serializable optionalSqlSplitEnumeratorState;
private ResultExtractor<OUT> resultExtractor;
-
+ private SplitterEnumerator splitterEnumerator;
private JdbcConnectionProvider connectionProvider;
JdbcSourceBuilder() {
@@ -132,6 +135,13 @@ public class JdbcSourceBuilder<OUT> {
this.autoCommit = true;
}
+ public JdbcSourceBuilder<OUT> setSplitter(SplitterEnumerator
splitterEnumerator) {
+ this.splitterEnumerator = splitterEnumerator;
+ return this;
+ }
+
+ /** @deprecated Use {@link #setSplitter(SplitterEnumerator)} instead. */
+ @Deprecated
public JdbcSourceBuilder<OUT> setSql(@Nonnull String sql) {
Preconditions.checkArgument(
!StringUtils.isNullOrWhitespaceOnly(sql),
@@ -196,7 +206,10 @@ public class JdbcSourceBuilder<OUT> {
/**
* If the value was set as an instance of {@link
JdbcSlideTimingParameterProvider}, it's
* required to specify the {@link #continuousUnBoundingSettings}.
+ *
+ * @deprecated Use {@link #setSplitter(SplitterEnumerator)} instead.
*/
+ @Deprecated
public JdbcSourceBuilder<OUT> setJdbcParameterValuesProvider(
@Nonnull JdbcParameterValuesProvider parameterValuesProvider) {
this.jdbcParameterValuesProvider =
Preconditions.checkNotNull(parameterValuesProvider);
@@ -289,36 +302,63 @@ public class JdbcSourceBuilder<OUT> {
JdbcSourceOptions.READER_FETCH_BATCH_SIZE,
splitReaderFetchBatchSize);
this.configuration.set(JdbcSourceOptions.AUTO_COMMIT, autoCommit);
- Preconditions.checkState(
- !StringUtils.isNullOrWhitespaceOnly(sql), "'sql' mustn't be
null or empty.");
Preconditions.checkNotNull(resultExtractor, "'resultExtractor' mustn't
be null.");
Preconditions.checkNotNull(typeInformation, "'typeInformation' mustn't
be null.");
- if (Objects.nonNull(continuousUnBoundingSettings)) {
- Preconditions.checkArgument(
- Objects.nonNull(jdbcParameterValuesProvider)
- && jdbcParameterValuesProvider
- instanceof
JdbcSlideTimingParameterProvider,
- INVALID_SLIDE_TIMING_CONTINUOUS_HINT);
- }
-
- if (Objects.nonNull(jdbcParameterValuesProvider)
- && jdbcParameterValuesProvider instanceof
JdbcSlideTimingParameterProvider) {
- Preconditions.checkArgument(
- Objects.nonNull(continuousUnBoundingSettings),
- INVALID_CONTINUOUS_SLIDE_TIMING_HINT);
+ Preconditions.checkArgument(
+ !(this.splitterEnumerator == null && sql == null),
+ "No splitter or query defined, one must be defined");
+ Preconditions.checkArgument(
+ !(this.splitterEnumerator != null && sql != null),
+ "Splitter and query defined, use only one of them");
+
+ if (this.splitterEnumerator == null) {
+ Preconditions.checkState(
+ !StringUtils.isNullOrWhitespaceOnly(sql), "'sql' mustn't
be null or empty.");
+
+ if (Objects.nonNull(continuousUnBoundingSettings)) {
+ Preconditions.checkArgument(
+ Objects.nonNull(jdbcParameterValuesProvider)
+ && jdbcParameterValuesProvider
+ instanceof
JdbcSlideTimingParameterProvider,
+ INVALID_SLIDE_TIMING_CONTINUOUS_HINT);
+ }
+
+ if (Objects.nonNull(jdbcParameterValuesProvider)
+ && jdbcParameterValuesProvider instanceof
JdbcSlideTimingParameterProvider) {
+ Preconditions.checkArgument(
+ Objects.nonNull(continuousUnBoundingSettings),
+ INVALID_CONTINUOUS_SLIDE_TIMING_HINT);
+ }
+
+ this.splitterEnumerator =
+ getSplitter(
+ sql,
+ jdbcParameterValuesProvider,
+ optionalSqlSplitEnumeratorState,
+ continuousUnBoundingSettings);
}
return new JdbcSource<>(
configuration,
connectionProvider,
- new
SqlTemplateSplitEnumerator.TemplateSqlSplitEnumeratorProvider()
-
.setOptionalSqlSplitEnumeratorState(optionalSqlSplitEnumeratorState)
- .setSqlTemplate(sql)
-
.setParameterValuesProvider(jdbcParameterValuesProvider),
+ splitterEnumerator,
resultExtractor,
typeInformation,
- deliveryGuarantee,
- continuousUnBoundingSettings);
+ deliveryGuarantee);
+ }
+
+ private SplitterEnumerator getSplitter(
+ String sqlTemplate,
+ JdbcParameterValuesProvider parameterProvider,
+ Serializable userDefinedState,
+ @Nullable ContinuousUnBoundingSettings continuousSettings) {
+ JdbcSqlSplitEnumeratorBase.Provider<?> provider =
+ new
SqlTemplateSplitEnumerator.TemplateSqlSplitEnumeratorProvider()
+ .setOptionalSqlSplitEnumeratorState(userDefinedState)
+ .setSqlTemplate(sqlTemplate)
+ .setParameterValuesProvider(parameterProvider);
+
+ return new JdbcSqlSplitterEnumerator(provider, continuousSettings);
}
}
diff --git
a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/config/ContinuousUnBoundingSettings.java
b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/config/ContinuousUnBoundingSettings.java
index da4a8d92..e0a4c1a7 100644
---
a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/config/ContinuousUnBoundingSettings.java
+++
b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/config/ContinuousUnBoundingSettings.java
@@ -31,6 +31,7 @@ import static
org.apache.flink.util.Preconditions.checkNotNull;
* continuous discovery and streaming mode.
*/
@PublicEvolving
+@Deprecated
public class ContinuousUnBoundingSettings implements Serializable {
private static final long serialVersionUID = 1L;
diff --git
a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/JdbcSourceEnumerator.java
b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/JdbcSourceEnumerator.java
index c545a08f..d371fd9f 100644
---
a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/JdbcSourceEnumerator.java
+++
b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/JdbcSourceEnumerator.java
@@ -18,12 +18,12 @@
package org.apache.flink.connector.jdbc.core.datastream.source.enumerator;
-import org.apache.flink.api.connector.source.Boundedness;
import org.apache.flink.api.connector.source.SourceEvent;
import org.apache.flink.api.connector.source.SplitEnumerator;
import org.apache.flink.api.connector.source.SplitEnumeratorContext;
-import
org.apache.flink.connector.jdbc.core.datastream.source.config.ContinuousUnBoundingSettings;
+import
org.apache.flink.connector.jdbc.core.datastream.source.enumerator.splitter.SplitterEnumerator;
import
org.apache.flink.connector.jdbc.core.datastream.source.split.JdbcSourceSplit;
+import
org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider;
import org.apache.flink.util.Preconditions;
import org.slf4j.Logger;
@@ -35,11 +35,9 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
-import java.util.LinkedHashMap;
import java.util.List;
-import java.util.Map;
-import java.util.Objects;
import java.util.Optional;
+import java.util.concurrent.atomic.AtomicInteger;
/** JDBC source enumerator. */
public class JdbcSourceEnumerator
@@ -47,50 +45,32 @@ public class JdbcSourceEnumerator
private static final Logger LOG =
LoggerFactory.getLogger(JdbcSourceEnumerator.class);
private final SplitEnumeratorContext<JdbcSourceSplit> context;
- private final Boundedness boundedness;
- private final LinkedHashMap<Integer, String> readersAwaitingSplit;
private final List<JdbcSourceSplit> unassigned;
- private final JdbcSqlSplitEnumeratorBase<JdbcSourceSplit>
sqlSplitEnumerator;
- private final @Nullable ContinuousUnBoundingSettings
continuousUnBoundingSettings;
+ private final SplitterEnumerator splitterEnumerator;
+ private final JdbcConnectionProvider connectionProvider;
+ private final List<Integer> readersWaitingForSplits = new ArrayList<>();
+ private final AtomicInteger asyncCallsPending = new AtomicInteger(0);
public JdbcSourceEnumerator(
SplitEnumeratorContext<JdbcSourceSplit> context,
- JdbcSqlSplitEnumeratorBase<JdbcSourceSplit> sqlSplitEnumerator,
- ContinuousUnBoundingSettings continuousUnBoundingSettings,
+ SplitterEnumerator splitterEnumerator,
+ JdbcConnectionProvider connectionProvider,
List<JdbcSourceSplit> unassigned) {
this.context = Preconditions.checkNotNull(context);
- this.sqlSplitEnumerator =
Preconditions.checkNotNull(sqlSplitEnumerator);
- this.continuousUnBoundingSettings = continuousUnBoundingSettings;
- this.boundedness =
- Objects.isNull(continuousUnBoundingSettings)
- ? Boundedness.BOUNDED
- : Boundedness.CONTINUOUS_UNBOUNDED;
+ this.splitterEnumerator =
Preconditions.checkNotNull(splitterEnumerator);
this.unassigned = Preconditions.checkNotNull(unassigned);
- this.readersAwaitingSplit = new LinkedHashMap<>();
+ this.connectionProvider = connectionProvider;
}
@Override
public void start() {
- sqlSplitEnumerator.open();
- if (boundedness == Boundedness.CONTINUOUS_UNBOUNDED
- && Objects.nonNull(continuousUnBoundingSettings)) {
- context.callAsync(
- () -> sqlSplitEnumerator.enumerateSplits(() -> 1024 -
unassigned.size() > 0),
- this::processNewSplits,
-
continuousUnBoundingSettings.getInitialDiscoveryDelay().toMillis(),
-
continuousUnBoundingSettings.getDiscoveryInterval().toMillis());
- } else {
- try {
- unassigned.addAll(sqlSplitEnumerator.enumerateSplits(() ->
true));
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
+ splitterEnumerator.start(connectionProvider);
+ preDiscoverSplits();
}
@Override
public void close() throws IOException {
- sqlSplitEnumerator.close();
+ splitterEnumerator.close();
}
@Override
@@ -100,11 +80,20 @@ public class JdbcSourceEnumerator
@Override
public void handleSplitRequest(int subtask, @Nullable String hostname) {
- if (boundedness == Boundedness.BOUNDED) {
- assignSplitsForBounded(subtask, hostname);
+ if (!context.registeredReaders().containsKey(subtask)) {
+ LOG.warn("Ignoring split request from unregistered reader {}",
subtask);
+ return;
+ }
+ final Optional<JdbcSourceSplit> nextSplit = getNextSplit();
+ if (nextSplit.isPresent()) {
+ context.assignSplit(nextSplit.get(), subtask);
+ LOG.info("Assigned split to subtask {} : {}", subtask,
nextSplit.get());
+ preDiscoverSplits();
} else {
- readersAwaitingSplit.put(subtask, hostname);
- assignSplitsForUnbounded();
+ if (!readersWaitingForSplits.contains(subtask)) {
+ readersWaitingForSplits.add(subtask);
+ }
+ preDiscoverSplits();
}
}
@@ -115,18 +104,13 @@ public class JdbcSourceEnumerator
@Override
public void addSplitsBack(List<JdbcSourceSplit> splits, int subtaskId) {
- LOG.debug("File Source Enumerator adds splits back: {}", splits);
+ LOG.debug("Source Enumerator adds splits back: {}", splits);
unassigned.addAll(splits);
- if (boundedness == Boundedness.BOUNDED) {
- context.registeredReaders()
- .keySet()
- .forEach(subTask -> assignSplitsForBounded(subTask, null));
- } else if (boundedness == Boundedness.CONTINUOUS_UNBOUNDED) {
- if (context.registeredReaders().containsKey(subtaskId)) {
- readersAwaitingSplit.put(subtaskId, null);
- }
- assignSplitsForUnbounded();
+ if (context.registeredReaders().containsKey(subtaskId)
+ && !readersWaitingForSplits.contains(subtaskId)) {
+ readersWaitingForSplits.add(subtaskId);
}
+ preDiscoverSplits();
}
@Override
@@ -136,7 +120,7 @@ public class JdbcSourceEnumerator
Collections.emptyList(),
Collections.emptyList(),
new ArrayList<>(unassigned),
- sqlSplitEnumerator.optionalSqlSplitEnumeratorState);
+ splitterEnumerator.serializableState());
}
private Optional<JdbcSourceSplit> getNextSplit() {
@@ -149,58 +133,64 @@ public class JdbcSourceEnumerator
return Optional.of(next);
}
- private void assignSplitsForBounded(int subtask, @Nullable String
hostname) {
- if (!context.registeredReaders().containsKey(subtask)) {
- return;
- }
- if (LOG.isInfoEnabled()) {
- final String hostInfo =
- hostname == null ? "(no host locality info)" : "(on host
'" + hostname + "')";
- LOG.info("Subtask {} {} is requesting a Jdbc source split",
subtask, hostInfo);
- }
- final Optional<JdbcSourceSplit> nextSplit = getNextSplit();
- if (nextSplit.isPresent()) {
- final JdbcSourceSplit split = nextSplit.get();
- context.assignSplit(split, subtask);
- LOG.info("Assigned split to subtask {} : {}", subtask, split);
- } else {
- context.signalNoMoreSplits(subtask);
- LOG.info("No more splits available for subtask {}", subtask);
+ private void preDiscoverSplits() {
+ int targetParallelism = context.currentParallelism();
+ while (asyncCallsPending.get() < targetParallelism
+ && !splitterEnumerator.isAllSplitsFinished()) {
+ asyncCallsPending.incrementAndGet();
+ context.callAsync(() -> splitterEnumerator.enumerateSplits(),
this::onSplitsDiscovered);
}
+
+ signalNoMoreSplitsIfDone();
}
- private void processNewSplits(List<JdbcSourceSplit> splits, Throwable
error) {
+ private void onSplitsDiscovered(List<JdbcSourceSplit> splits, Throwable
error) {
+ asyncCallsPending.decrementAndGet();
if (error != null) {
- LOG.error("Failed to enumerate sql splits.", error);
+ LOG.error("Failed to discover splits.", error);
+ preDiscoverSplits();
return;
}
- this.unassigned.addAll(splits);
- assignSplitsForUnbounded();
+ if (splits != null && !splits.isEmpty()) {
+ assignOrBuffer(splits);
+ preDiscoverSplits();
+ } else if (!splitterEnumerator.isAllSplitsFinished()) {
+ preDiscoverSplits();
+ } else {
+ signalNoMoreSplitsIfDone();
+ }
}
- private void assignSplitsForUnbounded() {
- final Iterator<Map.Entry<Integer, String>> awaitingReader =
- readersAwaitingSplit.entrySet().iterator();
-
- while (awaitingReader.hasNext()) {
- final Map.Entry<Integer, String> nextAwaiting =
awaitingReader.next();
-
- // if the reader that requested another split has failed in the
meantime, remove
- // it from the list of waiting readers
- if
(!context.registeredReaders().containsKey(nextAwaiting.getKey())) {
- awaitingReader.remove();
- continue;
+ private void assignOrBuffer(List<JdbcSourceSplit> splits) {
+ for (JdbcSourceSplit split : splits) {
+ if (!readersWaitingForSplits.isEmpty()) {
+ int subtaskId = readersWaitingForSplits.remove(0);
+ if (context.registeredReaders().containsKey(subtaskId)) {
+ LOG.info(
+ "Assigning discovered split {} to waiting subtask
{}",
+ split,
+ subtaskId);
+ context.assignSplit(split, subtaskId);
+ } else {
+ unassigned.add(split);
+ }
+ } else {
+ unassigned.add(split);
}
+ }
+ }
- final int awaitingSubtask = nextAwaiting.getKey();
- final Optional<JdbcSourceSplit> nextSplit = getNextSplit();
- if (nextSplit.isPresent()) {
- context.assignSplit(nextSplit.get(), awaitingSubtask);
- awaitingReader.remove();
- } else {
- break;
+ private void signalNoMoreSplitsIfDone() {
+ if (asyncCallsPending.get() == 0
+ && splitterEnumerator.isAllSplitsFinished()
+ && unassigned.isEmpty()
+ && !readersWaitingForSplits.isEmpty()) {
+ for (int subtaskId : readersWaitingForSplits) {
+ context.signalNoMoreSplits(subtaskId);
+ LOG.info("No more splits available for subtask {}", subtaskId);
}
+ readersWaitingForSplits.clear();
}
}
}
diff --git
a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/JdbcSqlSplitEnumeratorBase.java
b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/JdbcSqlSplitEnumeratorBase.java
index dd3eb34b..c47b7cef 100644
---
a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/JdbcSqlSplitEnumeratorBase.java
+++
b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/JdbcSqlSplitEnumeratorBase.java
@@ -19,6 +19,7 @@
package org.apache.flink.connector.jdbc.core.datastream.source.enumerator;
import org.apache.flink.annotation.PublicEvolving;
+import
org.apache.flink.connector.jdbc.core.datastream.source.enumerator.splitter.SplitterEnumerator;
import
org.apache.flink.connector.jdbc.core.datastream.source.split.JdbcSourceSplit;
import javax.annotation.Nonnull;
@@ -33,7 +34,9 @@ import java.util.function.Supplier;
* Base class for jdbc sql split enumerator.
*
* @param <SplitT> JDBC split type.
+ * @deprecated Use {@link SplitterEnumerator} instead.
*/
+@Deprecated
@PublicEvolving
public abstract class JdbcSqlSplitEnumeratorBase<SplitT> implements
AutoCloseable, Serializable {
private final char[] currentId = "0000000000".toCharArray();
@@ -44,6 +47,10 @@ public abstract class JdbcSqlSplitEnumeratorBase<SplitT>
implements AutoCloseabl
this.optionalSqlSplitEnumeratorState = optionalSqlSplitEnumeratorState;
}
+ public Serializable enumeratorState() {
+ return this.optionalSqlSplitEnumeratorState;
+ }
+
protected final String getNextId() {
// because we just increment numbers, we increment the char
representation directly,
// rather than incrementing an integer and converting it to a string
representation
@@ -78,6 +85,7 @@ public abstract class JdbcSqlSplitEnumeratorBase<SplitT>
implements AutoCloseabl
*
* @param <SplitT> Split type.
*/
+ @Deprecated
@PublicEvolving
public interface Provider<SplitT> extends Serializable {
diff --git
a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/SqlTemplateSplitEnumerator.java
b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/SqlTemplateSplitEnumerator.java
index b4565d95..3a16b376 100644
---
a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/SqlTemplateSplitEnumerator.java
+++
b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/SqlTemplateSplitEnumerator.java
@@ -19,6 +19,7 @@
package org.apache.flink.connector.jdbc.core.datastream.source.enumerator;
import org.apache.flink.annotation.VisibleForTesting;
+import
org.apache.flink.connector.jdbc.core.datastream.source.enumerator.splitter.PreparedSplitterEnumerator;
import
org.apache.flink.connector.jdbc.core.datastream.source.split.CheckpointedOffset;
import
org.apache.flink.connector.jdbc.core.datastream.source.split.JdbcSourceSplit;
import org.apache.flink.connector.jdbc.split.JdbcParameterValuesProvider;
@@ -37,7 +38,12 @@ import java.util.Collections;
import java.util.List;
import java.util.function.Supplier;
-/** A split enumerator based on sql-parameters grains. */
+/**
+ * A split enumerator based on sql-parameters grains.
+ *
+ * @deprecated Use {@link PreparedSplitterEnumerator} instead.
+ */
+@Deprecated
public final class SqlTemplateSplitEnumerator extends
JdbcSqlSplitEnumeratorBase<JdbcSourceSplit> {
public static final Logger LOG =
LoggerFactory.getLogger(SqlTemplateSplitEnumerator.class);
@@ -105,6 +111,7 @@ public final class SqlTemplateSplitEnumerator extends
JdbcSqlSplitEnumeratorBase
}
/** The {@link TemplateSqlSplitEnumeratorProvider} for {@link
SqlTemplateSplitEnumerator}. */
+ @Deprecated
public static class TemplateSqlSplitEnumeratorProvider
implements JdbcSqlSplitEnumeratorBase.Provider<JdbcSourceSplit> {
diff --git
a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/splitter/JdbcSqlSplitterEnumerator.java
b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/splitter/JdbcSqlSplitterEnumerator.java
new file mode 100644
index 00000000..30bba69f
--- /dev/null
+++
b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/splitter/JdbcSqlSplitterEnumerator.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.flink.connector.jdbc.core.datastream.source.enumerator.splitter;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.connector.source.Boundedness;
+import
org.apache.flink.connector.jdbc.core.datastream.source.config.ContinuousUnBoundingSettings;
+import
org.apache.flink.connector.jdbc.core.datastream.source.enumerator.JdbcSqlSplitEnumeratorBase;
+import
org.apache.flink.connector.jdbc.core.datastream.source.enumerator.SqlTemplateSplitEnumerator;
+import
org.apache.flink.connector.jdbc.core.datastream.source.split.JdbcSourceSplit;
+import
org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider;
+
+import javax.annotation.Nullable;
+
+import java.io.Serializable;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+
+/**
+ * A splitter enumerator for JdbcSqlSplitEnumeratorBase.
+ *
+ * <p>This class is used to allow the retrofit of existing implementations of
+ * JdbcSqlSplitEnumeratorBase to the new SplitterEnumerator interface. It
delegates the split
+ * enumeration to the underlying JdbcSqlSplitEnumeratorBase instance.
+ */
+@Deprecated
+@Internal
+public class JdbcSqlSplitterEnumerator implements SplitterEnumerator {
+ private final JdbcSqlSplitEnumeratorBase.Provider<?> provider;
+ private final Boundedness boundedness;
+ private final @Nullable ContinuousUnBoundingSettings
continuousUnBoundingSettings;
+ private JdbcSqlSplitEnumeratorBase<?> base;
+ private volatile boolean finished;
+ private volatile boolean initialDelayApplied;
+
+ public JdbcSqlSplitterEnumerator(
+ JdbcSqlSplitEnumeratorBase.Provider<?> provider,
+ @Nullable ContinuousUnBoundingSettings
continuousUnBoundingSettings) {
+ this.provider = provider;
+ this.base = provider.create();
+ this.finished = false;
+ this.boundedness =
+ Objects.nonNull(continuousUnBoundingSettings)
+ ? Boundedness.CONTINUOUS_UNBOUNDED
+ : Boundedness.BOUNDED;
+ this.continuousUnBoundingSettings = continuousUnBoundingSettings;
+ }
+
+ @Override
+ public Boundedness getBoundedness() {
+ return boundedness;
+ }
+
+ @Override
+ public void start(JdbcConnectionProvider connectionProvider) {
+ base.open();
+ }
+
+ @Override
+ public void close() {
+ base.close();
+ }
+
+ @Override
+ public synchronized boolean isAllSplitsFinished() {
+ if (boundedness == Boundedness.CONTINUOUS_UNBOUNDED) {
+ return false;
+ }
+ return this.finished;
+ }
+
+ @Override
+ public synchronized List<JdbcSourceSplit> enumerateSplits() {
+ try {
+ if (finished) {
+ return new ArrayList<>();
+ }
+ if (boundedness == Boundedness.CONTINUOUS_UNBOUNDED &&
!initialDelayApplied) {
+ initialDelayApplied = true;
+ Duration delayMs =
continuousUnBoundingSettings.getInitialDiscoveryDelay();
+ if (delayMs != null) {
+ Thread.sleep(delayMs.toMillis());
+ }
+ }
+ List<JdbcSourceSplit> splits = base.enumerateSplits(() -> true);
+ if (boundedness == Boundedness.BOUNDED) {
+ this.finished = true;
+ }
+ if (boundedness == Boundedness.CONTINUOUS_UNBOUNDED &&
splits.isEmpty()) {
+ Duration throttleMs =
continuousUnBoundingSettings.getDiscoveryInterval();
+ if (throttleMs != null) {
+ Thread.sleep(throttleMs.toMillis());
+ }
+ }
+ return splits;
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return new ArrayList<>();
+ } catch (Exception e) {
+ throw new RuntimeException("Error enumerating splits", e);
+ }
+ }
+
+ @Override
+ public List<String> lineageQueries() {
+ List<String> queries = new ArrayList<>();
+ if (base instanceof SqlTemplateSplitEnumerator) {
+ queries.add(((SqlTemplateSplitEnumerator) base).getSqlTemplate());
+ }
+ return queries;
+ }
+
+ @Override
+ public Serializable serializableState() {
+ return base.enumeratorState();
+ }
+
+ @Override
+ public SplitterEnumerator restoreState(Serializable state) {
+ this.base = provider.restore(state);
+ return this;
+ }
+}
diff --git
a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/splitter/PreparedSplitterEnumerator.java
b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/splitter/PreparedSplitterEnumerator.java
new file mode 100644
index 00000000..e0340381
--- /dev/null
+++
b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/splitter/PreparedSplitterEnumerator.java
@@ -0,0 +1,135 @@
+/*
+ * 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.flink.connector.jdbc.core.datastream.source.enumerator.splitter;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.connector.source.Boundedness;
+import
org.apache.flink.connector.jdbc.core.datastream.source.split.JdbcSourceSplit;
+import
org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider;
+import org.apache.flink.util.Preconditions;
+
+import javax.annotation.Nullable;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Objects;
+
+/** A split enumerator based on sql-parameters grains. */
+@PublicEvolving
+public class PreparedSplitterEnumerator extends SqlSplitterEnumerator {
+
+ private final Serializable[][] sqlParameters;
+ private volatile boolean finished;
+
+ protected PreparedSplitterEnumerator(String sqlTemplate, Serializable[][]
sqlParameters) {
+ super(Preconditions.checkNotNull(sqlTemplate));
+ this.sqlParameters = Preconditions.checkNotNull(sqlParameters);
+ this.finished = false;
+ }
+
+ public static PreparedSplitterEnumerator of(
+ String sqlTemplate, Serializable[][] sqlParameters) {
+ return new PreparedSplitterEnumerator(sqlTemplate, sqlParameters);
+ }
+
+ public static PreparedSplitterEnumerator of(String sqlTemplate) {
+ return new PreparedSplitterEnumerator(sqlTemplate, new
Serializable[0][]);
+ }
+
+ public static PreparedSplitterEnumerator of(
+ String sqlTemplate, long minValue, long maxValue, long batchSize) {
+ PreparedSplitterNumericParameters parameters =
+ new PreparedSplitterNumericParameters(minValue,
maxValue).withBatchSize(batchSize);
+ return of(sqlTemplate, parameters);
+ }
+
+ public static PreparedSplitterEnumerator of(
+ String sqlTemplate, long minValue, long maxValue, int batchNum) {
+ PreparedSplitterNumericParameters parameters =
+ new PreparedSplitterNumericParameters(minValue,
maxValue).withBatchNum(batchNum);
+ return of(sqlTemplate, parameters);
+ }
+
+ public static PreparedSplitterEnumerator of(
+ String sqlTemplate, PreparedSplitterNumericParameters
numericParameters) {
+ Serializable[][] parameters = numericParameters.getParameterValues();
+ return new PreparedSplitterEnumerator(sqlTemplate, parameters);
+ }
+
+ @Override
+ public Boundedness getBoundedness() {
+ return Boundedness.BOUNDED;
+ }
+
+ @Override
+ public void start(JdbcConnectionProvider connectionProvider) {}
+
+ @Override
+ public void close() {}
+
+ @Override
+ public synchronized boolean isAllSplitsFinished() {
+ return this.finished;
+ }
+
+ @VisibleForTesting
+ public Serializable[][] getSqlParameters() {
+ return sqlParameters;
+ }
+
+ @Override
+ public synchronized List<JdbcSourceSplit> enumerateSplits() {
+ if (finished) {
+ return new ArrayList<>();
+ }
+ List<JdbcSourceSplit> splitList = super.enumerateSplits();
+ this.finished = true;
+ return splitList;
+ }
+
+ @Override
+ public @Nullable Serializable serializableState() {
+ return null;
+ }
+
+ @Override
+ public PreparedSplitterEnumerator restoreState(Serializable state) {
+ return this;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (!(o instanceof PreparedSplitterEnumerator)) {
+ return false;
+ }
+ if (!super.equals(o)) {
+ return false;
+ }
+ PreparedSplitterEnumerator that = (PreparedSplitterEnumerator) o;
+ return Objects.deepEquals(sqlParameters, that.sqlParameters);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(super.hashCode(),
Arrays.deepHashCode(sqlParameters));
+ }
+}
diff --git
a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/split/JdbcNumericBetweenParametersProvider.java
b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/splitter/PreparedSplitterNumericParameters.java
similarity index 51%
copy from
flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/split/JdbcNumericBetweenParametersProvider.java
copy to
flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/splitter/PreparedSplitterNumericParameters.java
index 617e6c52..c48763c3 100644
---
a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/split/JdbcNumericBetweenParametersProvider.java
+++
b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/splitter/PreparedSplitterNumericParameters.java
@@ -16,62 +16,34 @@
* limitations under the License.
*/
-package org.apache.flink.connector.jdbc.split;
+package
org.apache.flink.connector.jdbc.core.datastream.source.enumerator.splitter;
+import org.apache.flink.annotation.Internal;
import org.apache.flink.util.Preconditions;
import java.io.Serializable;
/**
- * This query parameters generator is an helper class to parameterize from/to
queries on a numeric
- * column. The generated array of from/to values will be equally sized to
fetchSize (apart from the
+ * This query parameters generator is a helper class to parameterize from/to
queries on a numeric
+ * column. The generated array of from/to values will be equally sized to
batchSize (apart from the
* last one), ranging from minVal up to maxVal.
- *
- * <p>For example, if there's a table <CODE>BOOKS</CODE> with a numeric PK
<CODE>id</CODE>, using a
- * query like:
- *
- * <PRE>
- * SELECT * FROM BOOKS WHERE id BETWEEN ? AND ?
- * </PRE>
- *
- * <p>You can take advantage of this class to automatically generate the
parameters of the BETWEEN
- * clause, based on the passed constructor parameters.
*/
-public class JdbcNumericBetweenParametersProvider implements
JdbcParameterValuesProvider {
+@Internal
+public class PreparedSplitterNumericParameters implements Serializable {
private final long minVal;
private final long maxVal;
-
private long batchSize;
private int batchNum;
- /**
- * NumericBetweenParametersProviderJdbc constructor.
- *
- * @param minVal the lower bound of the produced "from" values
- * @param maxVal the upper bound of the produced "to" values
- */
- public JdbcNumericBetweenParametersProvider(long minVal, long maxVal) {
- Preconditions.checkArgument(minVal <= maxVal, "minVal must not be
larger than maxVal");
- this.minVal = minVal;
- this.maxVal = maxVal;
- }
-
- /**
- * NumericBetweenParametersProviderJdbc constructor.
- *
- * @param fetchSize the max distance between the produced from/to pairs
- * @param minVal the lower bound of the produced "from" values
- * @param maxVal the upper bound of the produced "to" values
- */
- public JdbcNumericBetweenParametersProvider(long fetchSize, long minVal,
long maxVal) {
- Preconditions.checkArgument(minVal <= maxVal, "minVal must not be
larger than maxVal");
+ public PreparedSplitterNumericParameters(long minVal, long maxVal) {
this.minVal = minVal;
this.maxVal = maxVal;
- ofBatchSize(fetchSize);
+ this.batchNum = 0;
+ this.batchSize = 0;
}
- public JdbcNumericBetweenParametersProvider ofBatchSize(long batchSize) {
+ public PreparedSplitterNumericParameters withBatchSize(long batchSize) {
Preconditions.checkArgument(batchSize > 0, "Batch size must be
positive");
long maxElemCount = (maxVal - minVal) + 1;
@@ -83,7 +55,7 @@ public class JdbcNumericBetweenParametersProvider implements
JdbcParameterValues
return this;
}
- public JdbcNumericBetweenParametersProvider ofBatchNum(int batchNum) {
+ public PreparedSplitterNumericParameters withBatchNum(int batchNum) {
Preconditions.checkArgument(batchNum > 0, "Batch number must be
positive");
long maxElemCount = (maxVal - minVal) + 1;
@@ -91,15 +63,14 @@ public class JdbcNumericBetweenParametersProvider
implements JdbcParameterValues
batchNum = (int) maxElemCount;
}
this.batchNum = batchNum;
- this.batchSize = new Double(Math.ceil((double) maxElemCount /
batchNum)).longValue();
+ this.batchSize = (maxElemCount + batchNum - 1) / batchNum;
return this;
}
- @Override
public Serializable[][] getParameterValues() {
Preconditions.checkState(
batchSize > 0,
- "Batch size and batch number must be positive. Have you called
`ofBatchSize` or `ofBatchNum`?");
+ "Batch size and batch number must be positive. Have you called
`withBatchSize` or `withBatchNum`?");
long maxElemCount = (maxVal - minVal) + 1;
long bigBatchNum = maxElemCount - (batchSize - 1) * batchNum;
@@ -113,19 +84,4 @@ public class JdbcNumericBetweenParametersProvider
implements JdbcParameterValues
}
return parameters;
}
-
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (o == null || getClass() != o.getClass()) {
- return false;
- }
- JdbcNumericBetweenParametersProvider that =
(JdbcNumericBetweenParametersProvider) o;
- return minVal == that.minVal
- && maxVal == that.maxVal
- && batchSize == that.batchSize
- && batchNum == that.batchNum;
- }
}
diff --git
a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/splitter/SlideTimingSplitterEnumerator.java
b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/splitter/SlideTimingSplitterEnumerator.java
new file mode 100644
index 00000000..3c1d9a42
--- /dev/null
+++
b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/splitter/SlideTimingSplitterEnumerator.java
@@ -0,0 +1,187 @@
+/*
+ * 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.flink.connector.jdbc.core.datastream.source.enumerator.splitter;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.connector.source.Boundedness;
+import
org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider;
+import org.apache.flink.util.Preconditions;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+
+/** A split enumerator based on sql-parameters grains. */
+@PublicEvolving
+public class SlideTimingSplitterEnumerator extends SqlSplitterEnumerator {
+
+ private final long slideStepMillis;
+ private final long slideSpanMillis;
+ private final long splitGenerateDelayMillis;
+
+ private @Nonnull Long startMillis;
+
+ protected SlideTimingSplitterEnumerator(
+ String sqlTemplate,
+ Long startMillis,
+ long slideSpanMillis,
+ long slideStepMillis,
+ long splitGenerateDelayMillis) {
+ super(sqlTemplate);
+ this.startMillis = Preconditions.checkNotNull(startMillis);
+ this.slideStepMillis = slideStepMillis;
+ this.slideSpanMillis = slideSpanMillis;
+ this.splitGenerateDelayMillis = splitGenerateDelayMillis;
+ }
+
+ @Override
+ public Boundedness getBoundedness() {
+ return Boundedness.CONTINUOUS_UNBOUNDED;
+ }
+
+ @Override
+ public void start(JdbcConnectionProvider connectionProvider) {}
+
+ @Override
+ public void close() {}
+
+ @Override
+ public boolean isAllSplitsFinished() {
+ return false;
+ }
+
+ @VisibleForTesting
+ public Serializable[][] getSqlParameters() {
+ List<Serializable[]> tmpList = new ArrayList<>();
+ while (nextSplitAvailable(startMillis)) {
+ Serializable[] params = new Serializable[] {startMillis,
startMillis + slideSpanMillis};
+ tmpList.add(params);
+ startMillis += slideStepMillis;
+ }
+ return tmpList.toArray(new Serializable[0][]);
+ }
+
+ @Override
+ public @Nullable Serializable serializableState() {
+ return this.startMillis;
+ }
+
+ @Override
+ public SlideTimingSplitterEnumerator restoreState(Serializable state) {
+ Preconditions.checkArgument((Long) state > 0L);
+ this.startMillis = (Long) state;
+ return this;
+ }
+
+ private boolean nextSplitAvailable(Long nextSpanStartMillis) {
+ final long delayedNextSpanStartMillis = nextSpanStartMillis +
splitGenerateDelayMillis;
+ final long currentAvailableMillis = currentAvailableMillis();
+ return currentAvailableMillis >= delayedNextSpanStartMillis
+ && (currentAvailableMillis - delayedNextSpanStartMillis >=
slideSpanMillis);
+ }
+
+ private Long currentAvailableMillis() {
+ return System.currentTimeMillis();
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (!(o instanceof SlideTimingSplitterEnumerator)) {
+ return false;
+ }
+ if (!super.equals(o)) {
+ return false;
+ }
+ SlideTimingSplitterEnumerator that = (SlideTimingSplitterEnumerator) o;
+ return slideStepMillis == that.slideStepMillis
+ && slideSpanMillis == that.slideSpanMillis
+ && splitGenerateDelayMillis == that.splitGenerateDelayMillis
+ && Objects.equals(startMillis, that.startMillis);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(
+ super.hashCode(),
+ slideStepMillis,
+ slideSpanMillis,
+ splitGenerateDelayMillis,
+ startMillis);
+ }
+
+ public static SlideTimingSplitterEnumeratorBuilder builder() {
+ return new SlideTimingSplitterEnumeratorBuilder();
+ }
+
+ /** A slide split enumerator builder. */
+ public static class SlideTimingSplitterEnumeratorBuilder {
+ private String sqlTemplate;
+ private Long startMillis;
+ private Long slideSpanMillis;
+ private Long slideStepMillis;
+ private Long splitGenerateDelayMillis;
+
+ public SlideTimingSplitterEnumeratorBuilder setSqlTemplate(String
sqlTemplate) {
+ this.sqlTemplate = sqlTemplate;
+ return this;
+ }
+
+ public SlideTimingSplitterEnumeratorBuilder setStartMillis(Long
startMillis) {
+ this.startMillis = startMillis;
+ return this;
+ }
+
+ public SlideTimingSplitterEnumeratorBuilder setSlideSpanMillis(Long
slideSpanMillis) {
+ this.slideSpanMillis = slideSpanMillis;
+ return this;
+ }
+
+ public SlideTimingSplitterEnumeratorBuilder setSlideStepMillis(Long
slideStepMillis) {
+ this.slideStepMillis = slideStepMillis;
+ return this;
+ }
+
+ public SlideTimingSplitterEnumeratorBuilder
setSplitGenerateDelayMillis(
+ Long splitGenerateDelayMillis) {
+ this.splitGenerateDelayMillis = splitGenerateDelayMillis;
+ return this;
+ }
+
+ public SlideTimingSplitterEnumerator build() {
+ Preconditions.checkArgument(startMillis > 0L, "'startMillis' must
be greater than 0. ");
+ Preconditions.checkArgument(
+ slideSpanMillis > 0 && slideStepMillis > 0,
+ "parameters must satisfy slideSpanMillis > 0 and
slideStepMillis > 0");
+ Preconditions.checkArgument(
+ splitGenerateDelayMillis >= 0L,
+ "parameters must satisfy splitGenerateDelayMillis >= 0");
+ return new SlideTimingSplitterEnumerator(
+ sqlTemplate,
+ startMillis,
+ slideSpanMillis,
+ slideStepMillis,
+ splitGenerateDelayMillis);
+ }
+ }
+}
diff --git
a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/splitter/SplitterEnumerator.java
b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/splitter/SplitterEnumerator.java
new file mode 100644
index 00000000..96f4947d
--- /dev/null
+++
b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/splitter/SplitterEnumerator.java
@@ -0,0 +1,73 @@
+/*
+ * 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.flink.connector.jdbc.core.datastream.source.enumerator.splitter;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.connector.source.Boundedness;
+import
org.apache.flink.connector.jdbc.core.datastream.source.split.JdbcSourceSplit;
+import
org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider;
+
+import java.io.Serializable;
+import java.util.List;
+
+/** Interface for jdbc sql split enumerator. */
+@PublicEvolving
+public interface SplitterEnumerator extends AutoCloseable, Serializable {
+
+ /** Returns whether this enumerator produces a bounded or continuous
unbounded stream. */
+ Boundedness getBoundedness();
+
+ /**
+ * Start the enumerator.
+ *
+ * @param connectionProvider The JDBC connection provider.
+ */
+ void start(JdbcConnectionProvider connectionProvider);
+
+ /** Close the enumerator. */
+ void close();
+
+ /** All splits have been enumerated. */
+ boolean isAllSplitsFinished();
+
+ /** Enumerate the JDBC splits. */
+ List<JdbcSourceSplit> enumerateSplits();
+
+ /**
+ * Get lineage queries for splits.
+ *
+ * @return The lineage queries.
+ */
+ List<String> lineageQueries();
+
+ /**
+ * Get the serializable state of the enumerator.
+ *
+ * @return The serializable state.
+ */
+ Serializable serializableState();
+
+ /**
+ * Restore the enumerator state from the given serializable state.
+ *
+ * @param state The serializable state.
+ * @return The restored SplitterEnumerator.
+ */
+ SplitterEnumerator restoreState(Serializable state);
+}
diff --git
a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/splitter/SqlSplitterEnumerator.java
b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/splitter/SqlSplitterEnumerator.java
new file mode 100644
index 00000000..ef9c0db1
--- /dev/null
+++
b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/splitter/SqlSplitterEnumerator.java
@@ -0,0 +1,106 @@
+/*
+ * 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.flink.connector.jdbc.core.datastream.source.enumerator.splitter;
+
+import org.apache.flink.annotation.VisibleForTesting;
+import
org.apache.flink.connector.jdbc.core.datastream.source.split.CheckpointedOffset;
+import
org.apache.flink.connector.jdbc.core.datastream.source.split.JdbcSourceSplit;
+import org.apache.flink.util.Preconditions;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+
+/** A sql base template split enumerator. */
+public abstract class SqlSplitterEnumerator implements SplitterEnumerator {
+
+ private final char[] currentId = "0000000000".toCharArray();
+
+ private final String sqlTemplate;
+
+ protected SqlSplitterEnumerator(String sqlTemplate) {
+ this.sqlTemplate = Preconditions.checkNotNull(sqlTemplate);
+ }
+
+ @VisibleForTesting
+ protected abstract Serializable[][] getSqlParameters();
+
+ protected String getSqlTemplate() {
+ return this.sqlTemplate;
+ }
+
+ @Override
+ public List<JdbcSourceSplit> enumerateSplits() {
+ Serializable[][] params = getSqlParameters();
+ if (params == null || params.length == 0) {
+ return Collections.emptyList();
+ }
+ int paramLength = params.length;
+ List<JdbcSourceSplit> splitList = new ArrayList<>(paramLength);
+ for (Serializable[] paramArr : params) {
+ splitList.add(createSplit(paramArr));
+ }
+ return splitList;
+ }
+
+ @Override
+ public List<String> lineageQueries() {
+ return Collections.singletonList(getSqlTemplate());
+ }
+
+ protected JdbcSourceSplit createSplit(Serializable[] paramArr) {
+ return new JdbcSourceSplit(
+ getNextId(), getSqlTemplate(), paramArr, new
CheckpointedOffset());
+ }
+
+ protected final String getNextId() {
+ // because we just increment numbers, we increment the char
representation directly,
+ // rather than incrementing an integer and converting it to a string
representation
+ // every time again (requires quite some expensive conversion logic).
+ incrementCharArrayByOne(currentId, currentId.length - 1);
+ return new String(currentId);
+ }
+
+ private static void incrementCharArrayByOne(char[] array, int pos) {
+ char c = array[pos];
+ c++;
+
+ if (c > '9') {
+ c = '0';
+ incrementCharArrayByOne(array, pos - 1);
+ }
+ array[pos] = c;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (!(o instanceof SqlSplitterEnumerator)) {
+ return false;
+ }
+ SqlSplitterEnumerator that = (SqlSplitterEnumerator) o;
+ return Objects.equals(sqlTemplate, that.sqlTemplate);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(sqlTemplate);
+ }
+}
diff --git
a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/lineage/LineageUtils.java
b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/lineage/LineageUtils.java
index 0a09671c..be5ea909 100644
---
a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/lineage/LineageUtils.java
+++
b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/lineage/LineageUtils.java
@@ -32,8 +32,8 @@ import io.openlineage.sql.DbTableMeta;
import io.openlineage.sql.OpenLineageSql;
import java.util.ArrayList;
-import java.util.Arrays;
import java.util.Collection;
+import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -56,7 +56,11 @@ public class LineageUtils {
}
public static Optional<String> tableNameOf(String query, boolean isSource)
{
- return OpenLineageSql.parse(Arrays.asList(query))
+ return tableNameOf(Collections.singletonList(query), isSource);
+ }
+
+ public static Optional<String> tableNameOf(List<String> queries, boolean
isSource) {
+ return OpenLineageSql.parse(queries)
.map(
sqlMeta ->
isSource
diff --git
a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/split/CompositeJdbcParameterValuesProvider.java
b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/split/CompositeJdbcParameterValuesProvider.java
index eebeac5a..ded972ef 100644
---
a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/split/CompositeJdbcParameterValuesProvider.java
+++
b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/split/CompositeJdbcParameterValuesProvider.java
@@ -24,6 +24,7 @@ import org.apache.flink.util.Preconditions;
import java.io.Serializable;
/** Combine 2 {@link JdbcParameterValuesProvider} into 1. */
+@Deprecated
@Internal
public class CompositeJdbcParameterValuesProvider implements
JdbcParameterValuesProvider {
JdbcParameterValuesProvider a;
diff --git
a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/split/JdbcGenericParameterValuesProvider.java
b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/split/JdbcGenericParameterValuesProvider.java
index d9637982..2d831231 100644
---
a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/split/JdbcGenericParameterValuesProvider.java
+++
b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/split/JdbcGenericParameterValuesProvider.java
@@ -20,13 +20,18 @@ package org.apache.flink.connector.jdbc.split;
import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.connector.jdbc.JdbcInputFormat;
+import
org.apache.flink.connector.jdbc.core.datastream.source.enumerator.splitter.PreparedSplitterEnumerator;
import java.io.Serializable;
/**
* This splits generator actually does nothing but wrapping the query
parameters computed by the
* user before creating the {@link JdbcInputFormat} instance.
+ *
+ * @deprecated Use {@link PreparedSplitterEnumerator} instead. As example:
+ * PreparedSplitterEnumerator.of(query, parameters).
*/
+@Deprecated
@PublicEvolving
public class JdbcGenericParameterValuesProvider implements
JdbcParameterValuesProvider {
diff --git
a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/split/JdbcNumericBetweenParametersProvider.java
b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/split/JdbcNumericBetweenParametersProvider.java
index 617e6c52..c06a9b82 100644
---
a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/split/JdbcNumericBetweenParametersProvider.java
+++
b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/split/JdbcNumericBetweenParametersProvider.java
@@ -18,6 +18,7 @@
package org.apache.flink.connector.jdbc.split;
+import
org.apache.flink.connector.jdbc.core.datastream.source.enumerator.splitter.PreparedSplitterEnumerator;
import org.apache.flink.util.Preconditions;
import java.io.Serializable;
@@ -36,7 +37,12 @@ import java.io.Serializable;
*
* <p>You can take advantage of this class to automatically generate the
parameters of the BETWEEN
* clause, based on the passed constructor parameters.
+ *
+ * @deprecated Use {@link PreparedSplitterEnumerator} instead. As example:
+ * PreparedSplitterEnumerator.of(query, minValue, maxValue, batchSize) or
+ * PreparedSplitterEnumerator.of(query, minValue, maxValue, batchNum).
*/
+@Deprecated
public class JdbcNumericBetweenParametersProvider implements
JdbcParameterValuesProvider {
private final long minVal;
diff --git
a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/split/JdbcParameterValuesProvider.java
b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/split/JdbcParameterValuesProvider.java
index 3f1c2aed..bb95829e 100644
---
a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/split/JdbcParameterValuesProvider.java
+++
b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/split/JdbcParameterValuesProvider.java
@@ -20,6 +20,7 @@ package org.apache.flink.connector.jdbc.split;
import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.connector.jdbc.JdbcInputFormat;
+import
org.apache.flink.connector.jdbc.core.datastream.source.enumerator.splitter.SplitterEnumerator;
import java.io.Serializable;
@@ -27,7 +28,10 @@ import java.io.Serializable;
* This interface is used by the {@link JdbcInputFormat} to compute the list
of parallel query to
* run (i.e. splits). Each query will be parameterized using a row of the
matrix provided by each
* {@link JdbcParameterValuesProvider} implementation.
+ *
+ * @deprecated Use {@link SplitterEnumerator} instead.
*/
+@Deprecated
@PublicEvolving
public interface JdbcParameterValuesProvider extends Serializable {
diff --git
a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/split/JdbcSlideTimingParameterProvider.java
b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/split/JdbcSlideTimingParameterProvider.java
index eae5891e..e94dadbd 100644
---
a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/split/JdbcSlideTimingParameterProvider.java
+++
b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/split/JdbcSlideTimingParameterProvider.java
@@ -19,6 +19,7 @@
package org.apache.flink.connector.jdbc.split;
import org.apache.flink.annotation.PublicEvolving;
+import
org.apache.flink.connector.jdbc.core.datastream.source.enumerator.splitter.SlideTimingSplitterEnumerator;
import org.apache.flink.util.Preconditions;
import javax.annotation.Nonnull;
@@ -27,7 +28,15 @@ import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;
-/** The parameters provider generate parameters by slide timing window
strategy. */
+/**
+ * The parameters provider generate parameters by slide timing window strategy.
+ *
+ * @deprecated Use {@link SlideTimingSplitterEnumerator} instead. As example:
+ * SlideTimingSplitterEnumerator.builder() .setSqlTemplate(query)
.setStartMills(startMills)
+ * .setSlideSpanMills(slideSpanMills) .setSlideStepMills(slideStepMills)
+ * .setSplitGenerateDelayMillis(splitGenerateDelayMillis) .build();
+ */
+@Deprecated
@PublicEvolving
public class JdbcSlideTimingParameterProvider implements
JdbcParameterValuesProvider {
diff --git
a/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/datastream/source/JdbcSourceBuilderTest.java
b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/datastream/source/JdbcSourceBuilderTest.java
index d116433b..488ae807 100644
---
a/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/datastream/source/JdbcSourceBuilderTest.java
+++
b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/datastream/source/JdbcSourceBuilderTest.java
@@ -22,8 +22,10 @@ import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import
org.apache.flink.connector.jdbc.core.datastream.source.config.ContinuousUnBoundingSettings;
-import
org.apache.flink.connector.jdbc.core.datastream.source.enumerator.SqlTemplateSplitEnumerator;
+import
org.apache.flink.connector.jdbc.core.datastream.source.enumerator.splitter.JdbcSqlSplitterEnumerator;
import
org.apache.flink.connector.jdbc.core.datastream.source.reader.extractor.ResultExtractor;
+import
org.apache.flink.connector.jdbc.core.datastream.source.split.CheckpointedOffset;
+import
org.apache.flink.connector.jdbc.core.datastream.source.split.JdbcSourceSplit;
import
org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider;
import
org.apache.flink.connector.jdbc.datasource.connections.SimpleJdbcConnectionProvider;
import
org.apache.flink.connector.jdbc.split.JdbcGenericParameterValuesProvider;
@@ -53,7 +55,7 @@ class JdbcSourceBuilderTest {
private final String dbUrl = "dbUrl";
private final ResultExtractor<Row> extractor =
ResultExtractor.ofRowResultExtractor();
private final JdbcParameterValuesProvider parameterValuesProvider =
- new JdbcNumericBetweenParametersProvider(0, 3);
+ new JdbcNumericBetweenParametersProvider(0, 3).ofBatchSize(1);
private final TypeInformation<Row> typeInformation = new TypeHint<Row>()
{}.getTypeInfo();
@@ -74,13 +76,15 @@ class JdbcSourceBuilderTest {
.isInstanceOf(IllegalArgumentException.class);
assertThatThrownBy(() -> JdbcSource.builder().setSql(emptySql))
.isInstanceOf(IllegalArgumentException.class);
- assertThatThrownBy(() -> JdbcSource.builder().setDBUrl(dbUrl).build())
- .isInstanceOf(IllegalStateException.class);
+
// For valid.
JdbcSource<Row> jdbcSource = sourceBuilder.build();
- SqlTemplateSplitEnumerator sqlEnumerator =
- (SqlTemplateSplitEnumerator)
jdbcSource.getSqlSplitEnumeratorProvider().create();
- assertThat(sqlEnumerator.getSqlTemplate()).isEqualTo(validSql);
+ assertThat(jdbcSource.getSplitterEnumerator())
+ .isInstanceOf(JdbcSqlSplitterEnumerator.class);
+ JdbcSqlSplitterEnumerator sqlEnumerator =
+ (JdbcSqlSplitterEnumerator) jdbcSource.getSplitterEnumerator();
+
+ assertThat(sqlEnumerator.lineageQueries()).containsExactly(validSql);
}
@Test
@@ -89,10 +93,32 @@ class JdbcSourceBuilderTest {
.isInstanceOf(NullPointerException.class);
JdbcSource<Row> jdbcSource =
sourceBuilder.setJdbcParameterValuesProvider(parameterValuesProvider).build();
- SqlTemplateSplitEnumerator sqlSplitEnumerator =
- (SqlTemplateSplitEnumerator)
jdbcSource.getSqlSplitEnumeratorProvider().create();
- assertThat(sqlSplitEnumerator.getParameterValuesProvider())
- .isEqualTo(parameterValuesProvider);
+ assertThat(jdbcSource.getSplitterEnumerator())
+ .isInstanceOf(JdbcSqlSplitterEnumerator.class);
+ JdbcSqlSplitterEnumerator sqlEnumerator =
+ (JdbcSqlSplitterEnumerator) jdbcSource.getSplitterEnumerator();
+ assertThat(sqlEnumerator.enumerateSplits())
+ .containsExactly(
+ new JdbcSourceSplit(
+ "0000000001",
+ validSql,
+ new Serializable[] {0L, 0L},
+ new CheckpointedOffset()),
+ new JdbcSourceSplit(
+ "0000000002",
+ validSql,
+ new Serializable[] {1L, 1L},
+ new CheckpointedOffset()),
+ new JdbcSourceSplit(
+ "0000000003",
+ validSql,
+ new Serializable[] {2L, 2L},
+ new CheckpointedOffset()),
+ new JdbcSourceSplit(
+ "0000000004",
+ validSql,
+ new Serializable[] {3L, 3L},
+ new CheckpointedOffset()));
}
@Test
diff --git
a/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/datastream/source/JdbcSourceStreamRelatedITCase.java
b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/datastream/source/JdbcSourceStreamRelatedITCase.java
index 23814eed..c45c9e64 100644
---
a/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/datastream/source/JdbcSourceStreamRelatedITCase.java
+++
b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/datastream/source/JdbcSourceStreamRelatedITCase.java
@@ -149,6 +149,7 @@ class JdbcSourceStreamRelatedITCase implements
DerbyTestBase, JdbcITCaseBase {
.addSink(new TestingSinkFunction());
waitExpectation(client, env, () -> collectedRecords.size() >=
TESTING_ENTRIES_SIZE);
+ assertThat(collectedRecords).hasSize(testEntries.size());
assertThat(collectedRecords).containsExactlyInAnyOrderElementsOf(testEntries);
}
diff --git
a/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/JdbcSourceEnumeratorTest.java
b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/JdbcSourceEnumeratorTest.java
index 268fb73f..8b1c87e1 100644
---
a/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/JdbcSourceEnumeratorTest.java
+++
b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/JdbcSourceEnumeratorTest.java
@@ -18,22 +18,21 @@
package org.apache.flink.connector.jdbc.core.datastream.source.enumerator;
+import org.apache.flink.api.connector.source.Boundedness;
import org.apache.flink.api.connector.source.SplitEnumeratorContext;
+import
org.apache.flink.connector.jdbc.core.datastream.source.enumerator.splitter.SplitterEnumerator;
import
org.apache.flink.connector.jdbc.core.datastream.source.split.CheckpointedOffset;
import
org.apache.flink.connector.jdbc.core.datastream.source.split.JdbcSourceSplit;
+import
org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider;
import
org.apache.flink.connector.testutils.source.reader.TestingSplitEnumeratorContext;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
-import javax.annotation.Nonnull;
-
-import java.io.IOException;
import java.io.Serializable;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
-import java.util.function.Supplier;
import java.util.stream.Collectors;
import static org.assertj.core.api.Assertions.assertThat;
@@ -108,12 +107,42 @@ class JdbcSourceEnumeratorTest {
return new JdbcSourceEnumerator(
context,
- new JdbcSqlSplitEnumeratorBase<JdbcSourceSplit>(null) {
+ new SplitterEnumerator() {
+ @Override
+ public Boundedness getBoundedness() {
+ return Boundedness.BOUNDED;
+ }
+
+ @Override
+ public void start(JdbcConnectionProvider
connectionProvider) {}
+
+ @Override
+ public void close() {}
+
+ @Override
+ public boolean isAllSplitsFinished() {
+ return true;
+ }
+
@Override
- public @Nonnull List<JdbcSourceSplit> enumerateSplits(
- @Nonnull Supplier<Boolean> splitGettable) throws
IOException {
+ public List<JdbcSourceSplit> enumerateSplits() {
return Collections.emptyList();
}
+
+ @Override
+ public List<String> lineageQueries() {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public Serializable serializableState() {
+ return null;
+ }
+
+ @Override
+ public SplitterEnumerator restoreState(Serializable state)
{
+ return null;
+ }
},
null,
Arrays.stream(splits).collect(Collectors.toList()));
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
new file mode 100644
index 00000000..0b21d40c
--- /dev/null
+++
b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/splitter/PreparedSplitterNumericParametersTest.java
@@ -0,0 +1,134 @@
+/*
+ * 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.flink.connector.jdbc.core.datastream.source.enumerator.splitter;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.Serializable;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class PreparedSplitterNumericParametersTest {
+
+ @Test
+ void testBatchSizeDivisible() {
+ Serializable[][] parameters =
+ new PreparedSplitterNumericParameters(-5,
9).withBatchSize(3).getParameterValues();
+
+ long[][] expected = {
+ new long[] {-5, -3},
+ new long[] {-2, 0},
+ new long[] {1, 3},
+ new long[] {4, 6},
+ new long[] {7, 9}
+ };
+ check(expected, parameters);
+ }
+
+ @Test
+ void testBatchSizeNotDivisible() {
+ Serializable[][] parameters =
+ new PreparedSplitterNumericParameters(-5,
11).withBatchSize(4).getParameterValues();
+
+ long[][] expected = {
+ new long[] {-5, -2},
+ new long[] {-1, 2},
+ new long[] {3, 5},
+ new long[] {6, 8},
+ new long[] {9, 11}
+ };
+ check(expected, parameters);
+ }
+
+ @Test
+ void testBatchSizeTooLarge() {
+ Serializable[][] parameters =
+ new PreparedSplitterNumericParameters(0,
2).withBatchSize(5).getParameterValues();
+
+ long[][] expected = {new long[] {0, 2}};
+ check(expected, parameters);
+ }
+
+ @Test
+ void testBatchNumDivisible() {
+ Serializable[][] parameters =
+ new PreparedSplitterNumericParameters(-5,
9).withBatchNum(5).getParameterValues();
+
+ long[][] expected = {
+ new long[] {-5, -3},
+ new long[] {-2, 0},
+ new long[] {1, 3},
+ new long[] {4, 6},
+ new long[] {7, 9}
+ };
+ check(expected, parameters);
+ }
+
+ @Test
+ void testBatchNumNotDivisible() {
+ Serializable[][] parameters =
+ new PreparedSplitterNumericParameters(-5,
11).withBatchNum(5).getParameterValues();
+
+ long[][] expected = {
+ new long[] {-5, -2},
+ new long[] {-1, 2},
+ new long[] {3, 5},
+ new long[] {6, 8},
+ new long[] {9, 11}
+ };
+ check(expected, parameters);
+ }
+
+ @Test
+ void testBatchNumTooLarge() {
+ Serializable[][] parameters =
+ new PreparedSplitterNumericParameters(0,
2).withBatchNum(5).getParameterValues();
+
+ long[][] expected = {
+ new long[] {0, 0},
+ new long[] {1, 1},
+ new long[] {2, 2}
+ };
+ check(expected, parameters);
+ }
+
+ @Test
+ void testBatchMaxMinTooLarge() {
+ Serializable[][] parameters =
+ new PreparedSplitterNumericParameters(2260418954055131340L,
3875220057236942850L)
+ .withBatchNum(3)
+ .getParameterValues();
+
+ long[][] expected = {
+ new long[] {2260418954055131340L, 2798685988449068510L},
+ new long[] {2798685988449068511L, 3336953022843005680L},
+ new long[] {3336953022843005681L, 3875220057236942850L}
+ };
+ check(expected, parameters);
+ }
+
+ private void check(long[][] expected, Serializable[][] actual) {
+ assertThat(actual).hasDimensions(expected.length, expected[0].length);
+ for (int i = 0; i < expected.length; i++) {
+ for (int j = 0; j < expected[i].length; j++) {
+ assertThat(((Long)
actual[i][j]).longValue()).isEqualTo(expected[i][j]);
+ }
+ }
+ }
+}
diff --git
a/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/splitter/SlideTimingSplitterEnumeratorTest.java
b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/splitter/SlideTimingSplitterEnumeratorTest.java
new file mode 100644
index 00000000..87e8e82c
--- /dev/null
+++
b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/splitter/SlideTimingSplitterEnumeratorTest.java
@@ -0,0 +1,142 @@
+/*
+ * 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.flink.connector.jdbc.core.datastream.source.enumerator.splitter;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.Serializable;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+class SlideTimingSplitterEnumeratorTest {
+
+ @Test
+ void testBuilderWithValidParameters() {
+ SlideTimingSplitterEnumerator enumerator = defaultEnumerator();
+
+ assertThat(enumerator).isNotNull();
+ }
+
+ @Test
+ void testBuilderWithInvalidStartMills() {
+ assertThatThrownBy(() -> createEnumerator(0L, 500L, 100L))
+ .isInstanceOf(IllegalArgumentException.class);
+ }
+
+ @Test
+ void testBuilderWithInvalidSlideSpanMillsAndStepMills() {
+ assertThatThrownBy(() -> createEnumerator(1000L, 0L, 0L))
+ .isInstanceOf(IllegalArgumentException.class);
+ }
+
+ @Test
+ void testBuilderWithNegativeSplitGenerateDelayMillis() {
+ assertThatThrownBy(() -> createEnumerator(1000L, 500L, 100L, -1L))
+ .isInstanceOf(IllegalArgumentException.class);
+ }
+
+ @Test
+ void testEnumerateSplitsWithNoParameters() {
+ SlideTimingSplitterEnumerator enumerator =
+ new SlideTimingSplitterEnumerator(
+ "SELECT * FROM table WHERE time >= ? AND time < ?",
1000L, 500L, 100L, 0L) {
+ @Override
+ public Serializable[][] getSqlParameters() {
+ return new Serializable[0][0];
+ }
+ };
+ assertThat(enumerator.enumerateSplits()).isEmpty();
+ }
+
+ @Test
+ void testEquals() {
+ SlideTimingSplitterEnumerator enumerator1 = defaultEnumerator();
+
+ SlideTimingSplitterEnumerator enumerator2 = defaultEnumerator();
+
+ assertThat(enumerator1).isEqualTo(enumerator2);
+ }
+
+ @Test
+ void testHashCode() {
+ SlideTimingSplitterEnumerator enumerator1 = defaultEnumerator();
+
+ SlideTimingSplitterEnumerator enumerator2 = defaultEnumerator();
+
+ assertThat(enumerator1.hashCode()).isEqualTo(enumerator2.hashCode());
+ }
+
+ @Test
+ void testStart() {
+ SlideTimingSplitterEnumerator enumerator = defaultEnumerator();
+
+ assertThatCode(() ->
enumerator.start(null)).doesNotThrowAnyException();
+ }
+
+ @Test
+ void testClose() {
+ SlideTimingSplitterEnumerator enumerator = defaultEnumerator();
+
+ assertThatCode(enumerator::close).doesNotThrowAnyException();
+ }
+
+ @Test
+ void testIsAllSplitsFinished() {
+ SlideTimingSplitterEnumerator enumerator = defaultEnumerator();
+
+ assertThat(enumerator.isAllSplitsFinished()).isFalse();
+ }
+
+ @Test
+ void testRestoreState() {
+ SlideTimingSplitterEnumerator enumerator = defaultEnumerator();
+
+ SlideTimingSplitterEnumerator restored =
enumerator.restoreState(2000L);
+ assertThat(restored).isNotNull();
+ }
+
+ @Test
+ void testSerializableState() {
+ SlideTimingSplitterEnumerator enumerator = defaultEnumerator();
+
+ assertThat(enumerator.serializableState()).isNotNull();
+ }
+
+ SlideTimingSplitterEnumerator defaultEnumerator() {
+ return createEnumerator(1000L, 500L, 100L);
+ }
+
+ SlideTimingSplitterEnumerator createEnumerator(
+ Long startMillis, Long slideSpanMillis, Long slideStepMillis) {
+ return createEnumerator(startMillis, slideSpanMillis, slideStepMillis,
0L);
+ }
+
+ SlideTimingSplitterEnumerator createEnumerator(
+ Long startMillis, Long slideSpanMillis, Long slideStepMillis, Long
splitDelayMillis) {
+ return SlideTimingSplitterEnumerator.builder()
+ .setSqlTemplate("SELECT * FROM table WHERE time >= ? AND time
< ?")
+ .setStartMillis(startMillis)
+ .setSlideSpanMillis(slideSpanMillis)
+ .setSlideStepMillis(slideStepMillis)
+ .setSplitGenerateDelayMillis(splitDelayMillis)
+ .build();
+ }
+}
