This is an automated email from the ASF dual-hosted git repository.
MartijnVisser pushed a commit to branch v4.1
in repository https://gitbox.apache.org/repos/asf/flink-connector-jdbc.git
The following commit(s) were added to refs/heads/v4.1 by this push:
new 45889cb3 [FLINK-40273][docs] Update JDBC documentation for the 4.x
module split and APIs
45889cb3 is described below
commit 45889cb39d6a5a9ea57f529fa1bf1d1da659de4d
Author: Purushottam Sinha <[email protected]>
AuthorDate: Thu Jul 30 15:12:40 2026 +0530
[FLINK-40273][docs] Update JDBC documentation for the 4.x module split and
APIs
The docs still described the 3.x connector. FLINK-36659 removed the
flink-connector-jdbc module, but docs/data/jdbc.yml was never updated,
and several features added since have gone undocumented.
Dependencies and artifacts:
- List one variant per published module in docs/data/jdbc.yml
instead of the removed aggregate artifact.
- Document the module split in both the Table and DataStream
dependency sections, including that no self-contained SQL uber jar
is published, that lib/ also needs the OpenLineage jars, and which
dialects need other dialect artifacts.
DataStream API:
- Rewrite the sink section for the Sink V2 builder
(JdbcSink.builder() / buildAtLeastOnce / buildExactlyOnce with
sinkTo). The documented JdbcSink.sink(), JdbcSink.exactlyOnceSink()
and JdbcXaSinkFunction were removed together with the old module.
- Lead the source section with setSplitter and document
PreparedSplitterEnumerator and SlideTimingSplitterEnumerator
(FLINK-38733), keeping the deprecated setSql path in its own
section with the migration.
- Fix JdbcSourceBuilder.builder() to JdbcSource.builder().
- Document the exactly-once requirements: TYPE_SCROLL_INSENSITIVE on
the source and maxRetries = 0 on the sink.
- Add a lineage section (FLINK-34467), noting that the namespace is
only set when the connector connects from a JDBC URL itself.
Table API:
- Correct scan.partition.lower-bound and upper-bound to Long.
- Document compatible-mode as OceanBase-only with its actual values.
- Rewrite the catalog base-url and default-database semantics for
FLINK-38851, including query-parameter passthrough and that a
base-url with a database but no query string pins the catalog.
- Add CrateDB and OceanBase to the upsert grammar table and describe
the insert-or-update fallback for Trino and Derby.
- Add PostgreSQL JSON/JSONB (FLINK-39224) and UUID (FLINK-38850) to
the data type mapping, including stringtype=unspecified for writes.
Applied to both docs/content and docs/content.zh.
Backport of 36caea9 to v4.1. Spanner (FLINK-37288), the
PreparedSplitterEnumerator.of(sql) example (FLINK-40747) and the
setContinuousUnBoundingSettings / setOptionalSqlSplitEnumeratorState
deprecations (FLINK-40746) are left out, as they are not on v4.1.
Generated-by: Claude Code (Claude Opus 5.5)
(cherry picked from commit 36caea9d91e3321ca1d2113be1133851d740a716)
---
docs/content.zh/docs/connectors/datastream/jdbc.md | 493 ++++++++++-----------
docs/content.zh/docs/connectors/table/jdbc.md | 110 ++++-
docs/content/docs/connectors/datastream/jdbc.md | 492 ++++++++++----------
docs/content/docs/connectors/table/jdbc.md | 125 ++++--
docs/data/jdbc.yml | 16 +
5 files changed, 665 insertions(+), 571 deletions(-)
diff --git a/docs/content.zh/docs/connectors/datastream/jdbc.md
b/docs/content.zh/docs/connectors/datastream/jdbc.md
index bdc45ec6..2539df35 100644
--- a/docs/content.zh/docs/connectors/datastream/jdbc.md
+++ b/docs/content.zh/docs/connectors/datastream/jdbc.md
@@ -29,9 +29,22 @@ under the License.
This connector provides a source that read data from a JDBC database and
provides a sink that writes data to a JDBC database.
-To use it, add the following dependency to your project (along with your JDBC
driver):
+Since version 4.0 the connector is no longer published as a single artifact.
It is split into
+`flink-connector-jdbc-core`, which contains the source and the sink, and one
artifact per supported
+database, which contains the dialect and the catalog for that database. Add
the artifact for the
+database you are connecting to (along with your JDBC driver); it pulls in
+`flink-connector-jdbc-core` transitively. For example, for PostgreSQL:
-{{< connector_artifact flink-connector-jdbc jdbc >}}
+{{< connector_artifact flink-connector-jdbc-postgres jdbc >}}
+
+The available database artifacts are `flink-connector-jdbc-mysql`,
`flink-connector-jdbc-oracle`,
+`flink-connector-jdbc-postgres`, `flink-connector-jdbc-sqlserver`,
`flink-connector-jdbc-cratedb`,
+`flink-connector-jdbc-db2`, `flink-connector-jdbc-trino` and
`flink-connector-jdbc-oceanbase`. The
+Derby dialect ships in `flink-connector-jdbc-core`.
+
+The `JdbcSource` and `JdbcSink` themselves live in `flink-connector-jdbc-core`
and work against any
+JDBC driver, so a database artifact is only required for the Table/SQL API,
where the dialect is used
+to generate statements.
Note that the streaming connectors are currently __NOT__ part of the binary
distribution.
See how to link with them for cluster execution [here]({{< ref
"docs/dev/configuration/overview" >}}).
@@ -42,28 +55,24 @@ Please consult your database documentation on how to add
the corresponding drive
### Usage
-{{< tabs "4ab65f13-607a-411a-8d24-e709f701cd41" >}}
-{{< tab "Java" >}}
```java
-JdbcSource source = JdbcSourceBuilder.builder()
+JdbcSource<OUT> source = JdbcSource.<OUT>builder()
// Required
- .setSql(...)
+ .setSplitter(...)
.setResultExtractor(...)
+ .setTypeInformation(...)
+ .setDBUrl(...)
+ .setDriverName(...)
.setUsername(...)
.setPassword(...)
- .setDriverName(...)
- .setDBUrl(...)
- .setTypeInformation(...)
-
- // Optional
- .setContinuousUnBoundingSettings(...)
- .setJdbcParameterValuesProvider(...)
+
+ // Optional
.setDeliveryGuarantee(...)
.setConnectionCheckTimeoutSeconds(...)
-
+
// The extended JDBC connection property passing
.setConnectionProperty("key", "value")
-
+
// other attributes
.setSplitReaderFetchBatchSize(...)
.setResultSetType(...)
@@ -72,15 +81,66 @@ JdbcSource source = JdbcSourceBuilder.builder()
.setResultSetFetchSize(...)
.setConnectionProvider(...)
.build();
+```
+`setSplitter` describes the query and how it is divided into splits, and is
the only required way to
+define what the source reads. The older `setSql` /
`setJdbcParameterValuesProvider` pair is
+deprecated in favour of it; see [Deprecated query API](#deprecated-query-api).
Setting both a
+splitter and a query fails at `build()` time.
+
+### SplitterEnumerator
+
+A `SplitterEnumerator` produces the `JdbcSourceSplit`s that the readers
execute, and decides whether
+the source is bounded or continuously unbounded.
+
+#### PreparedSplitterEnumerator
+
+`PreparedSplitterEnumerator` builds bounded splits from a parameterized query
template (i.e. a valid
+[JDBC prepared
statement](https://docs.oracle.com/en/java/javase/11/docs/api/java.sql/java/sql/PreparedStatement.html))
+together with the binding values. One split is generated per row of the
parameter array:
+
+```java
+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"};
+
+PreparedSplitterEnumerator.of(query, queryParameters);
```
-{{< /tab >}}
-{{< tab "Python" >}}
-```python
-Still not supported in Python API.
+
+For a numeric range there are convenience overloads that generate the
parameter pairs for you. The
+template must take a lower and an upper bound, and the range is divided either
into splits of a given
+size or into a given number of splits:
+
+```java
+// splits of at most 1000 values each
+PreparedSplitterEnumerator.of("select * from books where id between ? and ?",
1L, 10_000L, 1000L);
+
+// exactly 10 splits
+PreparedSplitterEnumerator.of("select * from books where id between ? and ?",
1L, 10_000L, 10);
```
-{{< /tab >}}
-{{< /tabs >}}
+
+Note that the two overloads differ only in whether the last argument is a
`long` (batch size) or an
+`int` (number of batches).
+
+#### SlideTimingSplitterEnumerator
+
+`SlideTimingSplitterEnumerator` generates splits continuously from a sliding
window over a timestamp
+column, which makes the source unbounded. The query template takes the window
start and end:
+
+```java
+SlideTimingSplitterEnumerator.builder()
+ .setSqlTemplate("select * from books where ts >= ? and ts < ?")
+ .setStartMillis(startMillis)
+ .setSlideSpanMillis(1000L)
+ .setSlideStepMillis(1000L)
+ .setSplitGenerateDelayMillis(100L)
+ .build();
+```
+
+`setStartMillis` is the epoch timestamp, in milliseconds, at which the first
window starts and must be
+greater than 0. `setSplitGenerateDelayMillis` holds a window back for the
given duration before it is
+emitted as a split, which gives late-arriving rows time to land.
### Delivery guarantee
@@ -93,12 +153,15 @@ remains unchanged in the whole lifecycle of
`JdbcSourceSplit` processing.
Unfortunately, this condition is not met in most databases and data scenarios.
See
[FLIP-239](https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=217386271)
for more details.
+Using `DeliveryGuarantee.EXACTLY_ONCE` requires
`setResultSetType(ResultSet.TYPE_SCROLL_INSENSITIVE)`;
+with the default `ResultSet.TYPE_FORWARD_ONLY`, `build()` fails.
+
### ResultExtractor
An `Extractor` to extract a record from `ResultSet` executed by a sql.
```java
-import org.apache.flink.connector.jdbc.source.reader.extractor.ResultExtractor;
+import
org.apache.flink.connector.jdbc.core.datastream.source.reader.extractor.ResultExtractor;
import java.sql.ResultSet;
import java.sql.SQLException;
@@ -113,73 +176,18 @@ class Book {
final String title;
};
-ResultExtractor resultExtractor = new ResultExtractor() {
+ResultExtractor<Book> resultExtractor = new ResultExtractor<Book>() {
@Override
- public Object extract(ResultSet resultSet) throws SQLException {
- return new Book(resultSet.getLong("id"),
resultSet.getString("titile"));
+ public Book extract(ResultSet resultSet) throws SQLException {
+ return new Book(resultSet.getLong("id"), resultSet.getString("title"));
}
};
-
-```
-
-### JdbcParameterValuesProvider
-
-A provider to provide parameters in sql to fulfill actual value in the
corresponding placeholders, which is in the form of two-dimension array.
-
-```java
-
-class TestEntry {
- ...
-};
-
-ResultSetExtractor extractor = ...;
-
-StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
-
-JdbcSource<TestEntry> jdbcSource =
- JdbcSource.<TestEntry>builder()
- .setTypeInformation(TypeInformation.of(TestEntry.class))
- .setSql("select * from testing_table where id >= ? and id <=
?")
- .setJdbcParameterValuesProvider(
- new JdbcGenericParameterValuesProvider(
- new Serializable[][] {{1001, 1005}, {1006,
1010}}))
- ...
- .build();
-env.fromSource(jdbcSource, WatermarkStrategy.noWatermarks(), "TestSource")
- .addSink(new DiscardSink());
-env.execute();
-
-```
-
-### Minimalist Streaming Semantic and ContinuousUnBoundingSettings
-
-If you want to generate continuous milliseconds parameters based on
sliding-window,
-please have a try on setting the followed attributes of `JdbcSource`:
-
-```java
-
-jdbcSourceBuilder =
- JdbcSource.<TestEntry>builder()
- .setSql("select * from testing_table where ts >= ? and ts < ?")
-
- // Required for streaming related semantic.
- .setContinuousUnBoundingSettings(new
ContinuousUnBoundingSettings(Duration.ofMillis(10L), Duration.ofSeconds(1L)))
- .setJdbcParameterValuesProvider(new
JdbcSlideTimingParameterProvider(0L, 1000L, 1000L, 100L))
- .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
- .setResultSetType(ResultSet.TYPE_SCROLL_INSENSITIVE);
-
- // other attributes
- ...
-
-JdbcSource source = jdbcSourceBuilder.build();
```
### Full example
-{{< tabs "4ab65f13-608a-411a-8d24-e303f348ds81" >}}
-{{< tab "Java" >}}
-
```java
+import org.apache.flink.streaming.api.functions.sink.v2.DiscardingSink;
public class JdbcSourceExample {
@@ -198,11 +206,11 @@ public class JdbcSourceExample {
JdbcSource<Book> jdbcSource =
JdbcSource.<Book>builder()
.setTypeInformation(TypeInformation.of(Book.class))
- .setSql("select * from testing_table where id < ?")
- .setDBUrl(...)
- .setJdbcParameterValuesProvider(
- new JdbcGenericParameterValuesProvider(
+ .setSplitter(
+ PreparedSplitterEnumerator.of(
+ "select * from books where id < ?",
new Serializable[][] {{1001L}}))
+ .setDBUrl(...)
.setDriverName(...)
.setResultExtractor(resultSet ->
new Book(
@@ -210,47 +218,76 @@ public class JdbcSourceExample {
resultSet.getString("title")))
.build();
env.fromSource(jdbcSource, WatermarkStrategy.noWatermarks(),
"TestSource")
- .addSink(new DiscardingSink());
+ .sinkTo(new DiscardingSink<>());
env.execute();
}
}
```
-{{< /tab >}}
-{{< tab "Python" >}}
-```python
-Still not supported in Python API.
+
+### Deprecated query API
+
+Before `SplitterEnumerator` was introduced, the query was defined with
`setSql` and the splits with a
+`JdbcParameterValuesProvider`. Both methods are deprecated.
`setContinuousUnBoundingSettings` and
+`setOptionalSqlSplitEnumeratorState` are only read on this path. A bounded
read maps directly to a
+`PreparedSplitterEnumerator`:
+
+```java
+// deprecated
+JdbcSource.<TestEntry>builder()
+ .setSql("select * from testing_table where id >= ? and id <= ?")
+ .setJdbcParameterValuesProvider(
+ new JdbcGenericParameterValuesProvider(
+ new Serializable[][] {{1001, 1005}, {1006, 1010}}))
+ ...
+
+// current
+JdbcSource.<TestEntry>builder()
+ .setSplitter(
+ PreparedSplitterEnumerator.of(
+ "select * from testing_table where id >= ? and id <=
?",
+ new Serializable[][] {{1001, 1005}, {1006, 1010}}))
+ ...
+```
+
+On this deprecated path, continuous unbounded reads are configured with
+`setContinuousUnBoundingSettings` and a `JdbcSlideTimingParameterProvider`,
which must be set
+together:
+
+```java
+// deprecated
+JdbcSource.<TestEntry>builder()
+ .setSql("select * from testing_table where ts >= ? and ts < ?")
+ .setContinuousUnBoundingSettings(
+ new ContinuousUnBoundingSettings(Duration.ofMillis(10L),
Duration.ofSeconds(1L)))
+ .setJdbcParameterValuesProvider(
+ new JdbcSlideTimingParameterProvider(startMillis, 1000L,
1000L, 100L))
+ .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
+ .setResultSetType(ResultSet.TYPE_SCROLL_INSENSITIVE);
```
-{{< /tab >}}
-{{< /tabs >}}
-## `JdbcSink.sink`
+The closest replacement is
[`SlideTimingSplitterEnumerator`](#slidetimingsplitterenumerator), but it
+is not like-for-like: it has no initial discovery delay and no discovery
interval. A window is emitted
+as a split as soon as it has closed and `splitGenerateDelayMillis` has passed.
+`setContinuousUnBoundingSettings` has no effect when a splitter is set.
+
+## JDBC Sink
-The JDBC sink provides at-least-once guarantee.
-Effectively though, exactly-once can be achieved by crafting upsert SQL
statements or idempotent SQL updates.
-Configuration goes as follows:
+The JDBC sink is built with `JdbcSink.builder()`. It provides an at-least-once
and an exactly-once
+variant, selected by which `build` method you call. Effectively though,
exactly-once can also be
+achieved on the at-least-once path by crafting upsert SQL statements or
idempotent SQL updates.
-{{< tabs "4ab65f13-607a-411a-8d24-e709f701cd4c" >}}
-{{< tab "Java" >}}
```java
-JdbcSink.sink(
- sqlDmlStatement, // mandatory
- jdbcStatementBuilder, // mandatory
- jdbcExecutionOptions, // optional
- jdbcConnectionOptions // mandatory
-);
+JdbcSink<IN> sink = JdbcSink.<IN>builder()
+ .withQueryStatement(sqlDmlStatement, jdbcStatementBuilder) //
mandatory
+ .withExecutionOptions(jdbcExecutionOptions) //
optional
+ .buildAtLeastOnce(jdbcConnectionOptions); //
mandatory
```
-{{< /tab >}}
-{{< tab "Python" >}}
-```python
-JdbcSink.sink(
- sql_dml_statement, # mandatory
- type_info, # mandatory
- jdbc_connection_options, # mandatory
- jdbc_execution_options # optional
-)
+
+The sink is attached to a stream with `sinkTo`:
+
+```java
+stream.sinkTo(sink);
```
-{{< /tab >}}
-{{< /tabs >}}
### SQL DML statement and JDBC statement builder
@@ -266,12 +303,14 @@ It then repeatedly calls a user-provided function to
update that prepared statem
(preparedStatement, someRecord) -> { ... update here the preparedStatement
with values from someRecord ... }
```
+The two are passed together to `withQueryStatement`. Alternatively, pass a
`JdbcQueryStatement`,
+which bundles both in one class. Either way the query is fixed: the writer
prepares it once, so it
+cannot vary per record.
+
### JDBC execution options
The SQL DML statements are executed in batches, which can optionally be
configured with the following instance:
-{{< tabs "4ab65f13-607a-411a-8d24-e709f512ed6k" >}}
-{{< tab "Java" >}}
```java
JdbcExecutionOptions.builder()
.withBatchIntervalMs(200) // optional: default = 0,
meaning no time-based execution is done
@@ -279,17 +318,6 @@ JdbcExecutionOptions.builder()
.withMaxRetries(5) // optional: default = 3
.build();
```
-{{< /tab >}}
-{{< tab "Python" >}}
-```python
-JdbcExecutionOptions.builder() \
- .with_batch_interval_ms(2000) \
- .with_batch_size(100) \
- .with_max_retries(5) \
- .build()
-```
-{{< /tab >}}
-{{< /tabs >}}
A JDBC batch is executed as soon as one of the following conditions is true:
@@ -299,12 +327,12 @@ A JDBC batch is executed as soon as one of the following
conditions is true:
### JDBC connection parameters
-The connection to the database is configured with a `JdbcConnectionOptions`
instance.
+The connection to the database is configured with a `JdbcConnectionOptions`
instance, which is passed
+to `buildAtLeastOnce`. A `JdbcConnectionProvider` can be passed instead to
reuse an existing
+connection strategy.
### Full example
-{{< tabs "4ab65f13-608a-411a-8d24-e303f348ds8d" >}}
-{{< tab "Java" >}}
```java
public class JdbcSinkExample {
@@ -324,138 +352,94 @@ public class JdbcSinkExample {
public static void main(String[] args) throws Exception {
var env = StreamExecutionEnvironment.getExecutionEnvironment();
- env.fromElements(
+ env.fromData(
new Book(101L, "Stream Processing with Apache Flink", "Fabian
Hueske, Vasiliki Kalavri", 2019),
new Book(102L, "Streaming Systems", "Tyler Akidau, Slava
Chernyak, Reuven Lax", 2018),
new Book(103L, "Designing Data-Intensive Applications",
"Martin Kleppmann", 2017),
new Book(104L, "Kafka: The Definitive Guide", "Gwen Shapira,
Neha Narkhede, Todd Palino", 2017)
- ).addSink(
- JdbcSink.sink(
- "insert into books (id, title, authors, year) values
(?, ?, ?, ?)",
- (statement, book) -> {
- statement.setLong(1, book.id);
- statement.setString(2, book.title);
- statement.setString(3, book.authors);
- statement.setInt(4, book.year);
- },
- JdbcExecutionOptions.builder()
- .withBatchSize(1000)
- .withBatchIntervalMs(200)
- .withMaxRetries(5)
- .build(),
- new
JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
-
.withUrl("jdbc:postgresql://dbhost:5432/postgresdb")
- .withDriverName("org.postgresql.Driver")
- .withUsername("someUser")
- .withPassword("somePassword")
- .build()
- ));
-
+ ).sinkTo(
+ JdbcSink.<Book>builder()
+ .withQueryStatement(
+ "insert into books (id, title, authors, year)
values (?, ?, ?, ?)",
+ (statement, book) -> {
+ statement.setLong(1, book.id);
+ statement.setString(2, book.title);
+ statement.setString(3, book.authors);
+ statement.setInt(4, book.year);
+ })
+ .withExecutionOptions(
+ JdbcExecutionOptions.builder()
+ .withBatchSize(1000)
+ .withBatchIntervalMs(200)
+ .withMaxRetries(5)
+ .build())
+ .buildAtLeastOnce(
+ new
JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
+
.withUrl("jdbc:postgresql://dbhost:5432/postgresdb")
+
.withDriverName("org.postgresql.Driver")
+ .withUsername("someUser")
+ .withPassword("somePassword")
+ .build()));
+
env.execute();
}
}
```
-{{< /tab >}}
-{{< tab "Python" >}}
-```python
-env = StreamExecutionEnvironment.get_execution_environment()
-type_info = Types.ROW([Types.INT(), Types.STRING(), Types.STRING(),
Types.INT()])
-env.from_collection(
- [(101, "Stream Processing with Apache Flink", "Fabian Hueske, Vasiliki
Kalavri", 2019),
- (102, "Streaming Systems", "Tyler Akidau, Slava Chernyak, Reuven Lax",
2018),
- (103, "Designing Data-Intensive Applications", "Martin Kleppmann", 2017),
- (104, "Kafka: The Definitive Guide", "Gwen Shapira, Neha Narkhede, Todd
Palino", 2017)
- ], type_info=type_info) \
- .add_sink(
- JdbcSink.sink(
- "insert into books (id, title, authors, year) values (?, ?, ?, ?)",
- type_info,
- JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
- .with_url('jdbc:postgresql://dbhost:5432/postgresdb')
- .with_driver_name('org.postgresql.Driver')
- .with_user_name('someUser')
- .with_password('somePassword')
- .build(),
- JdbcExecutionOptions.builder()
- .with_batch_interval_ms(1000)
- .with_batch_size(200)
- .with_max_retries(5)
- .build()
- ))
-
-env.execute()
-```
-{{< /tab >}}
-{{< /tabs >}}
-## `JdbcSink.exactlyOnceSink`
+### Exactly-once sink
-Since 1.13, Flink JDBC sink supports exactly-once mode.
-The implementation relies on the JDBC driver support of XA
+The exactly-once implementation relies on the JDBC driver support of XA
[standard](https://pubs.opengroup.org/onlinepubs/009680699/toc.pdf).
Most drivers support XA if the database also supports XA (so the driver is
usually the same).
-To use it, create a sink using `exactlyOnceSink()` method as above and
additionally provide:
+To use it, call `buildExactlyOnce()` instead of `buildAtLeastOnce()` and
provide:
- `JdbcExactlyOnceOptions`
-- `JdbcExecutionOptions`
-- [XA
DataSource](https://docs.oracle.com/javase/8/docs/api/javax/sql/XADataSource.html)
Supplier
+- an [XA
DataSource](https://docs.oracle.com/javase/8/docs/api/javax/sql/XADataSource.html)
Supplier
For example:
-{{< tabs "4ab65f13-608a-411a-8d24-e304f627ac8f" >}}
-{{< tab "Java" >}}
```java
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env
- .fromElements(...)
- .addSink(
- JdbcSink.exactlyOnceSink(
- "insert into books (id, title, author, price, qty) values
(?,?,?,?,?)",
- (ps, t) -> {
- ps.setInt(1, t.id);
- ps.setString(2, t.title);
- ps.setString(3, t.author);
- ps.setDouble(4, t.price);
- ps.setInt(5, t.qty);
- },
- JdbcExecutionOptions.builder()
- .withMaxRetries(0)
- .build(),
- JdbcExactlyOnceOptions.defaults(),
- () -> {
- // create a driver-specific XA DataSource
- // The following example is for derby
- EmbeddedXADataSource ds = new EmbeddedXADataSource();
- ds.setDatabaseName("my_db");
- return ds;
- }));
+ .fromData(...)
+ .sinkTo(JdbcSink.<Book>builder()
+ .withQueryStatement(
+ "insert into books (id, title, author, price, qty)
values (?,?,?,?,?)",
+ (ps, t) -> {
+ ps.setInt(1, t.id);
+ ps.setString(2, t.title);
+ ps.setString(3, t.author);
+ ps.setDouble(4, t.price);
+ ps.setInt(5, t.qty);
+ })
+ .withExecutionOptions(
+ JdbcExecutionOptions.builder()
+ .withMaxRetries(0)
+ .build())
+ .buildExactlyOnce(
+ JdbcExactlyOnceOptions.defaults(),
+ () -> {
+ // create a driver-specific XA DataSource
+ PGXADataSource ds = new PGXADataSource();
+
ds.setUrl("jdbc:postgresql://localhost:5432/postgres");
+ ds.setUser(username);
+ ds.setPassword(password);
+ return ds;
+ }));
env.execute();
```
-{{< /tab >}}
-{{< tab "Python" >}}
-```python
-Still not supported in Python API.
-```
-{{< /tab >}}
-{{< /tabs >}}
+
+An `XaConnectionProvider` can be passed instead of the supplier when the
connection strategy has to
+be controlled directly.
**NOTE:** Some databases only allow a single XA transaction per connection
(e.g. PostgreSQL, MySQL).
In such cases, please use the following API to construct
`JdbcExactlyOnceOptions`:
-{{< tabs "4ab65f13-608a-411a-8d24-e304f627cd4e" >}}
-{{< tab "Java" >}}
```java
JdbcExactlyOnceOptions.builder()
.withTransactionPerConnection(true)
.build();
```
-{{< /tab >}}
-{{< tab "Python" >}}
-```python
-Still not supported in Python API.
-```
-{{< /tab >}}
-{{< /tabs >}}
This will make Flink use a separate connection for every XA transaction. This
may require adjusting connection limits.
For PostgreSQL and MySQL, this can be done by increasing `max_connections`.
@@ -464,63 +448,48 @@ Furthermore, XA needs to be enabled and/or configured in
some databases.
For PostgreSQL, you should set `max_prepared_transactions` to some value
greater than zero.
For MySQL v8+, you should grant `XA_RECOVER_ADMIN` to Flink DB user.
-**ATTENTION:** Currently, `JdbcSink.exactlyOnceSink` can ensure exactly once
semantics
-with `JdbcExecutionOptions.maxRetries == 0`; otherwise, duplicated results
maybe produced.
+**ATTENTION:** The exactly-once sink requires
`JdbcExecutionOptions.maxRetries` to be set to `0`.
+The default is `3`, so set it explicitly with
`JdbcExecutionOptions.builder().withMaxRetries(0)`;
+otherwise the sink fails at startup.
### `XADataSource` examples
PostgreSQL `XADataSource` example:
-{{< tabs "4ab65f13-608a-411a-8d24-e304f323ab3a" >}}
-{{< tab "Java" >}}
```java
PGXADataSource xaDataSource = new org.postgresql.xa.PGXADataSource();
xaDataSource.setUrl("jdbc:postgresql://localhost:5432/postgres");
xaDataSource.setUser(username);
xaDataSource.setPassword(password);
```
-{{< /tab >}}
-{{< tab "Python" >}}
-```python
-Still not supported in Python API.
-```
-{{< /tab >}}
-{{< /tabs >}}
MySQL `XADataSource` example:
-{{< tabs "4ab65f13-608a-411a-8d24-b213f323ca3c" >}}
-{{< tab "Java" >}}
```java
MysqlXADataSource xaDataSource = new com.mysql.cj.jdbc.MysqlXADataSource();
xaDataSource.setUrl("jdbc:mysql://localhost:3306/");
xaDataSource.setUser(username);
xaDataSource.setPassword(password);
```
-{{< /tab >}}
-{{< tab "Python" >}}
-```python
-Still not supported in Python API.
-```
-{{< /tab >}}
-{{< /tabs >}}
Oracle `XADataSource` example:
-{{< tabs "4ab65f13-608a-411a-8d24-b213f337ad5f" >}}
-{{< tab "Java" >}}
```java
OracleXADataSource xaDataSource = new oracle.jdbc.xa.OracleXADataSource();
xaDataSource.setURL("jdbc:oracle:oci8:@");
xaDataSource.setUser("scott");
xaDataSource.setPassword("tiger");
```
-{{< /tab >}}
-{{< tab "Python" >}}
-```python
-Still not supported in Python API.
-```
-{{< /tab >}}
-{{< /tabs >}}
Please also take Oracle connection pooling into account.
-Please refer to the `JdbcXaSinkFunction` documentation for more details.
+## Lineage
+
+`JdbcSource` and `JdbcSink` expose lineage information as described in
+[FLIP-314](https://cwiki.apache.org/confluence/display/FLINK/FLIP-314%3A+Support+Customized+Job+Lineage+Listener),
+as do the Table API source and lookup function. No configuration is needed on
the connector side; the
+information is delivered to any `JobStatusChangedListener` registered in the
cluster.
+
+The dataset name is derived from the queries the job executes. The dataset
namespace is derived
+from the JDBC URL, but only when the connector opens connections from that URL
itself, i.e. a
+`JdbcSource` configured with `setDBUrl` or a `JdbcSink` built with
+`buildAtLeastOnce(JdbcConnectionOptions)`. With `buildExactlyOnce` or a custom
+`JdbcConnectionProvider` the namespace is empty.
{{< top >}}
diff --git a/docs/content.zh/docs/connectors/table/jdbc.md
b/docs/content.zh/docs/connectors/table/jdbc.md
index 0829c3aa..dd20afec 100644
--- a/docs/content.zh/docs/connectors/table/jdbc.md
+++ b/docs/content.zh/docs/connectors/table/jdbc.md
@@ -41,25 +41,32 @@ JDBC 连接器允许使用 JDBC 驱动向任意类型的关系型数据库读取
依赖
------------
-使用 JDBC 连接器时, `flink-connector-jdbc-core`, `flink-connector-jdbc-${database}`
和对应的 JDBC 驱动需要添加到依赖中。
+从 4.0 版本开始，JDBC 连接器不再作为单个构件发布。它被拆分为包含公共运行时的
+`flink-connector-jdbc-core`，以及每个支持的数据库各一个构件，后者包含该数据库的方言（dialect）和
+catalog。你需要引入所连接数据库对应的构件，它会以传递依赖的方式引入 `flink-connector-jdbc-core`。
+
+| 数据库 | 连接器构件 |
+|:-----------|:----------------------------------|
+| MySQL | `flink-connector-jdbc-mysql` |
+| Oracle | `flink-connector-jdbc-oracle` |
+| PostgreSQL | `flink-connector-jdbc-postgres` |
+| SQL Server | `flink-connector-jdbc-sqlserver` |
+| CrateDB | `flink-connector-jdbc-cratedb` |
+| Db2 | `flink-connector-jdbc-db2` |
+| Trino | `flink-connector-jdbc-trino` |
+| OceanBase | `flink-connector-jdbc-oceanbase` |
+| Derby | `flink-connector-jdbc-core` |
{{< sql_connector_download_table "jdbc" >}}
-连接到对应的数据库时，需要使用对应的 `flink-connector-jdbc-${database}` 依赖和驱动依赖。
+这些构件都不是自包含的 SQL uber jar。在使用 SQL Client 或 session 集群时，需要把
+`flink-connector-jdbc-core` 和所用数据库对应的构件同时放入 `lib/` 目录，并且还需要放入
+`flink-connector-jdbc-core` 所依赖的 `io.openlineage:openlineage-java` 和
+`io.openlineage:openlineage-sql-java` jar，否则构建作业时会抛出
`NoClassDefFoundError`。部分方言基于其他方言实现，
+因此还需要对应的构件：CrateDB 还需要 `flink-connector-jdbc-postgres`，OceanBase 还需要
+`flink-connector-jdbc-mysql` 和 `flink-connector-jdbc-oracle`。
-支持的 `flink-connector-jdbc-${database}` 依赖如下，其版本需要和 `flink-connector-jdbc-core`
依赖的版本相同：
-
-| Database | Group Id | Artifact Id |
JAR
|
-|:-----------|:--------------------------|:---------------------------------|:--------------------------------------------------------------------------------------------|
-| MySQL | `org.apache.flink` | `flink-connector-jdbc-mysql` |
[下载](https://repo.maven.apache.org/maven2/org/apache/flink/flink-connector-jdbc-mysql/)
|
-| Oracle | `org.apache.flink` | `flink-connector-jdbc-oracle` |
[下载](https://repo.maven.apache.org/maven2/org/apache/flink/flink-connector-jdbc-oracle/)
|
-| PostgreSQL | `org.apache.flink` | `flink-connector-jdbc-postgres` |
[下载](https://repo.maven.apache.org/maven2/org/apache/flink/flink-connector-jdbc-postgres/)
|
-| Derby | `org.apache.flink` | `flink-connector-jdbc-core` |
[下载](https://repo.maven.apache.org/maven2/org/apache/flink/flink-connector-jdbc-core/)
|
-| SQL Server | `org.apache.flink` | `flink-connector-jdbc-sqlserver` |
[下载](https://repo.maven.apache.org/maven2/org/apache/flink/flink-connector-jdbc-sqlserver/)
|
-| CrateDB | `org.apache.flink` | `flink-connector-jdbc-cratedb` |
[下载](https://repo.maven.apache.org/maven2/org/apache/flink/flink-connector-jdbc-cratedb/)
|
-| Db2 | `org.apache.flink` | `flink-connector-jdbc-db2` |
[下载](https://repo.maven.apache.org/maven2/org/apache/flink/flink-connector-jdbc-db2/)
|
-| Trino | `org.apache.flink` | `flink-connector-jdbc-trino` |
[下载](https://repo.maven.apache.org/maven2/org/apache/flink/flink-connector-jdbc-trino/)
|
-| OceanBase | `org.apache.flink` | `flink-connector-jdbc-oceanbase` |
[下载](https://repo.maven.apache.org/maven2/org/apache/flink/flink-connector-jdbc-oceanbase/)
|
+JDBC 连接器不是二进制发行版的一部分，请查阅[这里]({{< ref "docs/dev/configuration/overview"
>}})了解如何在集群运行中引用 JDBC 连接器。
在连接到具体数据库时，也需要对应的驱动依赖，目前支持的驱动如下：
@@ -161,7 +168,9 @@ ON myTopic.key = MyUserTable.id;
<td>可选</td>
<td style="word-wrap: break-word;">(none)</td>
<td>String</td>
- <td>数据库的兼容模式。</td>
+ <td>数据库的兼容模式。只有 OceanBase 支持该配置项，取值为 <code>'mysql'</code>（未设置时的默认值）
+ 和 <code>'oracle'</code>，用于选择方言所针对的租户模式。为其他数据库设置该项会以
+ <code>Not supported option 'compatible-mode'</code> 失败。</td>
</tr>
<tr>
<td><h5>username</h5></td>
@@ -202,14 +211,14 @@ ON myTopic.key = MyUserTable.id;
<td><h5>scan.partition.lower-bound</h5></td>
<td>可选</td>
<td style="word-wrap: break-word;">(none)</td>
- <td>Integer</td>
+ <td>Long</td>
<td>第一个分区的最小值。</td>
</tr>
<tr>
<td><h5>scan.partition.upper-bound</h5></td>
<td>可选</td>
<td style="word-wrap: break-word;">(none)</td>
- <td>Integer</td>
+ <td>Long</td>
<td>最后一个分区的最大值。</td>
</tr>
<tr>
@@ -450,9 +459,22 @@ lookup cache 的主要目的是用于提高时态表关联 JDBC 连接器的性
WHEN NOT MATCHED THEN INSERT (..) <br>
VALUES (..)</td>
</tr>
+ <tr>
+ <td>CrateDB</td>
+ <td>INSERT .. ON CONFLICT .. DO UPDATE SET ..</td>
+ </tr>
+ <tr>
+ <td>OceanBase</td>
+ <td>取决于 <code>'compatible-mode'</code>：<code>'mysql'</code> 模式下使用
MySQL 语法，
+ <code>'oracle'</code> 模式下使用 Oracle 语法。</td>
+ </tr>
</tbody>
</table>
+Trino 和 Derby 没有 upsert 语法。如果在 DDL 中定义了主键，sink 会先检查该主键对应的行是否存在，
+然后执行 `UPDATE` 或 `INSERT`，并对撤回消息执行 `DELETE`。对于 Trino，只有当表所在的 Trino
+connector 支持 `UPDATE` 和 `DELETE` 时才能正常工作。
+
<a name="jdbc-catalog"></a>
JDBC Catalog
@@ -482,14 +504,39 @@ tableExists(ObjectPath tablePath);
JDBC catalog 支持以下参数:
- `name`：必填，catalog 的名称。
-- `default-database`：必填，默认要连接的数据库。
- `username`：必填，数据库账户的用户名。
- `password`：必填，账户的密码。
-- `base-url`：必填，（不应该包含数据库名）
+- `base-url`：必填
- 对于 Postgres Catalog `base-url` 应为 `"jdbc:postgresql://<ip>:<port>"` 的格式。
- 对于 MySQL Catalog `base-url` 应为 `"jdbc:mysql://<ip>:<port>"` 的格式。
- 对于 OceanBase Catalog `base-url` 应为 `"jdbc:oceanbase://<ip>:<port>"` 的格式。
-- `compatible-mode`: 选填，数据库的兼容模式。
+- `default-database`：选填，默认要连接的数据库。
+- `compatible-mode`: 选填，数据库的兼容模式。只有 OceanBase 支持该配置项，
+ 参见 [连接器参数](#connector-options)。
+
+catalog 所连接的数据库必须能够从 `base-url`、`default-database` 或两者中确定。下列任意一种方式都是合法的：
+
+- 设置了 `default-database` 且 `base-url` 中不含数据库名，例如
+ `'base-url' = 'jdbc:postgresql://localhost:5432'` 配合 `'default-database' =
'mydb'`
+- `base-url` 中含有数据库名且省略 `default-database`，例如
+ `'base-url' = 'jdbc:postgresql://localhost:5432/mydb'`。如果不带查询字符串，这种写法会把
catalog
+ 固定在 `mydb` 上：catalog 中的所有数据库都通过同一个 URL 访问。
+- 两者都设置且数据库名一致
+
+如果两者不一致，或者两者都没有给出数据库名，创建 catalog 会失败。
+
+`base-url` 还可以通过查询参数携带任意的驱动参数，它们会被传递给 catalog 打开的每一个连接。
+数据库名会被替换到查询字符串之前，因此这些参数对该 catalog 下的所有数据库都生效。当 `base-url`
+带有查询字符串时，`base-url` 本身必须包含数据库名，只设置 `default-database` 会被拒绝：
+
+```sql
+CREATE CATALOG my_catalog WITH(
+ 'type' = 'jdbc',
+ 'username' = '...',
+ 'password' = '...',
+ 'base-url' =
'jdbc:postgresql://localhost:5432/mydb?ssl=true&connectTimeout=10'
+);
+```
{{< tabs "10bd8bfb-674c-46aa-8a36-385537df5791" >}}
{{< tab "SQL" >}}
@@ -979,7 +1026,9 @@ Flink 支持连接到多个使用方言（dialect）的数据库，如 MySQL、O
<code>CHARACTER(n)</code><br>
<code>VARCHAR(n)</code><br>
<code>CHARACTER VARYING(n)</code><br>
- <code>TEXT</code></td>
+ <code>TEXT</code><br>
+ <code>JSON</code><br>
+ <code>JSONB</code></td>
<td>
<code>CHAR(n)</code><br>
<code>CHARACTER(n)</code><br>
@@ -1038,6 +1087,18 @@ Flink 支持连接到多个使用方言（dialect）的数据库，如 MySQL、O
<code>BLOB</code></td>
<td><code>BYTES</code></td>
</tr>
+ <tr>
+ <td></td>
+ <td></td>
+ <td><code>UUID</code></td>
+ <td></td>
+ <td></td>
+ <td></td>
+ <td></td>
+ <td></td>
+ <td></td>
+ <td><code>VARCHAR(36)</code></td>
+ </tr>
<tr>
<td></td>
<td></td>
@@ -1053,4 +1114,9 @@ Flink 支持连接到多个使用方言（dialect）的数据库，如 MySQL、O
</tbody>
</table>
+PostgreSQL 的 `JSON` 和 `JSONB` 列以其文本表示形式读取。`UUID` 列映射为 `VARCHAR(36)`；
+在 DDL 中将这类列声明为 `STRING` 同样可以工作。sink 会以字符串形式绑定这三种类型的参数，而 PostgreSQL
+会拒绝向这些类型的列写入字符串，除非 JDBC URL 设置了 `stringtype=unspecified`，例如
+`'url' = 'jdbc:postgresql://localhost:5432/mydb?stringtype=unspecified'`。
+
{{< top >}}
diff --git a/docs/content/docs/connectors/datastream/jdbc.md
b/docs/content/docs/connectors/datastream/jdbc.md
index 4a6f4a37..a40ec840 100644
--- a/docs/content/docs/connectors/datastream/jdbc.md
+++ b/docs/content/docs/connectors/datastream/jdbc.md
@@ -29,9 +29,22 @@ under the License.
This connector provides a source that read data from a JDBC database and
provides a sink that writes data to a JDBC database.
-To use it, add the following dependency to your project (along with your JDBC
driver):
+Since version 4.0 the connector is no longer published as a single artifact.
It is split into
+`flink-connector-jdbc-core`, which contains the source and the sink, and one
artifact per supported
+database, which contains the dialect and the catalog for that database. Add
the artifact for the
+database you are connecting to (along with your JDBC driver); it pulls in
+`flink-connector-jdbc-core` transitively. For example, for PostgreSQL:
-{{< connector_artifact flink-connector-jdbc jdbc >}}
+{{< connector_artifact flink-connector-jdbc-postgres jdbc >}}
+
+The available database artifacts are `flink-connector-jdbc-mysql`,
`flink-connector-jdbc-oracle`,
+`flink-connector-jdbc-postgres`, `flink-connector-jdbc-sqlserver`,
`flink-connector-jdbc-cratedb`,
+`flink-connector-jdbc-db2`, `flink-connector-jdbc-trino` and
`flink-connector-jdbc-oceanbase`. The
+Derby dialect ships in `flink-connector-jdbc-core`.
+
+The `JdbcSource` and `JdbcSink` themselves live in `flink-connector-jdbc-core`
and work against any
+JDBC driver, so a database artifact is only required for the Table/SQL API,
where the dialect is used
+to generate statements.
Note that the streaming connectors are currently __NOT__ part of the binary
distribution.
See how to link with them for cluster execution [here]({{< ref
"docs/dev/configuration/overview" >}}).
@@ -42,28 +55,24 @@ Please consult your database documentation on how to add
the corresponding drive
### Usage
-{{< tabs "4ab65f13-607a-411a-8d24-e709f701cd41" >}}
-{{< tab "Java" >}}
```java
-JdbcSource source = JdbcSourceBuilder.builder()
+JdbcSource<OUT> source = JdbcSource.<OUT>builder()
// Required
- .setSql(...)
+ .setSplitter(...)
.setResultExtractor(...)
+ .setTypeInformation(...)
+ .setDBUrl(...)
+ .setDriverName(...)
.setUsername(...)
.setPassword(...)
- .setDriverName(...)
- .setDBUrl(...)
- .setTypeInformation(...)
-
- // Optional
- .setContinuousUnBoundingSettings(...)
- .setJdbcParameterValuesProvider(...)
+
+ // Optional
.setDeliveryGuarantee(...)
.setConnectionCheckTimeoutSeconds(...)
-
+
// The extended JDBC connection property passing
.setConnectionProperty("key", "value")
-
+
// other attributes
.setSplitReaderFetchBatchSize(...)
.setResultSetType(...)
@@ -72,15 +81,66 @@ JdbcSource source = JdbcSourceBuilder.builder()
.setResultSetFetchSize(...)
.setConnectionProvider(...)
.build();
+```
+`setSplitter` describes the query and how it is divided into splits, and is
the only required way to
+define what the source reads. The older `setSql` /
`setJdbcParameterValuesProvider` pair is
+deprecated in favour of it; see [Deprecated query API](#deprecated-query-api).
Setting both a
+splitter and a query fails at `build()` time.
+
+### SplitterEnumerator
+
+A `SplitterEnumerator` produces the `JdbcSourceSplit`s that the readers
execute, and decides whether
+the source is bounded or continuously unbounded.
+
+#### PreparedSplitterEnumerator
+
+`PreparedSplitterEnumerator` builds bounded splits from a parameterized query
template (i.e. a valid
+[JDBC prepared
statement](https://docs.oracle.com/en/java/javase/11/docs/api/java.sql/java/sql/PreparedStatement.html))
+together with the binding values. One split is generated per row of the
parameter array:
+
+```java
+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"};
+
+PreparedSplitterEnumerator.of(query, queryParameters);
```
-{{< /tab >}}
-{{< tab "Python" >}}
-```python
-Still not supported in Python API.
+
+For a numeric range there are convenience overloads that generate the
parameter pairs for you. The
+template must take a lower and an upper bound, and the range is divided either
into splits of a given
+size or into a given number of splits:
+
+```java
+// splits of at most 1000 values each
+PreparedSplitterEnumerator.of("select * from books where id between ? and ?",
1L, 10_000L, 1000L);
+
+// exactly 10 splits
+PreparedSplitterEnumerator.of("select * from books where id between ? and ?",
1L, 10_000L, 10);
```
-{{< /tab >}}
-{{< /tabs >}}
+
+Note that the two overloads differ only in whether the last argument is a
`long` (batch size) or an
+`int` (number of batches).
+
+#### SlideTimingSplitterEnumerator
+
+`SlideTimingSplitterEnumerator` generates splits continuously from a sliding
window over a timestamp
+column, which makes the source unbounded. The query template takes the window
start and end:
+
+```java
+SlideTimingSplitterEnumerator.builder()
+ .setSqlTemplate("select * from books where ts >= ? and ts < ?")
+ .setStartMillis(startMillis)
+ .setSlideSpanMillis(1000L)
+ .setSlideStepMillis(1000L)
+ .setSplitGenerateDelayMillis(100L)
+ .build();
+```
+
+`setStartMillis` is the epoch timestamp, in milliseconds, at which the first
window starts and must be
+greater than 0. `setSplitGenerateDelayMillis` holds a window back for the
given duration before it is
+emitted as a split, which gives late-arriving rows time to land.
### Delivery guarantee
@@ -93,12 +153,15 @@ remains unchanged in the whole lifecycle of
`JdbcSourceSplit` processing.
Unfortunately, this condition is not met in most databases and data scenarios.
See
[FLIP-239](https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=217386271)
for more details.
+Using `DeliveryGuarantee.EXACTLY_ONCE` requires
`setResultSetType(ResultSet.TYPE_SCROLL_INSENSITIVE)`;
+with the default `ResultSet.TYPE_FORWARD_ONLY`, `build()` fails.
+
### ResultExtractor
An `Extractor` to extract a record from `ResultSet` executed by a sql.
```java
-import org.apache.flink.connector.jdbc.source.reader.extractor.ResultExtractor;
+import
org.apache.flink.connector.jdbc.core.datastream.source.reader.extractor.ResultExtractor;
import java.sql.ResultSet;
import java.sql.SQLException;
@@ -113,73 +176,18 @@ class Book {
final String title;
};
-ResultExtractor resultExtractor = new ResultExtractor() {
+ResultExtractor<Book> resultExtractor = new ResultExtractor<Book>() {
@Override
- public Object extract(ResultSet resultSet) throws SQLException {
- return new Book(resultSet.getLong("id"),
resultSet.getString("titile"));
+ public Book extract(ResultSet resultSet) throws SQLException {
+ return new Book(resultSet.getLong("id"), resultSet.getString("title"));
}
};
-
-```
-
-### JdbcParameterValuesProvider
-
-A provider to provide parameters in sql to fulfill actual value in the
corresponding placeholders, which is in the form of two-dimension array.
-
-```java
-
-class TestEntry {
- ...
-};
-
-ResultSetExtractor extractor = ...;
-
-StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
-
-JdbcSource<TestEntry> jdbcSource =
- JdbcSource.<TestEntry>builder()
- .setTypeInformation(TypeInformation.of(TestEntry.class))
- .setSql("select * from testing_table where id >= ? and id <=
?")
- .setJdbcParameterValuesProvider(
- new JdbcGenericParameterValuesProvider(
- new Serializable[][] {{1001, 1005}, {1006,
1010}}))
- ...
- .build();
-env.fromSource(jdbcSource, WatermarkStrategy.noWatermarks(), "TestSource")
- .addSink(new DiscardSink());
-env.execute();
-
-```
-
-### Minimalist Streaming Semantic and ContinuousUnBoundingSettings
-
-If you want to generate continuous milliseconds parameters based on
sliding-window,
-please have a try on setting the followed attributes of `JdbcSource`:
-
-```java
-
-jdbcSourceBuilder =
- JdbcSource.<TestEntry>builder()
- .setSql("select * from testing_table where ts >= ? and ts < ?")
-
- // Required for streaming related semantic.
- .setContinuousUnBoundingSettings(new
ContinuousUnBoundingSettings(Duration.ofMillis(10L), Duration.ofSeconds(1L)))
- .setJdbcParameterValuesProvider(new
JdbcSlideTimingParameterProvider(0L, 1000L, 1000L, 100L))
- .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
- .setResultSetType(ResultSet.TYPE_SCROLL_INSENSITIVE);
-
- // other attributes
- ...
-
-JdbcSource source = jdbcSourceBuilder.build();
```
### Full example
-{{< tabs "4ab65f13-608a-411a-8d24-e303f348ds81" >}}
-{{< tab "Java" >}}
-
```java
+import org.apache.flink.streaming.api.functions.sink.v2.DiscardingSink;
public class JdbcSourceExample {
@@ -198,11 +206,11 @@ public class JdbcSourceExample {
JdbcSource<Book> jdbcSource =
JdbcSource.<Book>builder()
.setTypeInformation(TypeInformation.of(Book.class))
- .setSql("select * from testing_table where id < ?")
- .setDBUrl(...)
- .setJdbcParameterValuesProvider(
- new JdbcGenericParameterValuesProvider(
+ .setSplitter(
+ PreparedSplitterEnumerator.of(
+ "select * from books where id < ?",
new Serializable[][] {{1001L}}))
+ .setDBUrl(...)
.setDriverName(...)
.setResultExtractor(resultSet ->
new Book(
@@ -210,47 +218,76 @@ public class JdbcSourceExample {
resultSet.getString("title")))
.build();
env.fromSource(jdbcSource, WatermarkStrategy.noWatermarks(),
"TestSource")
- .addSink(new DiscardingSink());
+ .sinkTo(new DiscardingSink<>());
env.execute();
}
}
```
-{{< /tab >}}
-{{< tab "Python" >}}
-```python
-Still not supported in Python API.
+
+### Deprecated query API
+
+Before `SplitterEnumerator` was introduced, the query was defined with
`setSql` and the splits with a
+`JdbcParameterValuesProvider`. Both methods are deprecated.
`setContinuousUnBoundingSettings` and
+`setOptionalSqlSplitEnumeratorState` are only read on this path. A bounded
read maps directly to a
+`PreparedSplitterEnumerator`:
+
+```java
+// deprecated
+JdbcSource.<TestEntry>builder()
+ .setSql("select * from testing_table where id >= ? and id <= ?")
+ .setJdbcParameterValuesProvider(
+ new JdbcGenericParameterValuesProvider(
+ new Serializable[][] {{1001, 1005}, {1006, 1010}}))
+ ...
+
+// current
+JdbcSource.<TestEntry>builder()
+ .setSplitter(
+ PreparedSplitterEnumerator.of(
+ "select * from testing_table where id >= ? and id <=
?",
+ new Serializable[][] {{1001, 1005}, {1006, 1010}}))
+ ...
+```
+
+On this deprecated path, continuous unbounded reads are configured with
+`setContinuousUnBoundingSettings` and a `JdbcSlideTimingParameterProvider`,
which must be set
+together:
+
+```java
+// deprecated
+JdbcSource.<TestEntry>builder()
+ .setSql("select * from testing_table where ts >= ? and ts < ?")
+ .setContinuousUnBoundingSettings(
+ new ContinuousUnBoundingSettings(Duration.ofMillis(10L),
Duration.ofSeconds(1L)))
+ .setJdbcParameterValuesProvider(
+ new JdbcSlideTimingParameterProvider(startMillis, 1000L,
1000L, 100L))
+ .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
+ .setResultSetType(ResultSet.TYPE_SCROLL_INSENSITIVE);
```
-{{< /tab >}}
-{{< /tabs >}}
-## `JdbcSink.sink`
+The closest replacement is
[`SlideTimingSplitterEnumerator`](#slidetimingsplitterenumerator), but it
+is not like-for-like: it has no initial discovery delay and no discovery
interval. A window is emitted
+as a split as soon as it has closed and `splitGenerateDelayMillis` has passed.
+`setContinuousUnBoundingSettings` has no effect when a splitter is set.
+
+## JDBC Sink
-The JDBC sink provides at-least-once guarantee.
-Effectively though, exactly-once can be achieved by crafting upsert SQL
statements or idempotent SQL updates.
-Configuration goes as follows:
+The JDBC sink is built with `JdbcSink.builder()`. It provides an at-least-once
and an exactly-once
+variant, selected by which `build` method you call. Effectively though,
exactly-once can also be
+achieved on the at-least-once path by crafting upsert SQL statements or
idempotent SQL updates.
-{{< tabs "4ab65f13-607a-411a-8d24-e709f701cd4c" >}}
-{{< tab "Java" >}}
```java
-JdbcSink.sink(
- sqlDmlStatement, // mandatory
- jdbcStatementBuilder, // mandatory
- jdbcExecutionOptions, // optional
- jdbcConnectionOptions // mandatory
-);
+JdbcSink<IN> sink = JdbcSink.<IN>builder()
+ .withQueryStatement(sqlDmlStatement, jdbcStatementBuilder) //
mandatory
+ .withExecutionOptions(jdbcExecutionOptions) //
optional
+ .buildAtLeastOnce(jdbcConnectionOptions); //
mandatory
```
-{{< /tab >}}
-{{< tab "Python" >}}
-```python
-JdbcSink.sink(
- sql_dml_statement, # mandatory
- type_info, # mandatory
- jdbc_connection_options, # mandatory
- jdbc_execution_options # optional
-)
+
+The sink is attached to a stream with `sinkTo`:
+
+```java
+stream.sinkTo(sink);
```
-{{< /tab >}}
-{{< /tabs >}}
### SQL DML statement and JDBC statement builder
@@ -266,12 +303,14 @@ It then repeatedly calls a user-provided function to
update that prepared statem
(preparedStatement, someRecord) -> { ... update here the preparedStatement
with values from someRecord ... }
```
+The two are passed together to `withQueryStatement`. Alternatively, pass a
`JdbcQueryStatement`,
+which bundles both in one class. Either way the query is fixed: the writer
prepares it once, so it
+cannot vary per record.
+
### JDBC execution options
The SQL DML statements are executed in batches, which can optionally be
configured with the following instance:
-{{< tabs "4ab65f13-607a-411a-8d24-e709f512ed6k" >}}
-{{< tab "Java" >}}
```java
JdbcExecutionOptions.builder()
.withBatchIntervalMs(200) // optional: default = 0,
meaning no time-based execution is done
@@ -279,17 +318,6 @@ JdbcExecutionOptions.builder()
.withMaxRetries(5) // optional: default = 3
.build();
```
-{{< /tab >}}
-{{< tab "Python" >}}
-```python
-JdbcExecutionOptions.builder() \
- .with_batch_interval_ms(2000) \
- .with_batch_size(100) \
- .with_max_retries(5) \
- .build()
-```
-{{< /tab >}}
-{{< /tabs >}}
A JDBC batch is executed as soon as one of the following conditions is true:
@@ -299,12 +327,12 @@ A JDBC batch is executed as soon as one of the following
conditions is true:
### JDBC connection parameters
-The connection to the database is configured with a `JdbcConnectionOptions`
instance.
+The connection to the database is configured with a `JdbcConnectionOptions`
instance, which is passed
+to `buildAtLeastOnce`. A `JdbcConnectionProvider` can be passed instead to
reuse an existing
+connection strategy.
### Full example
-{{< tabs "4ab65f13-608a-411a-8d24-e303f348ds8d" >}}
-{{< tab "Java" >}}
```java
public class JdbcSinkExample {
@@ -324,137 +352,94 @@ public class JdbcSinkExample {
public static void main(String[] args) throws Exception {
var env = StreamExecutionEnvironment.getExecutionEnvironment();
- env.fromElements(
+ env.fromData(
new Book(101L, "Stream Processing with Apache Flink", "Fabian
Hueske, Vasiliki Kalavri", 2019),
new Book(102L, "Streaming Systems", "Tyler Akidau, Slava
Chernyak, Reuven Lax", 2018),
new Book(103L, "Designing Data-Intensive Applications",
"Martin Kleppmann", 2017),
new Book(104L, "Kafka: The Definitive Guide", "Gwen Shapira,
Neha Narkhede, Todd Palino", 2017)
- ).addSink(
- JdbcSink.sink(
- "insert into books (id, title, authors, year) values
(?, ?, ?, ?)",
- (statement, book) -> {
- statement.setLong(1, book.id);
- statement.setString(2, book.title);
- statement.setString(3, book.authors);
- statement.setInt(4, book.year);
- },
- JdbcExecutionOptions.builder()
- .withBatchSize(1000)
- .withBatchIntervalMs(200)
- .withMaxRetries(5)
- .build(),
- new
JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
-
.withUrl("jdbc:postgresql://dbhost:5432/postgresdb")
- .withDriverName("org.postgresql.Driver")
- .withUsername("someUser")
- .withPassword("somePassword")
- .build()
- ));
-
+ ).sinkTo(
+ JdbcSink.<Book>builder()
+ .withQueryStatement(
+ "insert into books (id, title, authors, year)
values (?, ?, ?, ?)",
+ (statement, book) -> {
+ statement.setLong(1, book.id);
+ statement.setString(2, book.title);
+ statement.setString(3, book.authors);
+ statement.setInt(4, book.year);
+ })
+ .withExecutionOptions(
+ JdbcExecutionOptions.builder()
+ .withBatchSize(1000)
+ .withBatchIntervalMs(200)
+ .withMaxRetries(5)
+ .build())
+ .buildAtLeastOnce(
+ new
JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
+
.withUrl("jdbc:postgresql://dbhost:5432/postgresdb")
+
.withDriverName("org.postgresql.Driver")
+ .withUsername("someUser")
+ .withPassword("somePassword")
+ .build()));
+
env.execute();
}
}
```
-{{< /tab >}}
-{{< tab "Python" >}}
-```python
-env = StreamExecutionEnvironment.get_execution_environment()
-type_info = Types.ROW([Types.INT(), Types.STRING(), Types.STRING(),
Types.INT()])
-env.from_collection(
- [(101, "Stream Processing with Apache Flink", "Fabian Hueske, Vasiliki
Kalavri", 2019),
- (102, "Streaming Systems", "Tyler Akidau, Slava Chernyak, Reuven Lax",
2018),
- (103, "Designing Data-Intensive Applications", "Martin Kleppmann", 2017),
- (104, "Kafka: The Definitive Guide", "Gwen Shapira, Neha Narkhede, Todd
Palino", 2017)
- ], type_info=type_info) \
- .add_sink(
- JdbcSink.sink(
- "insert into books (id, title, authors, year) values (?, ?, ?, ?)",
- type_info,
- JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
- .with_url('jdbc:postgresql://dbhost:5432/postgresdb')
- .with_driver_name('org.postgresql.Driver')
- .with_user_name('someUser')
- .with_password('somePassword')
- .build(),
- JdbcExecutionOptions.builder()
- .with_batch_interval_ms(1000)
- .with_batch_size(200)
- .with_max_retries(5)
- .build()
- ))
-
-env.execute()
-```
-{{< /tab >}}
-{{< /tabs >}}
-## `JdbcSink.exactlyOnceSink`
+### Exactly-once sink
-Since 1.13, Flink JDBC sink supports exactly-once mode.
-The implementation relies on the JDBC driver support of XA
+The exactly-once implementation relies on the JDBC driver support of XA
[standard](https://pubs.opengroup.org/onlinepubs/009680699/toc.pdf).
Most drivers support XA if the database also supports XA (so the driver is
usually the same).
-To use it, create a sink using `exactlyOnceSink()` method as above and
additionally provide:
+To use it, call `buildExactlyOnce()` instead of `buildAtLeastOnce()` and
provide:
- `JdbcExactlyOnceOptions`
-- `JdbcExecutionOptions`
-- [XA
DataSource](https://docs.oracle.com/javase/8/docs/api/javax/sql/XADataSource.html)
Supplier
+- an [XA
DataSource](https://docs.oracle.com/javase/8/docs/api/javax/sql/XADataSource.html)
Supplier
For example:
-{{< tabs "4ab65f13-608a-411a-8d24-e304f627ac8f" >}}
-{{< tab "Java" >}}
```java
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env
- .fromElements(...)
- .addSink(JdbcSink.exactlyOnceSink(
- "insert into books (id, title, author, price, qty) values
(?,?,?,?,?)",
- (ps, t) -> {
- ps.setInt(1, t.id);
- ps.setString(2, t.title);
- ps.setString(3, t.author);
- ps.setDouble(4, t.price);
- ps.setInt(5, t.qty);
- },
- JdbcExecutionOptions.builder()
- .withMaxRetries(0)
- .build(),
- JdbcExactlyOnceOptions.defaults(),
- () -> {
- // create a driver-specific XA DataSource
- // The following example is for derby
- EmbeddedXADataSource ds = new EmbeddedXADataSource();
- ds.setDatabaseName("my_db");
- return ds;
- });
+ .fromData(...)
+ .sinkTo(JdbcSink.<Book>builder()
+ .withQueryStatement(
+ "insert into books (id, title, author, price, qty)
values (?,?,?,?,?)",
+ (ps, t) -> {
+ ps.setInt(1, t.id);
+ ps.setString(2, t.title);
+ ps.setString(3, t.author);
+ ps.setDouble(4, t.price);
+ ps.setInt(5, t.qty);
+ })
+ .withExecutionOptions(
+ JdbcExecutionOptions.builder()
+ .withMaxRetries(0)
+ .build())
+ .buildExactlyOnce(
+ JdbcExactlyOnceOptions.defaults(),
+ () -> {
+ // create a driver-specific XA DataSource
+ PGXADataSource ds = new PGXADataSource();
+
ds.setUrl("jdbc:postgresql://localhost:5432/postgres");
+ ds.setUser(username);
+ ds.setPassword(password);
+ return ds;
+ }));
env.execute();
```
-{{< /tab >}}
-{{< tab "Python" >}}
-```python
-Still not supported in Python API.
-```
-{{< /tab >}}
-{{< /tabs >}}
+
+An `XaConnectionProvider` can be passed instead of the supplier when the
connection strategy has to
+be controlled directly.
**NOTE:** Some databases only allow a single XA transaction per connection
(e.g. PostgreSQL, MySQL).
In such cases, please use the following API to construct
`JdbcExactlyOnceOptions`:
-{{< tabs "4ab65f13-608a-411a-8d24-e304f627cd4e" >}}
-{{< tab "Java" >}}
```java
JdbcExactlyOnceOptions.builder()
.withTransactionPerConnection(true)
.build();
```
-{{< /tab >}}
-{{< tab "Python" >}}
-```python
-Still not supported in Python API.
-```
-{{< /tab >}}
-{{< /tabs >}}
This will make Flink use a separate connection for every XA transaction. This
may require adjusting connection limits.
For PostgreSQL and MySQL, this can be done by increasing `max_connections`.
@@ -463,63 +448,48 @@ Furthermore, XA needs to be enabled and/or configured in
some databases.
For PostgreSQL, you should set `max_prepared_transactions` to some value
greater than zero.
For MySQL v8+, you should grant `XA_RECOVER_ADMIN` to Flink DB user.
-**ATTENTION:** Currently, `JdbcSink.exactlyOnceSink` can ensure exactly once
semantics
-with `JdbcExecutionOptions.maxRetries == 0`; otherwise, duplicated results
maybe produced.
+**ATTENTION:** The exactly-once sink requires
`JdbcExecutionOptions.maxRetries` to be set to `0`.
+The default is `3`, so set it explicitly with
`JdbcExecutionOptions.builder().withMaxRetries(0)`;
+otherwise the sink fails at startup.
### `XADataSource` examples
PostgreSQL `XADataSource` example:
-{{< tabs "4ab65f13-608a-411a-8d24-e304f323ab3a" >}}
-{{< tab "Java" >}}
```java
PGXADataSource xaDataSource = new org.postgresql.xa.PGXADataSource();
xaDataSource.setUrl("jdbc:postgresql://localhost:5432/postgres");
xaDataSource.setUser(username);
xaDataSource.setPassword(password);
```
-{{< /tab >}}
-{{< tab "Python" >}}
-```python
-Still not supported in Python API.
-```
-{{< /tab >}}
-{{< /tabs >}}
MySQL `XADataSource` example:
-{{< tabs "4ab65f13-608a-411a-8d24-b213f323ca3c" >}}
-{{< tab "Java" >}}
```java
MysqlXADataSource xaDataSource = new com.mysql.cj.jdbc.MysqlXADataSource();
xaDataSource.setUrl("jdbc:mysql://localhost:3306/");
xaDataSource.setUser(username);
xaDataSource.setPassword(password);
```
-{{< /tab >}}
-{{< tab "Python" >}}
-```python
-Still not supported in Python API.
-```
-{{< /tab >}}
-{{< /tabs >}}
Oracle `XADataSource` example:
-{{< tabs "4ab65f13-608a-411a-8d24-b213f337ad5f" >}}
-{{< tab "Java" >}}
```java
OracleXADataSource xaDataSource = new oracle.jdbc.xa.OracleXADataSource();
xaDataSource.setURL("jdbc:oracle:oci8:@");
xaDataSource.setUser("scott");
xaDataSource.setPassword("tiger");
```
-{{< /tab >}}
-{{< tab "Python" >}}
-```python
-Still not supported in Python API.
-```
-{{< /tab >}}
-{{< /tabs >}}
Please also take Oracle connection pooling into account.
-Please refer to the `JdbcXaSinkFunction` documentation for more details.
+## Lineage
+
+`JdbcSource` and `JdbcSink` expose lineage information as described in
+[FLIP-314](https://cwiki.apache.org/confluence/display/FLINK/FLIP-314%3A+Support+Customized+Job+Lineage+Listener),
+as do the Table API source and lookup function. No configuration is needed on
the connector side; the
+information is delivered to any `JobStatusChangedListener` registered in the
cluster.
+
+The dataset name is derived from the queries the job executes. The dataset
namespace is derived
+from the JDBC URL, but only when the connector opens connections from that URL
itself, i.e. a
+`JdbcSource` configured with `setDBUrl` or a `JdbcSink` built with
+`buildAtLeastOnce(JdbcConnectionOptions)`. With `buildExactlyOnce` or a custom
+`JdbcConnectionProvider` the namespace is empty.
{{< top >}}
diff --git a/docs/content/docs/connectors/table/jdbc.md
b/docs/content/docs/connectors/table/jdbc.md
index 1d7ec9d8..1db9e29a 100644
--- a/docs/content/docs/connectors/table/jdbc.md
+++ b/docs/content/docs/connectors/table/jdbc.md
@@ -38,27 +38,37 @@ The JDBC sink operate in upsert mode for exchange
UPDATE/DELETE messages with th
Dependencies
------------
-When using the JDBC connector, `flink-connector-jdbc-core`,
`flink-connector-jdbc-${database}` and the JDBC driver needs to be added to the
dependencies.
+Since version 4.0 the JDBC connector is no longer published as a single
artifact. It is split into
+`flink-connector-jdbc-core`, which contains the shared runtime, and one
artifact per supported
+database, which contains the dialect and the catalog for that database. You
need the artifact of the
+database you are connecting to; it pulls in `flink-connector-jdbc-core`
transitively.
+
+| Database | Connector artifact |
+|:-----------|:----------------------------------|
+| MySQL | `flink-connector-jdbc-mysql` |
+| Oracle | `flink-connector-jdbc-oracle` |
+| PostgreSQL | `flink-connector-jdbc-postgres` |
+| SQL Server | `flink-connector-jdbc-sqlserver` |
+| CrateDB | `flink-connector-jdbc-cratedb` |
+| Db2 | `flink-connector-jdbc-db2` |
+| Trino | `flink-connector-jdbc-trino` |
+| OceanBase | `flink-connector-jdbc-oceanbase` |
+| Derby | `flink-connector-jdbc-core` |
{{< sql_connector_download_table "jdbc" >}}
-The `flink-connector-jdbc-${database}` dependency and a driver dependency are
required to connect to a specified database.
+None of these artifacts is a self-contained SQL uber jar. When using the SQL
Client or a session
+cluster, put both `flink-connector-jdbc-core` and the artifact for your
database into the `lib/`
+directory, together with the `io.openlineage:openlineage-java` and
`io.openlineage:openlineage-sql-java`
+jars that `flink-connector-jdbc-core` depends on. Without them, building the
job fails with a
+`NoClassDefFoundError`. Some dialects build on others and need their artifacts
as well: CrateDB also needs
+`flink-connector-jdbc-postgres`, and OceanBase needs both
`flink-connector-jdbc-mysql` and
+`flink-connector-jdbc-oracle`.
-The `flink-connector-jdbc-${database}` dependencies are as follows, whose
version should be the same as the `flink-connector-jdbc-core`:
+The JDBC connector is not part of the binary distribution.
+See how to link with it for cluster execution [here]({{< ref
"docs/dev/configuration/overview" >}}).
-| Database | Group Id | Artifact Id |
JAR
|
-|:-----------|:--------------------------|:---------------------------------|:--------------------------------------------------------------------------------------------------|
-| MySQL | `org.apache.flink` | `flink-connector-jdbc-mysql` |
[Download](https://repo.maven.apache.org/maven2/org/apache/flink/flink-connector-jdbc-mysql/)
|
-| Oracle | `org.apache.flink` | `flink-connector-jdbc-oracle` |
[Download](https://repo.maven.apache.org/maven2/org/apache/flink/flink-connector-jdbc-oracle/)
|
-| PostgreSQL | `org.apache.flink` | `flink-connector-jdbc-postgres` |
[Download](https://repo.maven.apache.org/maven2/org/apache/flink/flink-connector-jdbc-postgres/)
|
-| Derby | `org.apache.flink` | `flink-connector-jdbc-core` |
[Download](https://repo.maven.apache.org/maven2/org/apache/flink/flink-connector-jdbc-core/)
|
-| SQL Server | `org.apache.flink` | `flink-connector-jdbc-sqlserver` |
[Download](https://repo.maven.apache.org/maven2/org/apache/flink/flink-connector-jdbc-sqlserver/)
|
-| CrateDB | `org.apache.flink` | `flink-connector-jdbc-cratedb` |
[Download](https://repo.maven.apache.org/maven2/org/apache/flink/flink-connector-jdbc-cratedb/)
|
-| Db2 | `org.apache.flink` | `flink-connector-jdbc-db2` |
[Download](https://repo.maven.apache.org/maven2/org/apache/flink/flink-connector-jdbc-db2/)
|
-| Trino | `org.apache.flink` | `flink-connector-jdbc-trino` |
[Download](https://repo.maven.apache.org/maven2/org/apache/flink/flink-connector-jdbc-trino/)
|
-| OceanBase | `org.apache.flink` | `flink-connector-jdbc-oceanbase` |
[Download](https://repo.maven.apache.org/maven2/org/apache/flink/flink-connector-jdbc-oceanbase/)
|
-
-Here are drivers currently supported:
+A driver dependency is also required to connect to a specified database. Here
are drivers currently supported:
| Driver | Group Id | Artifact Id | JAR
|
|:-----------|:---------------------------|:-----------------------|:----------------------------------------------------------------------------------------------------------------------------------|
@@ -73,8 +83,7 @@ Here are drivers currently supported:
| OceanBase | `com.oceanbase` | `oceanbase-client` |
[Download](https://repo1.maven.org/maven2/com/oceanbase/oceanbase-client/)
|
-JDBC connector and drivers are not part of Flink's binary distribution.
-See how to link with them for cluster execution [here]({{< ref
"docs/dev/configuration/overview" >}}).
+JDBC connector and drivers are not part of Flink's binary distribution. See
how to link with them for cluster execution [here]({{< ref
"docs/dev/configuration/overview" >}}).
How to create a JDBC table
@@ -162,7 +171,10 @@ Connector Options
<td>yes</td>
<td style="word-wrap: break-word;">(none)</td>
<td>String</td>
- <td>The compatible mode of database.</td>
+ <td>The compatible mode of database. Only OceanBase supports this
option, with the values
+ <code>'mysql'</code> (the default when unset) and
<code>'oracle'</code>, selecting which
+ tenant mode the dialect targets. Setting it for any other database
fails with
+ <code>Not supported option 'compatible-mode'</code>.</td>
</tr>
<tr>
<td><h5>username</h5></td>
@@ -209,7 +221,7 @@ Connector Options
<td>optional</td>
<td>no</td>
<td style="word-wrap: break-word;">(none)</td>
- <td>Integer</td>
+ <td>Long</td>
<td>The smallest value of the first partition.</td>
</tr>
<tr>
@@ -217,7 +229,7 @@ Connector Options
<td>optional</td>
<td>no</td>
<td style="word-wrap: break-word;">(none)</td>
- <td>Integer</td>
+ <td>Long</td>
<td>The largest value of the last partition.</td>
</tr>
<tr>
@@ -463,9 +475,23 @@ As there is no standard syntax for upsert, the following
table describes the dat
WHEN NOT MATCHED THEN INSERT (..) <br>
VALUES (..)</td>
</tr>
+ <tr>
+ <td>CrateDB</td>
+ <td>INSERT .. ON CONFLICT .. DO UPDATE SET ..</td>
+ </tr>
+ <tr>
+ <td>OceanBase</td>
+ <td>Depends on <code>'compatible-mode'</code>: the MySQL grammar
in <code>'mysql'</code>
+ mode, the Oracle grammar in <code>'oracle'</code> mode.</td>
+ </tr>
</tbody>
</table>
+Trino and Derby have no upsert grammar. If a primary key is defined in the
DDL, the sink instead
+checks whether a row with that key exists and then issues an `UPDATE` or an
`INSERT`, and issues a
+`DELETE` for retractions. For Trino this only works if the Trino connector
behind the table supports
+`UPDATE` and `DELETE`.
+
JDBC Catalog
------------
@@ -491,14 +517,41 @@ Please refer to [Dependencies](#dependencies) section for
how to setup a JDBC co
The JDBC catalog supports the following options:
- `name`: required, name of the catalog.
-- `default-database`: required, default database to connect to.
- `username`: required, username of database account.
- `password`: required, password of the account.
-- `base-url`: required (should not contain the database name)
+- `base-url`: required
- for Postgres Catalog this should be `"jdbc:postgresql://<ip>:<port>"`
- for MySQL Catalog this should be `"jdbc:mysql://<ip>:<port>"`
- - for OceanBase Catalog this should be `jdbc:oceanbase://<ip>:<port>`
-- `compatible-mode`: optional, the compatible mode of database.
+ - for OceanBase Catalog this should be `"jdbc:oceanbase://<ip>:<port>"`
+- `default-database`: optional, default database to connect to.
+- `compatible-mode`: optional, the compatible mode of database. Only OceanBase
supports this option,
+ see [Connector Options](#connector-options).
+
+The database the catalog connects to has to be identifiable from `base-url`,
`default-database`, or
+both. Any one of the following is valid:
+
+- `default-database` is set and `base-url` carries no database name, for
example
+ `'base-url' = 'jdbc:postgresql://localhost:5432'` with `'default-database' =
'mydb'`
+- `base-url` carries the database name and `default-database` is omitted, for
example
+ `'base-url' = 'jdbc:postgresql://localhost:5432/mydb'`. Without a query
string, this form pins
+ the catalog to `mydb`: every database in the catalog is accessed through
this same URL.
+- both are set and the database names match
+
+If the two disagree, or if neither carries a database name, catalog creation
fails.
+
+`base-url` may also carry arbitrary driver options as query parameters, which
are passed through to
+every connection the catalog opens. The database name is substituted ahead of
the query string, so
+the options apply to all databases in the catalog. When `base-url` has a query
string, it must
+carry the database name itself; setting only `default-database` is rejected:
+
+```sql
+CREATE CATALOG my_catalog WITH(
+ 'type' = 'jdbc',
+ 'username' = '...',
+ 'password' = '...',
+ 'base-url' =
'jdbc:postgresql://localhost:5432/mydb?ssl=true&connectTimeout=10'
+);
+```
{{< tabs "10bd8bfb-674c-46aa-8a36-385537df5791" >}}
{{< tab "SQL" >}}
@@ -972,7 +1025,9 @@ Flink supports connect to several databases which uses
dialect like MySQL, Oracl
<code>CHARACTER(n)</code><br>
<code>VARCHAR(n)</code><br>
<code>CHARACTER VARYING(n)</code><br>
- <code>TEXT</code></td>
+ <code>TEXT</code><br>
+ <code>JSON</code><br>
+ <code>JSONB</code></td>
<td>
<code>CHAR(n)</code><br>
<code>CHARACTER(n)</code><br>
@@ -1031,6 +1086,18 @@ Flink supports connect to several databases which uses
dialect like MySQL, Oracl
<code>BLOB</code></td>
<td><code>BYTES</code></td>
</tr>
+ <tr>
+ <td></td>
+ <td></td>
+ <td><code>UUID</code></td>
+ <td></td>
+ <td></td>
+ <td></td>
+ <td></td>
+ <td></td>
+ <td></td>
+ <td><code>VARCHAR(36)</code></td>
+ </tr>
<tr>
<td></td>
<td></td>
@@ -1046,4 +1113,10 @@ Flink supports connect to several databases which uses
dialect like MySQL, Oracl
</tbody>
</table>
+PostgreSQL `JSON` and `JSONB` columns are read as their textual
representation. `UUID` columns are
+mapped to `VARCHAR(36)`; declaring such a column as `STRING` in the DDL also
works. The sink binds
+all three as strings, which PostgreSQL rejects for these column types unless
the JDBC URL sets
+`stringtype=unspecified`, for example
+`'url' = 'jdbc:postgresql://localhost:5432/mydb?stringtype=unspecified'`.
+
{{< top >}}
diff --git a/docs/data/jdbc.yml b/docs/data/jdbc.yml
index f479cf29..23faa920 100644
--- a/docs/data/jdbc.yml
+++ b/docs/data/jdbc.yml
@@ -21,3 +21,19 @@ flink_compatibility: ["2.1", "2.2"]
variants:
- maven: flink-connector-jdbc-core
sql_url:
https://repo.maven.apache.org/maven2/org/apache/flink/flink-connector-jdbc-core/$full_version/flink-connector-jdbc-core-$full_version.jar
+ - maven: flink-connector-jdbc-cratedb
+ sql_url:
https://repo.maven.apache.org/maven2/org/apache/flink/flink-connector-jdbc-cratedb/$full_version/flink-connector-jdbc-cratedb-$full_version.jar
+ - maven: flink-connector-jdbc-db2
+ sql_url:
https://repo.maven.apache.org/maven2/org/apache/flink/flink-connector-jdbc-db2/$full_version/flink-connector-jdbc-db2-$full_version.jar
+ - maven: flink-connector-jdbc-mysql
+ sql_url:
https://repo.maven.apache.org/maven2/org/apache/flink/flink-connector-jdbc-mysql/$full_version/flink-connector-jdbc-mysql-$full_version.jar
+ - maven: flink-connector-jdbc-oceanbase
+ sql_url:
https://repo.maven.apache.org/maven2/org/apache/flink/flink-connector-jdbc-oceanbase/$full_version/flink-connector-jdbc-oceanbase-$full_version.jar
+ - maven: flink-connector-jdbc-oracle
+ sql_url:
https://repo.maven.apache.org/maven2/org/apache/flink/flink-connector-jdbc-oracle/$full_version/flink-connector-jdbc-oracle-$full_version.jar
+ - maven: flink-connector-jdbc-postgres
+ sql_url:
https://repo.maven.apache.org/maven2/org/apache/flink/flink-connector-jdbc-postgres/$full_version/flink-connector-jdbc-postgres-$full_version.jar
+ - maven: flink-connector-jdbc-sqlserver
+ sql_url:
https://repo.maven.apache.org/maven2/org/apache/flink/flink-connector-jdbc-sqlserver/$full_version/flink-connector-jdbc-sqlserver-$full_version.jar
+ - maven: flink-connector-jdbc-trino
+ sql_url:
https://repo.maven.apache.org/maven2/org/apache/flink/flink-connector-jdbc-trino/$full_version/flink-connector-jdbc-trino-$full_version.jar
