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:03 UTC
[flink] 05/10: [FLINK-15782][connectors/jdbc] generalize output
formats and sinks (from Row to ) Motivation: use existing code oriented
to Table API in DataStream API
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 62065c82ffa909f9b5794ed3e2877da279088b68
Author: Roman Khachatryan <kh...@gmail.com>
AuthorDate: Fri Jan 24 15:45:15 2020 +0100
[FLINK-15782][connectors/jdbc] generalize output formats and sinks (from Row to <T>)
Motivation: use existing code oriented to Table API in DataStream API
---
.../api/java/io/jdbc/AbstractJdbcOutputFormat.java | 7 +-
...kFunction.java => GenericJdbcSinkFunction.java} | 16 ++--
.../flink/api/java/io/jdbc/JDBCOutputFormat.java | 7 +-
.../api/java/io/jdbc/JDBCUpsertTableSink.java | 5 +-
.../api/java/io/jdbc/JdbcBatchingOutputFormat.java | 91 +++++++++++-----------
.../flink/api/java/io/jdbc/JdbcSinkFunction.java | 4 +
.../api/java/io/jdbc/JdbcTypedQueryOptions.java | 2 +-
...ormat.java => TableJdbcUpsertOutputFormat.java} | 29 +++----
.../jdbc/executor/JdbcBatchStatementExecutor.java | 34 ++++----
.../java/io/jdbc/JdbcTableOutputFormatTest.java | 4 +-
10 files changed, 103 insertions(+), 96 deletions(-)
diff --git a/flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/AbstractJdbcOutputFormat.java b/flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/AbstractJdbcOutputFormat.java
index 6d806bb..7bd812f 100644
--- a/flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/AbstractJdbcOutputFormat.java
+++ b/flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/AbstractJdbcOutputFormat.java
@@ -24,12 +24,13 @@ import org.apache.flink.configuration.Configuration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.io.Flushable;
import java.io.IOException;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
-abstract class AbstractJdbcOutputFormat<T> extends RichOutputFormat<T> {
+abstract class AbstractJdbcOutputFormat<T> extends RichOutputFormat<T> implements Flushable {
private static final long serialVersionUID = 1L;
static final int DEFAULT_FLUSH_MAX_SIZE = 5000;
@@ -83,4 +84,8 @@ abstract class AbstractJdbcOutputFormat<T> extends RichOutputFormat<T> {
}
}
}
+
+ @Override
+ public void flush() throws IOException {
+ }
}
diff --git a/flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/JdbcUpsertSinkFunction.java b/flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/GenericJdbcSinkFunction.java
similarity index 80%
rename from flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/JdbcUpsertSinkFunction.java
rename to flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/GenericJdbcSinkFunction.java
index 7b6aac7..0fe21b4 100644
--- a/flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/JdbcUpsertSinkFunction.java
+++ b/flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/GenericJdbcSinkFunction.java
@@ -19,18 +19,18 @@
package org.apache.flink.api.java.io.jdbc;
import org.apache.flink.api.common.functions.RuntimeContext;
-import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
-import org.apache.flink.types.Row;
-class JdbcUpsertSinkFunction extends RichSinkFunction<Tuple2<Boolean, Row>> implements CheckpointedFunction {
- private final JdbcBatchingOutputFormat outputFormat;
+import java.io.IOException;
- JdbcUpsertSinkFunction(JdbcBatchingOutputFormat outputFormat) {
+class GenericJdbcSinkFunction<T> extends RichSinkFunction<T> implements CheckpointedFunction {
+ private final AbstractJdbcOutputFormat<T > outputFormat;
+
+ GenericJdbcSinkFunction(AbstractJdbcOutputFormat<T > outputFormat) {
this.outputFormat = outputFormat;
}
@@ -43,12 +43,12 @@ class JdbcUpsertSinkFunction extends RichSinkFunction<Tuple2<Boolean, Row>> impl
}
@Override
- public void invoke(Tuple2<Boolean, Row> value, Context context) throws Exception {
+ public void invoke(T value, Context context) throws IOException {
outputFormat.writeRecord(value);
}
@Override
- public void initializeState(FunctionInitializationContext context) throws Exception {
+ public void initializeState(FunctionInitializationContext context) {
}
@Override
@@ -57,7 +57,7 @@ class JdbcUpsertSinkFunction extends RichSinkFunction<Tuple2<Boolean, Row>> impl
}
@Override
- public void close() throws Exception {
+ public void close() {
outputFormat.close();
}
}
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 217dbf5..0a1107f 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
@@ -37,6 +37,10 @@ import static org.apache.flink.api.java.io.jdbc.JDBCUtils.setRecordToStatement;
* @see Row
* @see DriverManager
*/
+/**
+ * @deprecated use {@link JdbcBatchingOutputFormat}
+ */
+@Deprecated
public class JDBCOutputFormat extends AbstractJdbcOutputFormat<Row> {
private static final long serialVersionUID = 1L;
@@ -99,7 +103,8 @@ public class JDBCOutputFormat extends AbstractJdbcOutputFormat<Row> {
}
}
- void flush() {
+ @Override
+ public void flush() {
try {
upload.executeBatch();
batchCount = 0;
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 7585462..af4076a 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
@@ -20,6 +20,7 @@ package org.apache.flink.api.java.io.jdbc;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.api.java.io.jdbc.executor.JdbcBatchStatementExecutor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.api.java.typeutils.TupleTypeInfo;
@@ -69,7 +70,7 @@ public class JDBCUpsertTableSink implements UpsertStreamTableSink<Row> {
this.maxRetryTime = maxRetryTime;
}
- private JdbcBatchingOutputFormat newFormat() {
+ private JdbcBatchingOutputFormat<Tuple2<Boolean, Row>, Row, JdbcBatchStatementExecutor<Row>> newFormat() {
if (!isAppendOnly && (keyFields == null || keyFields.length == 0)) {
throw new UnsupportedOperationException("JDBCUpsertTableSink can not support ");
}
@@ -92,7 +93,7 @@ public class JDBCUpsertTableSink implements UpsertStreamTableSink<Row> {
@Override
public DataStreamSink<?> consumeDataStream(DataStream<Tuple2<Boolean, Row>> dataStream) {
return dataStream
- .addSink(new JdbcUpsertSinkFunction(newFormat()))
+ .addSink(new GenericJdbcSinkFunction<>(newFormat()))
.setParallelism(dataStream.getParallelism())
.name(TableConnectorUtils.generateRuntimeName(this.getClass(), schema.getFieldNames()));
}
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 213ad36..2d28c7a 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
@@ -18,9 +18,12 @@
package org.apache.flink.api.java.io.jdbc;
-import org.apache.flink.api.java.io.jdbc.dialect.JDBCDialect;
+import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.java.io.jdbc.executor.JdbcBatchStatementExecutor;
+import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.runtime.util.ExecutorThreadFactory;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.Preconditions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -31,10 +34,11 @@ import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
import static org.apache.flink.util.Preconditions.checkNotNull;
-abstract class JdbcBatchingOutputFormat<In, JdbcIn, JdbcExec extends JdbcBatchStatementExecutor<JdbcIn>> extends AbstractJdbcOutputFormat<In> {
+class JdbcBatchingOutputFormat<In, JdbcIn, JdbcExec extends JdbcBatchStatementExecutor<JdbcIn>> extends AbstractJdbcOutputFormat<In> {
private static final long serialVersionUID = 1L;
@@ -42,39 +46,29 @@ abstract class JdbcBatchingOutputFormat<In, JdbcIn, JdbcExec extends JdbcBatchSt
static final int DEFAULT_MAX_RETRY_TIMES = 3;
- protected final JdbcDmlOptions dmlOptions;
+ final JdbcDmlOptions dmlOptions;
private final JdbcBatchOptions batchOptions;
- protected transient JdbcExec jdbcStatementExecutor;
+ private transient JdbcExec jdbcStatementExecutor;
private transient int batchCount = 0;
private transient volatile boolean closed = false;
private transient ScheduledExecutorService scheduler;
- private transient ScheduledFuture scheduledFuture;
+ private transient ScheduledFuture<?> scheduledFuture;
private transient volatile Exception flushException;
-
- /**
- * @deprecated use {@link #JdbcBatchingOutputFormat(JdbcConnectionOptions, JdbcDmlOptions, JdbcBatchOptions)}
- */
- @Deprecated
- public JdbcBatchingOutputFormat(
- JDBCOptions options,
- String[] fieldNames,
- String[] keyFields,
- int[] fieldTypes,
- int flushMaxSize,
- long flushIntervalMills,
- int maxRetryTimes) {
- this(
- options,
- JdbcDmlOptions.builder().withFieldNames(fieldNames).withKeyFields(keyFields).withFieldTypes(fieldTypes).withMaxRetries(maxRetryTimes).build(),
- JdbcBatchOptions.builder().withSize(flushMaxSize).withIntervalMs(flushIntervalMills).build());
- }
-
- public JdbcBatchingOutputFormat(JdbcConnectionOptions connectionOptions, JdbcDmlOptions dmlOptions, JdbcBatchOptions batchOptions) {
+ 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) {
super(connectionOptions);
this.dmlOptions = dmlOptions;
this.batchOptions = batchOptions;
+ this.statementRunnerCreator = statementExecutorCreator;
+ this.jdbcRecordExtractor = recordExtractor;
}
/**
@@ -85,7 +79,7 @@ abstract class JdbcBatchingOutputFormat<In, JdbcIn, JdbcExec extends JdbcBatchSt
@Override
public void open(int taskNumber, int numTasks) throws IOException {
super.open(taskNumber, numTasks);
- jdbcStatementExecutor = createStatementRunner(dmlOptions.getDialect());
+ jdbcStatementExecutor = statementRunnerCreator.apply(getRuntimeContext());
try {
jdbcStatementExecutor.open(connection);
} catch (SQLException e) {
@@ -128,15 +122,12 @@ abstract class JdbcBatchingOutputFormat<In, JdbcIn, JdbcExec extends JdbcBatchSt
}
}
- abstract JdbcExec createStatementRunner(JDBCDialect dialect);
-
void doWriteRecord(In record) throws SQLException {
- jdbcStatementExecutor.process(extractJdbcRecord(record));
+ jdbcStatementExecutor.process(jdbcRecordExtractor.apply(record));
}
- abstract JdbcIn extractJdbcRecord(In record);
-
- public synchronized void flush() throws Exception {
+ @Override
+ public synchronized void flush() throws IOException {
checkFlushException();
for (int i = 1; i <= dmlOptions.getMaxRetries(); i++) {
@@ -147,9 +138,14 @@ abstract class JdbcBatchingOutputFormat<In, JdbcIn, JdbcExec extends JdbcBatchSt
} catch (SQLException e) {
LOG.error("JDBC executeBatch error, retry times = {}", i, e);
if (i >= dmlOptions.getMaxRetries()) {
- throw e;
+ throw new IOException(e);
+ }
+ try {
+ Thread.sleep(1000 * i);
+ } catch (InterruptedException ex) {
+ Thread.currentThread().interrupt();
+ throw new IOException("unable to flush; interrupted while doing another attempt", e);
}
- Thread.sleep(1000 * i);
}
}
}
@@ -218,7 +214,7 @@ abstract class JdbcBatchingOutputFormat<In, JdbcIn, JdbcExec extends JdbcBatchSt
/**
* required, field names of this jdbc sink.
*/
- public Builder setFieldNames(String[] fieldNames) {
+ Builder setFieldNames(String[] fieldNames) {
this.fieldNames = fieldNames;
return this;
}
@@ -226,7 +222,7 @@ abstract class JdbcBatchingOutputFormat<In, JdbcIn, JdbcExec extends JdbcBatchSt
/**
* required, upsert unique keys.
*/
- public Builder setKeyFields(String[] keyFields) {
+ Builder setKeyFields(String[] keyFields) {
this.keyFields = keyFields;
return this;
}
@@ -234,7 +230,7 @@ abstract class JdbcBatchingOutputFormat<In, JdbcIn, JdbcExec extends JdbcBatchSt
/**
* required, field types of this jdbc sink.
*/
- public Builder setFieldTypes(int[] fieldTypes) {
+ Builder setFieldTypes(int[] fieldTypes) {
this.fieldTypes = fieldTypes;
return this;
}
@@ -243,7 +239,7 @@ abstract class JdbcBatchingOutputFormat<In, JdbcIn, JdbcExec extends JdbcBatchSt
* optional, flush max size (includes all append, upsert and delete records),
* over this number of records, will flush data.
*/
- public Builder setFlushMaxSize(int flushMaxSize) {
+ Builder setFlushMaxSize(int flushMaxSize) {
this.flushMaxSize = flushMaxSize;
return this;
}
@@ -251,7 +247,7 @@ abstract class JdbcBatchingOutputFormat<In, JdbcIn, JdbcExec extends JdbcBatchSt
/**
* optional, flush interval mills, over this time, asynchronous threads will flush data.
*/
- public Builder setFlushIntervalMills(long flushIntervalMills) {
+ Builder setFlushIntervalMills(long flushIntervalMills) {
this.flushIntervalMills = flushIntervalMills;
return this;
}
@@ -259,7 +255,7 @@ abstract class JdbcBatchingOutputFormat<In, JdbcIn, JdbcExec extends JdbcBatchSt
/**
* optional, max retry times for jdbc connector.
*/
- public Builder setMaxRetryTimes(int maxRetryTimes) {
+ Builder setMaxRetryTimes(int maxRetryTimes) {
this.maxRetryTimes = maxRetryTimes;
return this;
}
@@ -269,19 +265,26 @@ abstract class JdbcBatchingOutputFormat<In, JdbcIn, JdbcExec extends JdbcBatchSt
*
* @return Configured JDBCUpsertOutputFormat
*/
- public JdbcBatchingOutputFormat build() {
+ public JdbcBatchingOutputFormat<Tuple2<Boolean, Row>, Row, JdbcBatchStatementExecutor<Row>> build() {
checkNotNull(options, "No options supplied.");
checkNotNull(fieldNames, "No fieldNames supplied.");
JdbcDmlOptions dml = JdbcDmlOptions.builder()
.withTableName(options.getTableName()).withDialect(options.getDialect())
.withFieldNames(fieldNames).withKeyFields(keyFields).withFieldTypes(fieldTypes).withMaxRetries(maxRetryTimes).build();
if (dml.getKeyFields() == null || dml.getKeyFields().length == 0) {
- return new JdbcBatchingOutputFormat(
+ // 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());
+ JdbcBatchOptions.builder().withSize(flushMaxSize).withIntervalMs(flushIntervalMills).build(),
+ unused -> JdbcBatchStatementExecutor.simpleRow(sql, dml.getFieldTypes()),
+ tuple2 -> {
+ Preconditions.checkArgument(tuple2.f0);
+ return tuple2.f1;
+ });
} else {
- return new TableJdbcOutputFormat(
+ return new TableJdbcUpsertOutputFormat(
options,
dml,
JdbcBatchOptions.builder().withSize(flushMaxSize).withIntervalMs(flushIntervalMills).build());
diff --git a/flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/JdbcSinkFunction.java b/flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/JdbcSinkFunction.java
index da3f611..05e069e 100644
--- a/flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/JdbcSinkFunction.java
+++ b/flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/JdbcSinkFunction.java
@@ -26,6 +26,10 @@ import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.types.Row;
+/**
+ * @deprecated use {@link GenericJdbcSinkFunction}
+ */
+@Deprecated
class JdbcSinkFunction extends RichSinkFunction<Row> implements CheckpointedFunction {
final JDBCOutputFormat outputFormat;
diff --git a/flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/JdbcTypedQueryOptions.java b/flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/JdbcTypedQueryOptions.java
index b4b409a..22d7eb6 100644
--- a/flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/JdbcTypedQueryOptions.java
+++ b/flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/JdbcTypedQueryOptions.java
@@ -30,7 +30,7 @@ abstract class JdbcTypedQueryOptions implements Serializable {
this.fieldTypes = fieldTypes;
}
- int[] getFieldTypes() {
+ public int[] getFieldTypes() {
return fieldTypes;
}
diff --git a/flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/TableJdbcOutputFormat.java b/flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/TableJdbcUpsertOutputFormat.java
similarity index 74%
rename from flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/TableJdbcOutputFormat.java
rename to flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/TableJdbcUpsertOutputFormat.java
index a9a5625..9935349 100644
--- a/flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/TableJdbcOutputFormat.java
+++ b/flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/TableJdbcUpsertOutputFormat.java
@@ -17,7 +17,6 @@
package org.apache.flink.api.java.io.jdbc;
-import org.apache.flink.api.java.io.jdbc.dialect.JDBCDialect;
import org.apache.flink.api.java.io.jdbc.executor.JdbcBatchStatementExecutor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.types.Row;
@@ -28,14 +27,20 @@ import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.sql.SQLException;
import java.util.Arrays;
+import java.util.function.Function;
-class TableJdbcOutputFormat extends JdbcBatchingOutputFormat<Tuple2<Boolean, Row>, Row, JdbcBatchStatementExecutor<Row>> {
- private static final Logger LOG = LoggerFactory.getLogger(TableJdbcOutputFormat.class);
+class TableJdbcUpsertOutputFormat extends JdbcBatchingOutputFormat<Tuple2<Boolean, Row>, Row, JdbcBatchStatementExecutor<Row>> {
+ private static final Logger LOG = LoggerFactory.getLogger(TableJdbcUpsertOutputFormat.class);
private JdbcBatchStatementExecutor<Row> deleteExecutor;
- TableJdbcOutputFormat(JdbcConnectionOptions connectionOptions, JdbcDmlOptions dmlOptions, JdbcBatchOptions batchOptions) {
- super(connectionOptions, dmlOptions, batchOptions);
+ 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;
+ }
+ });
}
@Override
@@ -58,27 +63,15 @@ class TableJdbcOutputFormat extends JdbcBatchingOutputFormat<Tuple2<Boolean, Row
}
@Override
- JdbcBatchStatementExecutor<Row> createStatementRunner(JDBCDialect dialect) {
- return JdbcBatchStatementExecutor.upsertRow(
- dialect, dmlOptions.getTableName(), dmlOptions.getFieldNames(), dmlOptions.getFieldTypes(), dmlOptions.getKeyFields(),
- getRuntimeContext().getExecutionConfig().isObjectReuseEnabled());
- }
-
- @Override
void doWriteRecord(Tuple2<Boolean, Row> record) throws SQLException {
if (record.f0) {
super.doWriteRecord(record);
} else {
- deleteExecutor.process(extractJdbcRecord(record));
+ deleteExecutor.process(jdbcRecordExtractor.apply(record));
}
}
@Override
- Row extractJdbcRecord(Tuple2<Boolean, Row> record) {
- return record.f1;
- }
-
- @Override
public synchronized void close() {
try {
super.close();
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 58632f6..3ca04f5 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
@@ -19,6 +19,8 @@
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;
@@ -54,32 +56,26 @@ public interface JdbcBatchStatementExecutor<T> {
*/
void close() throws SQLException;
- static JdbcBatchStatementExecutor<Row> upsertRow(
- JDBCDialect dialect,
- String tableName,
- String[] fieldNames,
- int[] fieldTypes,
- String[] keyFields,
- boolean objectReuse) {
+ static JdbcBatchStatementExecutor<Row> upsertRow(JdbcDmlOptions opt, RuntimeContext ctx) {
+ checkNotNull(opt.getKeyFields());
+ JDBCDialect dialect = opt.getDialectName().getInstance(opt);
- checkNotNull(keyFields);
-
- int[] pkFields = Arrays.stream(keyFields).mapToInt(Arrays.asList(fieldNames)::indexOf).toArray();
- int[] pkTypes = fieldTypes == null ? null : Arrays.stream(pkFields).map(f -> fieldTypes[f]).toArray();
+ 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
- .getUpsertStatement(tableName, fieldNames, keyFields)
- .map(sql -> keyedRow(pkFields, fieldTypes, sql))
+ .getUpsertStatement(opt.getTableName(), opt.getFieldNames(), opt.getKeyFields())
+ .map(sql -> keyedRow(pkFields, opt.getFieldTypes(), sql))
.orElseGet(() ->
new InsertOrUpdateJdbcExecutor<>(
- dialect.getRowExistsStatement(tableName, keyFields),
- dialect.getInsertIntoStatement(tableName, fieldNames),
- dialect.getUpdateStatement(tableName, fieldNames, keyFields),
+ dialect.getRowExistsStatement(opt.getTableName(), opt.getKeyFields()),
+ dialect.getInsertIntoStatement(opt.getTableName(), opt.getFieldNames()),
+ dialect.getUpdateStatement(opt.getTableName(), opt.getFieldNames(), opt.getKeyFields()),
ParameterSetter.forRow(pkTypes),
- ParameterSetter.forRow(fieldTypes),
- ParameterSetter.forRow(fieldTypes),
+ ParameterSetter.forRow(opt.getFieldTypes()),
+ ParameterSetter.forRow(opt.getFieldTypes()),
rowKeyExtractor(pkFields),
- objectReuse ? Row::copy : Function.identity()));
+ ctx.getExecutionConfig().isObjectReuseEnabled() ? Row::copy : Function.identity()));
}
static Function<Row, Row> rowKeyExtractor(int[] pkFields) {
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 2473f7d..5a105a9 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
@@ -47,7 +47,7 @@ import static org.mockito.Mockito.doReturn;
*/
public class JdbcTableOutputFormatTest extends JDBCTestBase {
- private TableJdbcOutputFormat format;
+ private TableJdbcUpsertOutputFormat format;
private String[] fieldNames;
private String[] keyFields;
@@ -66,7 +66,7 @@ public class JdbcTableOutputFormatTest extends JDBCTestBase {
JdbcDmlOptions dmlOptions = JdbcDmlOptions.builder()
.withTableName(options.getTableName()).withDialect(options.getDialect().getName())
.withFieldNames(fieldNames).withKeyFields(keyFields).build();
- format = new TableJdbcOutputFormat(options, dmlOptions, JdbcBatchOptions.defaults());
+ format = new TableJdbcUpsertOutputFormat(options, dmlOptions, JdbcBatchOptions.defaults());
RuntimeContext context = Mockito.mock(RuntimeContext.class);
ExecutionConfig config = Mockito.mock(ExecutionConfig.class);
doReturn(config).when(context).getExecutionConfig();