You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by kk...@apache.org on 2020/03/10 16:17:04 UTC
[flink] 06/10: [FLINK-15782][connectors/jdbc] rename Batch to
Execution Options
This is an automated email from the ASF dual-hosted git repository.
kkloudas pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 3d2e47d6fffed53b4bd8f9dec8a4ca9cf2f86182
Author: Roman Khachatryan <kh...@gmail.com>
AuthorDate: Wed Jan 29 19:25:47 2020 +0100
[FLINK-15782][connectors/jdbc] rename Batch to Execution Options
---
.../flink/api/java/io/jdbc/JDBCLookupOptions.java | 2 +-
.../flink/api/java/io/jdbc/JDBCOutputFormat.java | 13 +++---
.../api/java/io/jdbc/JDBCUpsertTableSink.java | 2 +-
.../api/java/io/jdbc/JdbcBatchingOutputFormat.java | 52 ++++++++++++----------
.../flink/api/java/io/jdbc/JdbcDmlOptions.java | 38 +++++-----------
...BatchOptions.java => JdbcExecutionOptions.java} | 45 ++++++++++++-------
.../java/io/jdbc/TableJdbcUpsertOutputFormat.java | 12 ++---
.../jdbc/executor/JdbcBatchStatementExecutor.java | 10 ++---
.../java/io/jdbc/JdbcTableOutputFormatTest.java | 4 +-
9 files changed, 88 insertions(+), 90 deletions(-)
diff --git a/flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/JDBCLookupOptions.java b/flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/JDBCLookupOptions.java
index 62d9d92..1f55cd6 100644
--- a/flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/JDBCLookupOptions.java
+++ b/flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/JDBCLookupOptions.java
@@ -21,7 +21,7 @@ package org.apache.flink.api.java.io.jdbc;
import java.io.Serializable;
import java.util.Objects;
-import static org.apache.flink.api.java.io.jdbc.JdbcBatchingOutputFormat.DEFAULT_MAX_RETRY_TIMES;
+import static org.apache.flink.api.java.io.jdbc.JdbcExecutionOptions.DEFAULT_MAX_RETRY_TIMES;
/**
* Options for the JDBC lookup.
diff --git a/flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/JDBCOutputFormat.java b/flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/JDBCOutputFormat.java
index 0a1107f..9bd094b 100644
--- a/flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/JDBCOutputFormat.java
+++ b/flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/JDBCOutputFormat.java
@@ -24,7 +24,6 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
-import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
@@ -48,22 +47,22 @@ public class JDBCOutputFormat extends AbstractJdbcOutputFormat<Row> {
private static final Logger LOG = LoggerFactory.getLogger(JDBCOutputFormat.class);
final JdbcInsertOptions insertOptions;
- private final JdbcBatchOptions batchOptions;
+ private final JdbcExecutionOptions batchOptions;
private transient PreparedStatement upload;
private transient int batchCount = 0;
/**
- * @deprecated use {@link #JDBCOutputFormat(JdbcConnectionOptions, JdbcInsertOptions, JdbcBatchOptions)}}.
+ * @deprecated use {@link #JDBCOutputFormat(JdbcConnectionOptions, JdbcInsertOptions, JdbcExecutionOptions)}}.
*/
@Deprecated
public JDBCOutputFormat(String username, String password, String drivername, String dbURL, String query, int batchInterval, int[] typesArray) {
this(new JdbcConnectionOptions(dbURL, drivername, username, password),
new JdbcInsertOptions(query, typesArray),
- JdbcBatchOptions.builder().withSize(batchInterval).build());
+ JdbcExecutionOptions.builder().withBatchSize(batchInterval).build());
}
- public JDBCOutputFormat(JdbcConnectionOptions connectionOptions, JdbcInsertOptions insertOptions, JdbcBatchOptions batchOptions) {
+ public JDBCOutputFormat(JdbcConnectionOptions connectionOptions, JdbcInsertOptions insertOptions, JdbcExecutionOptions batchOptions) {
super(connectionOptions);
this.insertOptions = insertOptions;
this.batchOptions = batchOptions;
@@ -97,7 +96,7 @@ public class JDBCOutputFormat extends AbstractJdbcOutputFormat<Row> {
batchCount++;
- if (batchCount >= batchOptions.getSize()) {
+ if (batchCount >= batchOptions.getBatchSize()) {
// execute batch
flush();
}
@@ -202,7 +201,7 @@ public class JDBCOutputFormat extends AbstractJdbcOutputFormat<Row> {
public JDBCOutputFormat finish() {
return new JDBCOutputFormat(buildConnectionOptions(),
new JdbcInsertOptions(query, typesArray),
- JdbcBatchOptions.builder().withSize(batchInterval).build());
+ JdbcExecutionOptions.builder().withBatchSize(batchInterval).build());
}
public JdbcConnectionOptions buildConnectionOptions() {
diff --git a/flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/JDBCUpsertTableSink.java b/flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/JDBCUpsertTableSink.java
index af4076a..4acc35d 100644
--- a/flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/JDBCUpsertTableSink.java
+++ b/flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/JDBCUpsertTableSink.java
@@ -39,7 +39,7 @@ import java.util.Objects;
import static org.apache.flink.api.java.io.jdbc.AbstractJdbcOutputFormat.DEFAULT_FLUSH_INTERVAL_MILLS;
import static org.apache.flink.api.java.io.jdbc.AbstractJdbcOutputFormat.DEFAULT_FLUSH_MAX_SIZE;
-import static org.apache.flink.api.java.io.jdbc.JdbcBatchingOutputFormat.DEFAULT_MAX_RETRY_TIMES;
+import static org.apache.flink.api.java.io.jdbc.JdbcExecutionOptions.DEFAULT_MAX_RETRY_TIMES;
import static org.apache.flink.api.java.io.jdbc.JdbcTypeUtil.normalizeTableSchema;
import static org.apache.flink.util.Preconditions.checkNotNull;
diff --git a/flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/JdbcBatchingOutputFormat.java b/flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/JdbcBatchingOutputFormat.java
index 2d28c7a..df94334 100644
--- a/flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/JdbcBatchingOutputFormat.java
+++ b/flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/JdbcBatchingOutputFormat.java
@@ -29,6 +29,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
+import java.io.Serializable;
import java.sql.SQLException;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
@@ -39,15 +40,22 @@ import java.util.function.Function;
import static org.apache.flink.util.Preconditions.checkNotNull;
class JdbcBatchingOutputFormat<In, JdbcIn, JdbcExec extends JdbcBatchStatementExecutor<JdbcIn>> extends AbstractJdbcOutputFormat<In> {
+ interface RecordExtractor<F, T> extends Function<F, T>, Serializable {
+ static <T> RecordExtractor<T, T> identity() {
+ return x -> x;
+ }
+ }
+
+ interface ExecutorCreator<T extends JdbcBatchStatementExecutor<?>> extends Function<RuntimeContext, T>, Serializable {
+ }
private static final long serialVersionUID = 1L;
private static final Logger LOG = LoggerFactory.getLogger(JdbcBatchingOutputFormat.class);
- static final int DEFAULT_MAX_RETRY_TIMES = 3;
-
- final JdbcDmlOptions dmlOptions;
- private final JdbcBatchOptions batchOptions;
+ private final JdbcExecutionOptions executionOptions;
+ private final ExecutorCreator<JdbcExec> statementRunnerCreator;
+ final RecordExtractor<In, JdbcIn> jdbcRecordExtractor;
private transient JdbcExec jdbcStatementExecutor;
private transient int batchCount = 0;
@@ -56,17 +64,14 @@ class JdbcBatchingOutputFormat<In, JdbcIn, JdbcExec extends JdbcBatchStatementEx
private transient ScheduledExecutorService scheduler;
private transient ScheduledFuture<?> scheduledFuture;
private transient volatile Exception flushException;
- private final Function<RuntimeContext, JdbcExec> statementRunnerCreator;
- final Function<In, JdbcIn> jdbcRecordExtractor;
-
- JdbcBatchingOutputFormat(JdbcConnectionOptions connectionOptions,
- JdbcDmlOptions dmlOptions,
- JdbcBatchOptions batchOptions,
- Function<RuntimeContext, JdbcExec> statementExecutorCreator,
- Function<In, JdbcIn> recordExtractor) {
+
+ JdbcBatchingOutputFormat(
+ JdbcConnectionOptions connectionOptions,
+ JdbcExecutionOptions executionOptions,
+ ExecutorCreator<JdbcExec> statementExecutorCreator,
+ RecordExtractor<In, JdbcIn> recordExtractor) {
super(connectionOptions);
- this.dmlOptions = dmlOptions;
- this.batchOptions = batchOptions;
+ this.executionOptions = executionOptions;
this.statementRunnerCreator = statementExecutorCreator;
this.jdbcRecordExtractor = recordExtractor;
}
@@ -85,7 +90,7 @@ class JdbcBatchingOutputFormat<In, JdbcIn, JdbcExec extends JdbcBatchStatementEx
} catch (SQLException e) {
throw new IOException("unable to open JDBC writer", e);
}
- if (batchOptions.getIntervalMs() != 0 && batchOptions.getSize() != 1) {
+ if (executionOptions.getBatchIntervalMs() != 0 && executionOptions.getBatchSize() != 1) {
this.scheduler = Executors.newScheduledThreadPool(1, new ExecutorThreadFactory("jdbc-upsert-output-format"));
this.scheduledFuture = this.scheduler.scheduleWithFixedDelay(() -> {
synchronized (JdbcBatchingOutputFormat.this) {
@@ -97,7 +102,7 @@ class JdbcBatchingOutputFormat<In, JdbcIn, JdbcExec extends JdbcBatchStatementEx
}
}
}
- }, batchOptions.getIntervalMs(), batchOptions.getIntervalMs(), TimeUnit.MILLISECONDS);
+ }, executionOptions.getBatchIntervalMs(), executionOptions.getBatchIntervalMs(), TimeUnit.MILLISECONDS);
}
}
@@ -114,7 +119,7 @@ class JdbcBatchingOutputFormat<In, JdbcIn, JdbcExec extends JdbcBatchStatementEx
try {
doWriteRecord(record);
batchCount++;
- if (batchCount >= batchOptions.getSize()) {
+ if (batchCount >= executionOptions.getBatchSize()) {
flush();
}
} catch (Exception e) {
@@ -130,14 +135,14 @@ class JdbcBatchingOutputFormat<In, JdbcIn, JdbcExec extends JdbcBatchStatementEx
public synchronized void flush() throws IOException {
checkFlushException();
- for (int i = 1; i <= dmlOptions.getMaxRetries(); i++) {
+ for (int i = 1; i <= executionOptions.getMaxRetries(); i++) {
try {
attemptFlush();
batchCount = 0;
break;
} catch (SQLException e) {
LOG.error("JDBC executeBatch error, retry times = {}", i, e);
- if (i >= dmlOptions.getMaxRetries()) {
+ if (i >= executionOptions.getMaxRetries()) {
throw new IOException(e);
}
try {
@@ -201,7 +206,7 @@ class JdbcBatchingOutputFormat<In, JdbcIn, JdbcExec extends JdbcBatchStatementEx
private int[] fieldTypes;
private int flushMaxSize = DEFAULT_FLUSH_MAX_SIZE;
private long flushIntervalMills = DEFAULT_FLUSH_INTERVAL_MILLS;
- private int maxRetryTimes = DEFAULT_MAX_RETRY_TIMES;
+ private int maxRetryTimes = JdbcExecutionOptions.DEFAULT_MAX_RETRY_TIMES;
/**
* required, jdbc options.
@@ -270,14 +275,13 @@ class JdbcBatchingOutputFormat<In, JdbcIn, JdbcExec extends JdbcBatchStatementEx
checkNotNull(fieldNames, "No fieldNames supplied.");
JdbcDmlOptions dml = JdbcDmlOptions.builder()
.withTableName(options.getTableName()).withDialect(options.getDialect())
- .withFieldNames(fieldNames).withKeyFields(keyFields).withFieldTypes(fieldTypes).withMaxRetries(maxRetryTimes).build();
+ .withFieldNames(fieldNames).withKeyFields(keyFields).withFieldTypes(fieldTypes).build();
if (dml.getKeyFields() == null || dml.getKeyFields().length == 0) {
// warn: don't close over builder fields
String sql = options.getDialect().getInsertIntoStatement(dml.getTableName(), dml.getFieldNames());
return new JdbcBatchingOutputFormat<>(
options,
- dml,
- JdbcBatchOptions.builder().withSize(flushMaxSize).withIntervalMs(flushIntervalMills).build(),
+ JdbcExecutionOptions.builder().withBatchSize(flushMaxSize).withMaxRetries(maxRetryTimes).withBatchIntervalMs(flushIntervalMills).build(),
unused -> JdbcBatchStatementExecutor.simpleRow(sql, dml.getFieldTypes()),
tuple2 -> {
Preconditions.checkArgument(tuple2.f0);
@@ -287,7 +291,7 @@ class JdbcBatchingOutputFormat<In, JdbcIn, JdbcExec extends JdbcBatchStatementEx
return new TableJdbcUpsertOutputFormat(
options,
dml,
- JdbcBatchOptions.builder().withSize(flushMaxSize).withIntervalMs(flushIntervalMills).build());
+ JdbcExecutionOptions.builder().withBatchSize(flushMaxSize).withBatchIntervalMs(flushIntervalMills).build());
}
}
}
diff --git a/flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/JdbcDmlOptions.java b/flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/JdbcDmlOptions.java
index 361d4f4..5af701f 100644
--- a/flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/JdbcDmlOptions.java
+++ b/flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/JdbcDmlOptions.java
@@ -31,30 +31,22 @@ public class JdbcDmlOptions extends JdbcTypedQueryOptions {
private static final long serialVersionUID = 1L;
- private static final int DEFAULT_MAX_RETRY_TIMES = 3;
-
private final String[] fieldNames;
@Nullable
private final String[] keyFields;
private final String tableName;
private final JDBCDialect dialect;
- private final int maxRetries;
- public static JDBCUpsertOptionsBuilder builder() {
- return new JDBCUpsertOptionsBuilder();
+ public static JdbcDmlOptionsBuilder builder() {
+ return new JdbcDmlOptionsBuilder();
}
- private JdbcDmlOptions(String tableName, JDBCDialect dialect, String[] fieldNames, int[] fieldTypes, String[] keyFields, int maxRetries) {
+ private JdbcDmlOptions(String tableName, JDBCDialect dialect, String[] fieldNames, int[] fieldTypes, String[] keyFields) {
super(fieldTypes);
this.tableName = Preconditions.checkNotNull(tableName, "table is empty");
this.dialect = Preconditions.checkNotNull(dialect, "dialect name is empty");
this.fieldNames = Preconditions.checkNotNull(fieldNames, "field names is empty");
this.keyFields = keyFields;
- this.maxRetries = maxRetries;
- }
-
- public int getMaxRetries() {
- return maxRetries;
}
public String getTableName() {
@@ -77,56 +69,50 @@ public class JdbcDmlOptions extends JdbcTypedQueryOptions {
/**
* JDBCUpsertOptionsBuilder.
*/
- public static class JDBCUpsertOptionsBuilder extends JDBCUpdateQueryOptionsBuilder<JDBCUpsertOptionsBuilder> {
+ public static class JdbcDmlOptionsBuilder extends JDBCUpdateQueryOptionsBuilder<JdbcDmlOptionsBuilder> {
private String tableName;
private String[] fieldNames;
private String[] keyFields;
private JDBCDialect dialect;
private JDBCDialect customDialect;
- private int maxRetries = DEFAULT_MAX_RETRY_TIMES;
-
- public JDBCUpsertOptionsBuilder withMaxRetries(int maxRetries) {
- this.maxRetries = maxRetries;
- return this;
- }
@Override
- protected JDBCUpsertOptionsBuilder self() {
+ protected JdbcDmlOptionsBuilder self() {
return this;
}
- public JDBCUpsertOptionsBuilder withFieldNames(String field, String... fieldNames) {
+ public JdbcDmlOptionsBuilder withFieldNames(String field, String... fieldNames) {
this.fieldNames = concat(field, fieldNames);
return this;
}
- public JDBCUpsertOptionsBuilder withFieldNames(String[] fieldNames) {
+ public JdbcDmlOptionsBuilder withFieldNames(String[] fieldNames) {
this.fieldNames = fieldNames;
return this;
}
- public JDBCUpsertOptionsBuilder withKeyFields(String keyField, String... keyFields) {
+ public JdbcDmlOptionsBuilder withKeyFields(String keyField, String... keyFields) {
this.keyFields = concat(keyField, keyFields);
return this;
}
- public JDBCUpsertOptionsBuilder withKeyFields(String[] keyFields) {
+ public JdbcDmlOptionsBuilder withKeyFields(String[] keyFields) {
this.keyFields = keyFields;
return this;
}
- public JDBCUpsertOptionsBuilder withTableName(String tableName) {
+ public JdbcDmlOptionsBuilder withTableName(String tableName) {
this.tableName = tableName;
return self();
}
- public JDBCUpsertOptionsBuilder withDialect(JDBCDialect dialect) {
+ public JdbcDmlOptionsBuilder withDialect(JDBCDialect dialect) {
this.dialect = dialect;
return self();
}
public JdbcDmlOptions build() {
- return new JdbcDmlOptions(tableName, dialect, fieldNames, fieldTypes, keyFields, maxRetries);
+ return new JdbcDmlOptions(tableName, dialect, fieldNames, fieldTypes, keyFields);
}
static String[] concat(String first, String... next) {
diff --git a/flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/JdbcBatchOptions.java b/flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/JdbcExecutionOptions.java
similarity index 57%
rename from flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/JdbcBatchOptions.java
rename to flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/JdbcExecutionOptions.java
index 69b14de..af22d6c 100644
--- a/flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/JdbcBatchOptions.java
+++ b/flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/JdbcExecutionOptions.java
@@ -18,6 +18,7 @@
package org.apache.flink.api.java.io.jdbc;
import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.util.Preconditions;
import java.io.Serializable;
@@ -25,31 +26,39 @@ import java.io.Serializable;
* JDBC sink batch options.
*/
@PublicEvolving
-public class JdbcBatchOptions implements Serializable {
+public class JdbcExecutionOptions implements Serializable {
+ static final int DEFAULT_MAX_RETRY_TIMES = 3;
private static final int DEFAULT_INTERVAL_MILLIS = 0;
private static final int DEFAULT_SIZE = 5000;
- private final long intervalMs;
- private final int size;
+ private final long batchIntervalMs;
+ private final int batchSize;
+ private final int maxRetries;
- private JdbcBatchOptions(long intervalMs, int size) {
- this.intervalMs = intervalMs;
- this.size = size;
+ private JdbcExecutionOptions(long batchIntervalMs, int batchSize, int maxRetries) {
+ Preconditions.checkArgument(maxRetries >= 1);
+ this.batchIntervalMs = batchIntervalMs;
+ this.batchSize = batchSize;
+ this.maxRetries = maxRetries;
}
- public long getIntervalMs() {
- return intervalMs;
+ public long getBatchIntervalMs() {
+ return batchIntervalMs;
}
- public int getSize() {
- return size;
+ public int getBatchSize() {
+ return batchSize;
+ }
+
+ public int getMaxRetries() {
+ return maxRetries;
}
public static JdbcBatchOptionsBuilder builder() {
return new JdbcBatchOptionsBuilder();
}
- public static JdbcBatchOptions defaults() {
+ public static JdbcExecutionOptions defaults() {
return builder().build();
}
@@ -59,19 +68,25 @@ public class JdbcBatchOptions implements Serializable {
public static final class JdbcBatchOptionsBuilder {
private long intervalMs = DEFAULT_INTERVAL_MILLIS;
private int size = DEFAULT_SIZE;
+ private int maxRetries = DEFAULT_MAX_RETRY_TIMES;
- public JdbcBatchOptionsBuilder withSize(int size) {
+ public JdbcBatchOptionsBuilder withBatchSize(int size) {
this.size = size;
return this;
}
- public JdbcBatchOptionsBuilder withIntervalMs(long intervalMs) {
+ public JdbcBatchOptionsBuilder withBatchIntervalMs(long intervalMs) {
this.intervalMs = intervalMs;
return this;
}
- public JdbcBatchOptions build() {
- return new JdbcBatchOptions(intervalMs, size);
+ public JdbcBatchOptionsBuilder withMaxRetries(int maxRetries) {
+ this.maxRetries = maxRetries;
+ return this;
+ }
+
+ public JdbcExecutionOptions build() {
+ return new JdbcExecutionOptions(intervalMs, size, maxRetries);
}
}
}
diff --git a/flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/TableJdbcUpsertOutputFormat.java b/flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/TableJdbcUpsertOutputFormat.java
index 9935349..1a31d54 100644
--- a/flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/TableJdbcUpsertOutputFormat.java
+++ b/flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/TableJdbcUpsertOutputFormat.java
@@ -27,20 +27,16 @@ import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.sql.SQLException;
import java.util.Arrays;
-import java.util.function.Function;
class TableJdbcUpsertOutputFormat extends JdbcBatchingOutputFormat<Tuple2<Boolean, Row>, Row, JdbcBatchStatementExecutor<Row>> {
private static final Logger LOG = LoggerFactory.getLogger(TableJdbcUpsertOutputFormat.class);
private JdbcBatchStatementExecutor<Row> deleteExecutor;
+ private final JdbcDmlOptions dmlOptions;
- TableJdbcUpsertOutputFormat(JdbcConnectionOptions connectionOptions, JdbcDmlOptions dmlOptions, JdbcBatchOptions batchOptions) {
- super(connectionOptions, dmlOptions, batchOptions, ctx -> JdbcBatchStatementExecutor.upsertRow(dmlOptions, ctx), new Function<Tuple2<Boolean, Row>, Row>() {
- @Override
- public Row apply(Tuple2<Boolean, Row> record) {
- return record.f1;
- }
- });
+ TableJdbcUpsertOutputFormat(JdbcConnectionOptions connectionOptions, JdbcDmlOptions dmlOptions, JdbcExecutionOptions batchOptions) {
+ super(connectionOptions, batchOptions, ctx -> JdbcBatchStatementExecutor.upsertRow(dmlOptions, ctx), tuple2 -> tuple2.f1);
+ this.dmlOptions = dmlOptions;
}
@Override
diff --git a/flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/executor/JdbcBatchStatementExecutor.java b/flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/executor/JdbcBatchStatementExecutor.java
index 3ca04f5..7e436d0 100644
--- a/flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/executor/JdbcBatchStatementExecutor.java
+++ b/flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/executor/JdbcBatchStatementExecutor.java
@@ -21,7 +21,6 @@ package org.apache.flink.api.java.io.jdbc.executor;
import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.java.io.jdbc.JdbcDmlOptions;
-import org.apache.flink.api.java.io.jdbc.dialect.JDBCDialect;
import org.apache.flink.types.Row;
import java.sql.Connection;
@@ -58,19 +57,18 @@ public interface JdbcBatchStatementExecutor<T> {
static JdbcBatchStatementExecutor<Row> upsertRow(JdbcDmlOptions opt, RuntimeContext ctx) {
checkNotNull(opt.getKeyFields());
- JDBCDialect dialect = opt.getDialectName().getInstance(opt);
int[] pkFields = Arrays.stream(opt.getKeyFields()).mapToInt(Arrays.asList(opt.getFieldNames())::indexOf).toArray();
int[] pkTypes = opt.getFieldTypes() == null ? null : Arrays.stream(pkFields).map(f -> opt.getFieldTypes()[f]).toArray();
- return dialect
+ return opt.getDialect()
.getUpsertStatement(opt.getTableName(), opt.getFieldNames(), opt.getKeyFields())
.map(sql -> keyedRow(pkFields, opt.getFieldTypes(), sql))
.orElseGet(() ->
new InsertOrUpdateJdbcExecutor<>(
- dialect.getRowExistsStatement(opt.getTableName(), opt.getKeyFields()),
- dialect.getInsertIntoStatement(opt.getTableName(), opt.getFieldNames()),
- dialect.getUpdateStatement(opt.getTableName(), opt.getFieldNames(), opt.getKeyFields()),
+ opt.getDialect().getRowExistsStatement(opt.getTableName(), opt.getKeyFields()),
+ opt.getDialect().getInsertIntoStatement(opt.getTableName(), opt.getFieldNames()),
+ opt.getDialect().getUpdateStatement(opt.getTableName(), opt.getFieldNames(), opt.getKeyFields()),
ParameterSetter.forRow(pkTypes),
ParameterSetter.forRow(opt.getFieldTypes()),
ParameterSetter.forRow(opt.getFieldTypes()),
diff --git a/flink-connectors/flink-jdbc/src/test/java/org/apache/flink/api/java/io/jdbc/JdbcTableOutputFormatTest.java b/flink-connectors/flink-jdbc/src/test/java/org/apache/flink/api/java/io/jdbc/JdbcTableOutputFormatTest.java
index 5a105a9..0d54558 100644
--- a/flink-connectors/flink-jdbc/src/test/java/org/apache/flink/api/java/io/jdbc/JdbcTableOutputFormatTest.java
+++ b/flink-connectors/flink-jdbc/src/test/java/org/apache/flink/api/java/io/jdbc/JdbcTableOutputFormatTest.java
@@ -64,9 +64,9 @@ public class JdbcTableOutputFormatTest extends JDBCTestBase {
.setTableName(OUTPUT_TABLE)
.build();
JdbcDmlOptions dmlOptions = JdbcDmlOptions.builder()
- .withTableName(options.getTableName()).withDialect(options.getDialect().getName())
+ .withTableName(options.getTableName()).withDialect(options.getDialect())
.withFieldNames(fieldNames).withKeyFields(keyFields).build();
- format = new TableJdbcUpsertOutputFormat(options, dmlOptions, JdbcBatchOptions.defaults());
+ format = new TableJdbcUpsertOutputFormat(options, dmlOptions, JdbcExecutionOptions.defaults());
RuntimeContext context = Mockito.mock(RuntimeContext.class);
ExecutionConfig config = Mockito.mock(ExecutionConfig.class);
doReturn(config).when(context).getExecutionConfig();