This is an automated email from the ASF dual-hosted git repository.
davidzollo pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new ac7a5d9433 [Docs] Improve JDBC sink guide with verified quick start
(#11483)
ac7a5d9433 is described below
commit ac7a5d9433230c58f72ec7e58240e06fec44069e
Author: Daniel <[email protected]>
AuthorDate: Sun Jul 19 12:42:16 2026 +0800
[Docs] Improve JDBC sink guide with verified quick start (#11483)
Co-authored-by: DanielLeens <[email protected]>
---
docs/en/connectors/sink/Jdbc.md | 276 ++++++++++++-----
docs/zh/connectors/sink/Jdbc.md | 345 ++++++++++++++++-----
.../seatunnel/jdbc/JdbcAutoGenerateSQLIT.java | 52 ++++
.../resources/jdbc_sink_postgres_quick_start.conf | 54 ++++
4 files changed, 582 insertions(+), 145 deletions(-)
diff --git a/docs/en/connectors/sink/Jdbc.md b/docs/en/connectors/sink/Jdbc.md
index efd007b6f5..20f40d4455 100644
--- a/docs/en/connectors/sink/Jdbc.md
+++ b/docs/en/connectors/sink/Jdbc.md
@@ -2,41 +2,158 @@ import ChangeLog from '../changelog/connector-jdbc.md';
# JDBC
-> JDBC sink connector
-
## Description
-Write data through jdbc. Support Batch mode and Streaming mode, support
concurrent writing, support exactly-once
-semantics (using XA transaction guarantee).
+The JDBC Sink connector writes SeaTunnel rows to databases through a JDBC
driver. It supports batch and streaming jobs, parallel writers, generated or
custom SQL, multi-table writes, CDC events, and optional exactly-once delivery
through XA transactions.
+
+If this is your first JDBC Sink job, start with [Choose a write
mode](#choose-a-write-mode) and the [Quick start](#quick-start-postgresql). The
complete option reference follows those sections.
## Using Dependency
-### For Spark/Flink Engine
+Install the `connector-jdbc` plugin first:
+
+```plugin_config
+--seatunnel-connectors--
+connector-jdbc
+--end--
+```
+
+```bash
+cd "${SEATUNNEL_HOME}"
+sh bin/install-plugin.sh
+```
+
+JDBC driver licenses and redistribution terms vary by database vendor, and the
driver version must also be compatible with your database and Java runtime.
SeaTunnel therefore does not bundle every JDBC driver. Download the appropriate
driver yourself and place its JAR in the engine-specific directory below before
starting the job.
+
+### Spark and Flink engines
+
+Place the JDBC driver in `${SEATUNNEL_HOME}/plugins/Jdbc/lib/` on every node
that runs SeaTunnel.
+
+### Zeta engine
+
+Place the JDBC driver in `${SEATUNNEL_HOME}/lib/` on every SeaTunnel node,
then restart the affected SeaTunnel processes so the driver is loaded.
+
+See the [driver reference](#driver-reference) for common driver class names
and download locations.
+
+## Choose a write mode
+
+JDBC Sink has two mutually exclusive write modes. Choose one before
configuring the remaining options.
+
+| Use case | Required configuration | Behavior |
+|----------|------------------------|----------|
+| SeaTunnel generates SQL | `generate_sink_sql = true`, `database`, and
normally `table` | Recommended for most jobs. SeaTunnel can generate INSERT,
database-native UPSERT, UPDATE, and DELETE statements from the upstream schema
and row kind. Save modes and automatic table creation are available. |
+| You provide SQL | `query = "INSERT ... VALUES (?, ...)"` | Use when the
target statement must be fully controlled. The `?` parameters are bound in
upstream field order. Save mode options are not executed in this mode. |
+
+Do not configure both modes. `generate_sink_sql` defaults to `false`, so a job
without `generate_sink_sql = true` must provide `query`.
+
+For generated SQL, configure `primary_keys` when the target needs upsert,
update, or delete behavior. If it is omitted, SeaTunnel tries to inherit a
primary key, then the first unique key, from upstream catalog metadata. Without
any usable key, generated SQL falls back to plain INSERT.
+
+## Quick start: PostgreSQL
+
+This beginner example uses generated SQL and a pre-created target table. The
configuration and expected result below have been verified against PostgreSQL
14, including the inserted rows and their final database values.
+
+1. Put a compatible PostgreSQL JDBC driver in the directory described in
[Using Dependency](#using-dependency).
+
+2. Create the target table:
+
+```sql
+CREATE TABLE public.orders (
+ id BIGINT PRIMARY KEY,
+ customer_name VARCHAR(100) NOT NULL,
+ amount DECIMAL(10, 2) NOT NULL
+);
+```
+
+3. Save the following job as
`${SEATUNNEL_HOME}/config/jdbc-sink-quick-start.conf`. Replace the host,
credentials, and database name for your environment.
+
+```hocon
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
+
+source {
+ FakeSource {
+ row.num = 3
+ schema = {
+ fields {
+ id = bigint
+ customer_name = string
+ amount = "decimal(10, 2)"
+ }
+ }
+ rows = [
+ { kind = INSERT, fields = [1, "Alice", 120.50] }
+ { kind = INSERT, fields = [2, "Bob", 80.00] }
+ { kind = INSERT, fields = [3, "Carol", 42.00] }
+ ]
+ }
+}
+
+sink {
+ Jdbc {
+ url = "jdbc:postgresql://localhost:5432/sales"
+ driver = "org.postgresql.Driver"
+ username = "postgres"
+ password = "change_me"
+ generate_sink_sql = true
+ database = "sales"
+ table = "public.orders"
+ primary_keys = ["id"]
+ schema_save_mode = "ERROR_WHEN_SCHEMA_NOT_EXIST"
+ data_save_mode = "APPEND_DATA"
+ }
+}
+```
+
+4. Run the job:
+
+```bash
+cd "${SEATUNNEL_HOME}"
+./bin/seatunnel.sh --config ./config/jdbc-sink-quick-start.conf -m local
+```
+
+5. Verify the result:
+
+```sql
+SELECT id, customer_name, amount
+FROM public.orders
+ORDER BY id;
+```
-> 1. You need to ensure that the jdbc driver jar package has been placed in
directory `${SEATUNNEL_HOME}/plugins/`.
+Expected rows:
-### For SeaTunnel Zeta Engine
+| id | customer_name | amount |
+|----|---------------|-------:|
+| 1 | Alice | 120.50 |
+| 2 | Bob | 80.00 |
+| 3 | Carol | 42.00 |
-> 1. You need to ensure that the jdbc driver jar package has been placed in
directory `${SEATUNNEL_HOME}/lib/`.
+If the job fails before writing, check [Troubleshooting](#troubleshooting)
first.
## Key Features
- [x] [exactly-once](../../introduction/concepts/connector-v2-features.md)
-Use `Xa transactions` to ensure `exactly-once`. So only support `exactly-once`
for the database which is
-support `Xa transactions`. You can set `is_exactly_once=true` to enable it.
+Exactly-once delivery uses XA transactions and therefore requires XA support
from both the database and its JDBC driver. See [Exactly-once
prerequisites](#exactly-once-prerequisites).
- [x] [cdc](../../introduction/concepts/connector-v2-features.md)
- [x] [support multiple table
write](../../introduction/concepts/connector-v2-features.md)
-- [x] [timer flush](../../introduction/concepts/connector-v2-features.md)
+- [x] [timer flush](../../introduction/concepts/connector-v2-features.md)
(Zeta engine only)
## Options
+`url`, `driver`, `schema_save_mode`, and `data_save_mode` are always required
by the connector option rule. The two save mode options have defaults, so they
can normally be omitted from job files. Other options become required according
to the selected write mode:
+
+- Generated SQL: set `generate_sink_sql = true` and configure `database`;
configure `table` unless upstream metadata supplies the target table
dynamically.
+- Custom SQL: leave `generate_sink_sql = false` and configure `query`.
+- Exactly-once: set `is_exactly_once = true`, `xa_data_source_class_name`, and
`max_retries = 0`.
+
| Name | Type | Required | Default
|
|-------------------------------------------|---------|----------|------------------------------|
| url | String | Yes | -
|
| driver | String | Yes | -
|
-| username | String | No | -
|
+| username | String | No | -
|
| password | String | No | -
|
| query | String | No | -
|
| compatible_mode | String | No | -
|
@@ -83,9 +200,9 @@ support `Xa transactions`. You can set
`is_exactly_once=true` to enable it.
The jdbc class name used to connect to the remote data source, if you use
MySQL the value is `com.mysql.cj.jdbc.Driver`.
-### user [string]
+### username [string]
-userName
+The database login name. `username` is the canonical option. The legacy key
`user` is still accepted as a fallback when `username` is not set.
### password [string]
@@ -97,7 +214,7 @@ The URL of the JDBC connection. Refer to a case:
jdbc:postgresql://localhost/tes
### query [string]
-Use this sql write upstream input datas to database. e.g `INSERT ...`
+The parameterized SQL statement used to write each upstream row, for example
`INSERT INTO target(id, name) VALUES (?, ?)`. SeaTunnel binds the `?`
parameters in upstream field order. Use this option only in custom SQL mode; do
not combine it with `generate_sink_sql = true`.
Current limitation: when sink `query` is configured (custom write SQL), JDBC
sink does not apply save mode handling. `schema_save_mode`, `data_save_mode`,
and `custom_sql` are not executed in this mode. If you need save mode handling,
use `generate_sink_sql = true` with `database` and `table`.
@@ -131,15 +248,11 @@ If one dialect not supported by SeaTunnel, it will use
the default dialect `Gene
| YashanDB | | |
### database [string]
-Use this `database` and `table-name` auto-generate sql and receive upstream
input datas write to database.
-
-This option is mutually exclusive with `query` and has a higher priority.
+The target database or catalog used in generated SQL mode. This option is
required when `generate_sink_sql = true` and must not be combined with `query`.
### table [string]
-Use `database` and this `table-name` auto-generate sql and receive upstream
input datas write to database.
-
-This option is mutually exclusive with `query` and has a higher priority.
+The target table used in generated SQL mode. Do not combine it with `query`.
The table parameter can fill in the target table name, which will eventually
be used as the created or written table name, and supports variables
(`${table_name}`, `${schema_name}`). Replacement rules: `${schema_name}` will
replace the SCHEMA name passed to the target side, and `${table_name}` will
replace the table name passed to the target side.
@@ -167,7 +280,7 @@ Deprecated. Use `table` with table placeholders instead.
For example, use `table
### primary_keys [array]
-This option is used to support operations such as `insert`, `delete`, and
`update` when automatically generate sql.
+The target key columns used to generate database-native UPSERT, UPDATE, and
DELETE statements. If omitted, SeaTunnel attempts to inherit a primary key or
the first unique key from upstream catalog metadata. If no key is available,
generated SQL uses plain INSERT.
### connection_check_timeout_sec [int]
@@ -183,12 +296,11 @@ Socket read timeout in milliseconds after the JDBC
connection is established. Th
### max_retries [int]
-The number of retries to submit failed (executeBatch)
+The number of retries after a failed JDBC `executeBatch`. Exactly-once mode
requires this option to be `0`; retrying a failed XA batch can violate
transaction guarantees.
### batch_size [int]
-For batch writing, when the number of buffered records reaches the number of
`batch_size` or the time reaches `checkpoint.interval`
-, the data will be flushed into the database
+The maximum number of buffered rows per batch. The sink flushes when the
buffer reaches `batch_size`, when a checkpoint prepares a commit, or when the
writer closes. A larger value can improve throughput but uses more memory and
increases the amount of work retried after a failure.
### batch_interval_ms [long]
@@ -196,12 +308,11 @@ The flush interval in milliseconds. When set to a value
greater than 0, if the e
### is_exactly_once [boolean]
-Whether to enable exactly-once semantics, which will use Xa transactions. If
on, you need to
-set `xa_data_source_class_name`.
+Enables exactly-once delivery through XA transactions. This requires
`xa_data_source_class_name`, `max_retries = 0`, and XA support from the
database and driver. Timer flush is not supported in this mode.
### generate_sink_sql [boolean]
-Generate sql statements based on the database table you want to write to
+When `true`, SeaTunnel generates write statements from the upstream schema and
row kind. Configure `database` and normally `table`; do not configure `query`.
The default is `false`, which means `query` is required.
### xa_data_source_class_name [string]
@@ -380,16 +491,20 @@ The secret_access_key in AWS authentication. Only valid
for dialect="dsql"
### region [String]
The area where Amazon Aurora DSQL is located. Only valid for dialect="dsql"
-## tips
+## Exactly-once prerequisites
+
+When `is_exactly_once = true`, JDBC Sink uses XA transactions. Before enabling
it:
-In the case of is_exactly_once = "true", Xa transactions are used. This
requires database support, and some databases require some setup :
-1 postgres needs to set `max_prepared_transactions > 1` such as `ALTER SYSTEM
set max_prepared_transactions to 10`.
-2 mysql version need >= `8.0.29` and Non-root users need to grant
`XA_RECOVER_ADMIN` permissions. such as `grant XA_RECOVER_ADMIN on test_db.* to
'user1'@'%'`.
-3 mysql can try to add `rewriteBatchedStatements=true` parameter in url for
better performance.
+- Set `max_retries = 0` and configure the correct `xa_data_source_class_name`
for the installed driver.
+- PostgreSQL must allow prepared transactions. Set `max_prepared_transactions`
to a positive value large enough for the expected concurrent transactions, then
restart PostgreSQL if the server requires it.
+- MySQL requires a server and Connector/J combination that supports the XA
operations used by the sink. Accounts that perform XA recovery may also require
the `XA_RECOVER_ADMIN` privilege; check the requirements for your MySQL version.
+- Do not configure `sink.flush.interval`; XA transaction boundaries are
controlled by checkpoints.
-## appendix
+For MySQL non-XA batch jobs, adding `rewriteBatchedStatements=true` to the
JDBC URL can improve throughput. Validate the effect with your driver version
and workload.
-there are some reference value for params above.
+## Driver reference
+
+The following values are starting points. Confirm the driver artifact and
version against the database vendor's compatibility matrix.
| datasource | driver | url
|
xa_data_source_class_name | maven
|
|-------------------|----------------------------------------------|---------------------------------------------------------------------|----------------------------------------------------|-------------------------------------------------------------------------------------------------------------------------------|
@@ -400,7 +515,7 @@ there are some reference value for params above.
| SQL Server | com.microsoft.sqlserver.jdbc.SQLServerDriver |
jdbc:sqlserver://localhost:1433 |
com.microsoft.sqlserver.jdbc.SQLServerXADataSource |
https://mvnrepository.com/artifact/com.microsoft.sqlserver/mssql-jdbc
|
| Oracle | oracle.jdbc.OracleDriver |
jdbc:oracle:thin:@localhost:1521/xepdb1 |
oracle.jdbc.xa.OracleXADataSource |
https://mvnrepository.com/artifact/com.oracle.database.jdbc/ojdbc8
|
| sqlite | org.sqlite.JDBC |
jdbc:sqlite:test.db | /
|
https://mvnrepository.com/artifact/org.xerial/sqlite-jdbc
|
-| GBase8a | com.gbase.jdbc.Driver |
jdbc:gbase://e2e_gbase8aDb:5258/test | /
|
https://cdn.gbase.cn/products/30/p5CiVwXBKQYIUGN8ecHvk/gbase-connector-java-9.5.0.7-build1-bin.jar
|
+| GBase8a | com.gbase.jdbc.Driver |
jdbc:gbase://localhost:5258/test | /
|
https://cdn.gbase.cn/products/30/p5CiVwXBKQYIUGN8ecHvk/gbase-connector-java-9.5.0.7-build1-bin.jar
|
| StarRocks | com.mysql.cj.jdbc.Driver |
jdbc:mysql://localhost:3306/test | /
|
https://mvnrepository.com/artifact/mysql/mysql-connector-java
|
| db2 | com.ibm.db2.jcc.DB2Driver |
jdbc:db2://localhost:50000/testdb |
com.ibm.db2.jcc.DB2XADataSource |
https://mvnrepository.com/artifact/com.ibm.db2.jcc/db2jcc/db2jcc4
|
| saphana | com.sap.db.jdbc.Driver |
jdbc:sap://localhost:39015 | /
|
https://mvnrepository.com/artifact/com.sap.cloud.db.jdbc/ngdbc
|
@@ -418,45 +533,45 @@ there are some reference value for params above.
| Dsql | org.postgresql.Driver |
jdbc:postgresql://Amazon Aurora DSQL Cluster Endpoint:5432/postgres |
org.postgresql.xa.PGXADataSource |
https://mvnrepository.com/artifact/org.postgresql/postgresql
|
| YashanDB | com.yashandb.jdbc.Driver |
jdbc:yasdb://localhost:1688/SYS | /
|
https://mvnrepository.com/artifact/com.yashandb/yashandb-jdbc
|
-## Example
+## Common patterns
-Simple
+### Custom SQL
-```
+```hocon
jdbc {
url = "jdbc:mysql://localhost:3306/test"
driver = "com.mysql.cj.jdbc.Driver"
- user = "root"
+ username = "root"
password = "123456"
query = "insert into test_table(name,age) values(?,?)"
}
```
-Exactly-once
+### Exactly-once with custom SQL
Turn on exact one-time semantics by setting `is_exactly_once`
-```
+```hocon
jdbc {
url = "jdbc:mysql://localhost:3306/test"
driver = "com.mysql.cj.jdbc.Driver"
max_retries = 0
- user = "root"
+ username = "root"
password = "123456"
query = "insert into test_table(name,age) values(?,?)"
- is_exactly_once = "true"
+ is_exactly_once = true
xa_data_source_class_name = "com.mysql.cj.jdbc.MysqlXADataSource"
}
```
-Timer flush
+### Timer flush on Zeta
-Enable timer-based flush by configuring `sink.flush.interval` in the `env`
block. The JDBC sink automatically supports timer flush — when
`sink.flush.interval` is set, the engine periodically injects a `FlushSignal`
into the record stream, and the sink flushes all buffered records to the
database immediately, regardless of whether `batch_size` has been reached.
+This engine-level feature is supported only by Zeta. Spark and Flink do not
inject `FlushSignal` records, so `sink.flush.interval` does not enable timer
flush on those engines. On Zeta, configure `sink.flush.interval` in the `env`
block. The engine periodically injects a `FlushSignal` into the record stream,
and JDBC Sink flushes all buffered records immediately, regardless of whether
`batch_size` has been reached.
:::tip
@@ -475,80 +590,82 @@ sink {
jdbc {
url = "jdbc:mysql://localhost:3306/test"
driver = "com.mysql.cj.jdbc.Driver"
- user = "root"
+ username = "root"
password = "123456"
database = "sink_database"
table = "sink_table"
+ generate_sink_sql = true
primary_keys = ["id"]
batch_size = 10000
}
}
```
-CDC(Change data capture) event
+### Change data capture events
jdbc receive CDC example
-```
+```hocon
sink {
jdbc {
url = "jdbc:mysql://localhost:3306"
driver = "com.mysql.cj.jdbc.Driver"
- user = "root"
+ username = "root"
password = "123456"
database = "sink_database"
table = "sink_table"
- primary_keys = ["key1", "key2", ...]
+ generate_sink_sql = true
+ primary_keys = ["key1", "key2"]
}
}
```
-Add saveMode function
+### Create a missing target table
To facilitate the creation of tables when they do not already exist, set the
`schema_save_mode` to `CREATE_SCHEMA_WHEN_NOT_EXIST`.
-```
+```hocon
sink {
jdbc {
url = "jdbc:mysql://localhost:3306"
driver = "com.mysql.cj.jdbc.Driver"
- user = "root"
+ username = "root"
password = "123456"
- generate_sink_sql = "true"
+ generate_sink_sql = true
database = "sink_database"
table = "sink_table"
- primary_keys = ["key1", "key2", ...]
+ primary_keys = ["key1", "key2"]
schema_save_mode = "CREATE_SCHEMA_WHEN_NOT_EXIST"
- data_save_mode="APPEND_DATA"
+ data_save_mode = "APPEND_DATA"
}
}
```
-Postgresql 9.5 version below support CDC(Change data capture) event
+### PostgreSQL 9.5 and earlier CDC compatibility
For PostgreSQL versions 9.5 and below, setting `compatible_mode` to
`postgresLow` to enable support for PostgreSQL Change Data Capture (CDC)
operations.
-```
+```hocon
sink {
jdbc {
url = "jdbc:postgresql://localhost:5432"
driver = "org.postgresql.Driver"
- user = "root"
+ username = "root"
password = "123456"
- compatible_mode="postgresLow"
+ compatible_mode = "postgresLow"
database = "sink_database"
table = "sink_table"
generate_sink_sql = true
- primary_keys = ["key1", "key2", ...]
+ primary_keys = ["key1", "key2"]
}
}
```
-### Multiple table
+### Multiple tables
-#### example1
+#### MySQL CDC source
```hocon
env {
@@ -574,7 +691,7 @@ sink {
jdbc {
url = "jdbc:mysql://localhost:3306"
driver = "com.mysql.cj.jdbc.Driver"
- user = "root"
+ username = "root"
password = "123456"
generate_sink_sql = true
@@ -585,7 +702,7 @@ sink {
}
```
-#### example2
+#### JDBC source
```hocon
env {
@@ -597,7 +714,7 @@ source {
Jdbc {
driver = oracle.jdbc.driver.OracleDriver
url = "jdbc:oracle:thin:@localhost:1521/XE"
- user = testUser
+ username = testUser
password = testPassword
table_list = [
@@ -618,7 +735,7 @@ sink {
jdbc {
url = "jdbc:mysql://localhost:3306"
driver = "com.mysql.cj.jdbc.Driver"
- user = "root"
+ username = "root"
password = "123456"
generate_sink_sql = true
@@ -629,7 +746,7 @@ sink {
}
```
-#### Dsql example
+#### Amazon Aurora DSQL
```hocon
env {
@@ -641,7 +758,7 @@ source {
Jdbc {
driver = oracle.jdbc.driver.OracleDriver
url = "jdbc:oracle:thin:@localhost:1521/XE"
- user = testUser
+ username = testUser
password = testPassword
table_list = [
@@ -677,7 +794,7 @@ sink {
}
```
-## FAQ
+## Troubleshooting
### Does JDBC Sink support automatic table creation?
@@ -699,10 +816,13 @@ sink {
jdbc {
url = "jdbc:mysql://localhost:3306/mydb"
driver = "com.mysql.cj.jdbc.Driver"
- user = "root"
+ username = "root"
password = "password"
+ max_retries = 0
is_exactly_once = true
xa_data_source_class_name = "com.mysql.cj.jdbc.MysqlXADataSource"
+ generate_sink_sql = true
+ database = "mydb"
table = "target_table"
primary_keys = ["id"]
}
@@ -720,9 +840,13 @@ When a final key set exists and `enable_upsert = true`,
SeaTunnel prefers the da
```hocon
sink {
jdbc {
- url = "jdbc:mysql://localhost:3306/mydb"
- driver = "com.mysql.cj.jdbc.Driver"
- ...
+ url = "jdbc:postgresql://localhost:5432/sales"
+ driver = "org.postgresql.Driver"
+ username = "postgres"
+ password = "password"
+ generate_sink_sql = true
+ database = "sales"
+ table = "public.orders"
primary_keys = ["id"]
}
}
@@ -758,7 +882,7 @@ Use `table = "${table_name}"` and `database =
"${schema_name}"` as placeholders.
### Why is my JDBC driver not found?
-SeaTunnel does not bundle all JDBC drivers due to licensing restrictions.
Place the JDBC driver JAR in `$SEATUNNEL_HOME/lib/` manually before starting
the job. Common drivers:
+SeaTunnel does not bundle all JDBC drivers. For Spark and Flink, place the JAR
in `${SEATUNNEL_HOME}/plugins/Jdbc/lib/` on every execution node. For Zeta,
place it in `${SEATUNNEL_HOME}/lib/` on every SeaTunnel node and restart the
affected processes. Common driver file names include:
- MySQL: `mysql-connector-j-8.x.x.jar`
- PostgreSQL: `postgresql-42.x.x.jar`
diff --git a/docs/zh/connectors/sink/Jdbc.md b/docs/zh/connectors/sink/Jdbc.md
index d501df35ff..e61f108e25 100644
--- a/docs/zh/connectors/sink/Jdbc.md
+++ b/docs/zh/connectors/sink/Jdbc.md
@@ -2,39 +2,157 @@ import ChangeLog from '../changelog/connector-jdbc.md';
# JDBC
-> JDBC 数据接收器
-
## 描述
-通过jdbc写入数据。支持批处理模式和流处理模式，支持并发写入，支持精确一次语义(使用XA事务保证)
+JDBC Sink 通过数据库厂商提供的 JDBC 驱动写入数据。它支持批处理和流处理、并行写入、自动生成 SQL 或自定义 SQL、多表写入、CDC
事件，以及基于 XA 事务的可选精确一次语义。
+
+第一次配置 JDBC Sink 时，请先阅读[选择写入模式](#选择写入模式)和[快速入门](#快速入门postgresql)，再按需查阅后面的完整参数说明。
## 使用依赖
-### 用于Spark/Flink引擎
+先安装 `connector-jdbc` 插件：
+
+```plugin_config
+--seatunnel-connectors--
+connector-jdbc
+--end--
+```
+
+```bash
+cd "${SEATUNNEL_HOME}"
+sh bin/install-plugin.sh
+```
+
+不同数据库厂商的 JDBC 驱动具有不同的许可证和再分发条款，而且驱动版本还必须同时兼容目标数据库和 Java 运行时，因此 SeaTunnel
不会统一内置所有 JDBC 驱动。请自行下载合适的驱动，并在启动任务前把 JAR 放入对应引擎的目录。
+
+### Spark 和 Flink 引擎
+
+把 JDBC 驱动放到每个 SeaTunnel 执行节点的 `${SEATUNNEL_HOME}/plugins/Jdbc/lib/`。
+
+### Zeta 引擎
+
+把 JDBC 驱动放到每个 SeaTunnel 节点的 `${SEATUNNEL_HOME}/lib/`，然后重启受影响的 SeaTunnel
进程，让驱动进入类路径。
+
+常用驱动类名和下载地址见[驱动参考](#驱动参考)。
+
+## 选择写入模式
+
+JDBC Sink 有两种互斥的写入模式。请先确定模式，再配置其余参数。
+
+| 使用场景 | 必需配置 | 行为 |
+|----------|----------|------|
+| 由 SeaTunnel 生成 SQL | `generate_sink_sql = true`、`database`，通常还要配置 `table` |
推荐大多数任务使用。SeaTunnel 可以根据上游 schema 和 RowKind 生成 INSERT、数据库原生 UPSERT、UPDATE 和
DELETE；可以使用 SaveMode 和自动建表。 |
+| 用户提供 SQL | `query = "INSERT ... VALUES (?, ...)"` | 适合必须完全控制目标 SQL 的场景。`?`
参数按照上游字段顺序绑定；此模式不会执行 SaveMode 相关配置。 |
+
+不要同时配置两种模式。`generate_sink_sql` 默认是 `false`，因此没有显式设置 `generate_sink_sql = true`
的任务必须提供 `query`。
+
+使用自动生成 SQL 时，如果目标端需要 Upsert、Update 或 Delete，请配置 `primary_keys`。未配置时，SeaTunnel
会尝试从上游 Catalog 元数据继承主键，再尝试第一个唯一键；仍然没有可用键时，会退化为普通 INSERT。
+
+## 快速入门：PostgreSQL
+
+下面使用自动生成 SQL 写入一张已经创建的 PostgreSQL 表。这份配置和预期结果已经在 PostgreSQL 14
中验证，包括示例数据写入和最终数据库值。
+
+1. 按照[使用依赖](#使用依赖)的说明放置兼容版本的 PostgreSQL JDBC 驱动。
+
+2. 创建目标表：
+
+```sql
+CREATE TABLE public.orders (
+ id BIGINT PRIMARY KEY,
+ customer_name VARCHAR(100) NOT NULL,
+ amount DECIMAL(10, 2) NOT NULL
+);
+```
+
+3. 把下面的任务保存为
`${SEATUNNEL_HOME}/config/jdbc-sink-quick-start.conf`，并按实际环境替换主机、账号和数据库名。
+
+```hocon
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
+
+source {
+ FakeSource {
+ row.num = 3
+ schema = {
+ fields {
+ id = bigint
+ customer_name = string
+ amount = "decimal(10, 2)"
+ }
+ }
+ rows = [
+ { kind = INSERT, fields = [1, "Alice", 120.50] }
+ { kind = INSERT, fields = [2, "Bob", 80.00] }
+ { kind = INSERT, fields = [3, "Carol", 42.00] }
+ ]
+ }
+}
+
+sink {
+ Jdbc {
+ url = "jdbc:postgresql://localhost:5432/sales"
+ driver = "org.postgresql.Driver"
+ username = "postgres"
+ password = "change_me"
+ generate_sink_sql = true
+ database = "sales"
+ table = "public.orders"
+ primary_keys = ["id"]
+ schema_save_mode = "ERROR_WHEN_SCHEMA_NOT_EXIST"
+ data_save_mode = "APPEND_DATA"
+ }
+}
+```
-> 1. 需要确保jdbc驱动jar包已经放在目录`${SEATUNNEL_HOME}/plugins/`下。
+4. 运行任务：
-### 适用于 SeaTunnel Zeta 引擎
+```bash
+cd "${SEATUNNEL_HOME}"
+./bin/seatunnel.sh --config ./config/jdbc-sink-quick-start.conf -m local
+```
+
+5. 验证结果：
+
+```sql
+SELECT id, customer_name, amount
+FROM public.orders
+ORDER BY id;
+```
-> 1. 需要确保jdbc驱动jar包已经放到`${SEATUNNEL_HOME}/lib/`目录下。
+预期结果：
+
+| id | customer_name | amount |
+|----|---------------|-------:|
+| 1 | Alice | 120.50 |
+| 2 | Bob | 80.00 |
+| 3 | Carol | 42.00 |
+
+如果任务在写入前失败，请先检查[故障排查](#故障排查)。
## 主要特性
- [x] [精确一次](../../introduction/concepts/connector-v2-features.md)
-使用 `Xa transactions` 来确保 `exactly-once`。所以仅对于支持 `Xa transactions` 的数据库支持
`exactly-once`
-。你可以设置 `is_exactly_once=true` 来启用它。
+Exactly-once 依赖 XA 事务，因此数据库和 JDBC 驱动都必须支持 XA。具体要求见 [Exactly-once
前置条件](#exactly-once-前置条件)。
- [x] [cdc](../../introduction/concepts/connector-v2-features.md)
-- [x] [定时刷新](../../introduction/concepts/connector-v2-features.md)
+- [x] [定时刷新](../../introduction/concepts/connector-v2-features.md)（仅 Zeta 引擎）
## Options
+连接器的 OptionRule 始终要求提供 `url`、`driver`、`schema_save_mode` 和 `data_save_mode`。两个
SaveMode 参数有默认值，任务配置通常可以省略。其他参数会随写入模式变为必填：
+
+- 自动生成 SQL：设置 `generate_sink_sql = true` 并配置 `database`；除非目标表由上游元数据动态提供，否则还要配置
`table`。
+- 自定义 SQL：保持 `generate_sink_sql = false` 并配置 `query`。
+- Exactly-once：设置 `is_exactly_once = true`、`xa_data_source_class_name` 和
`max_retries = 0`。
+
| 名称 | 类型 | 是否必须 | 默认值
|
|-------------------------------------------|---------|------|------------------------------|
| url | String | 是 | -
|
| driver | String | 是 | -
|
-| user | String | 否 | -
|
+| username | String | 否 | -
|
| password | String | 否 | -
|
| query | String | 否 | -
|
| compatible_mode | String | 否 | -
|
@@ -80,9 +198,9 @@ import ChangeLog from '../changelog/connector-jdbc.md';
用于连接远程数据源的 jdbc 类名，如果使用MySQL，则值为`com.mysql.cj.jdbc.Driver`
-### user [string]
+### username [string]
-用户名
+数据库登录用户名。`username` 是规范参数名；未设置 `username` 时，仍兼容旧参数 `user`。
### password [string]
@@ -94,7 +212,7 @@ JDBC 连接的 URL。参考案例：`jdbc:postgresql://localhost/test`
### query [string]
-使用 sql 语句将上游输入数据写入到数据库。如 `INSERT ...`
+用于写入每条上游数据的参数化 SQL，例如 `INSERT INTO target(id, name) VALUES (?, ?)`。SeaTunnel
按上游字段顺序绑定 `?` 参数。该参数只用于自定义 SQL 模式，不能与 `generate_sink_sql = true` 同时使用。
当前限制：当 sink 配置了 `query`（自定义写入 SQL）时，JDBC sink 不会执行 save mode 处理。此模式下
`schema_save_mode`、`data_save_mode`、`custom_sql` 不生效。如需使用 save mode，请改用
`generate_sink_sql = true` 并配置 `database`、`table`。
@@ -129,15 +247,11 @@ Postgres 9.5及以下版本，请设置为 `postgresLow` 来支持 CDC
### database [string]
-使用此 `database` 和 `table-name` 自动生成 SQL，并接收上游输入的数据写入数据库。
-
-此选项与 `query` 选项是互斥的，此选项具有更高的优先级。
+自动生成 SQL 模式下的目标 database 或 catalog。`generate_sink_sql = true` 时必填，不能与 `query`
同时使用。
### table [string]
-使用 `database` 和此 `table-name` 自动生成 SQL，并接收上游输入的数据写入数据库。
-
-此选项与 `query` 选项是互斥的，此选项具有更高的优先级。
+自动生成 SQL 模式下的目标表，不能与 `query` 同时使用。
table 参数可以填入目标表名，这个名字最终会被用作创建或写入的表名，并且支持变量（`${table_name}`，`${schema_name}`）。
替换规则如下：`${schema_name}` 将替换传递给目标端的 SCHEMA 名称，`${table_name}` 将替换传递给目标端的表名。
@@ -166,7 +280,7 @@ Tip: 如果目标数据库有 SCHEMA 的概念，则表参数必须写成 `xxx.x
### primary_keys [array]
-该选项用于辅助生成 insert、delete、update 等 sql 语句。设置了该选项，将会根据该选项生成对应的 sql 语句
+生成数据库原生 UPSERT、UPDATE 和 DELETE 语句时使用的目标键列。未配置时，SeaTunnel 会尝试从上游 catalog
元数据继承主键或第一组唯一键；仍无可用键时，自动生成的 SQL 退回普通 INSERT。
### connection_check_timeout_sec [int]
@@ -182,11 +296,11 @@ JDBC 连接建立后的 socket 读取超时时间，单位毫秒。默认值为
### max_retries [int]
-重试提交失败的最大次数（executeBatch）
+JDBC `executeBatch` 失败后的重试次数。Exactly-once 模式要求设置为 `0`，重试失败的 XA batch 可能破坏事务保证。
### batch_size [int]
-对于批量写入，当缓冲的记录数达到 `batch_size` 数量或者时间达到 `checkpoint.interval` 时，数据将被刷新到数据库中
+每个 batch 最多缓存的行数。达到 `batch_size`、checkpoint 准备提交或 writer 关闭时会执行
flush。增大该值可能提高吞吐，但会占用更多内存，并增加故障后需要重试的数据量。
### batch_interval_ms [long]
@@ -194,11 +308,11 @@ JDBC 连接建立后的 socket 读取超时时间，单位毫秒。默认值为
### is_exactly_once [boolean]
-是否启用通过XA事务实现的精确一次语义。开启，你还需要设置 `xa_data_source_class_name`
+是否通过 XA 事务启用 exactly-once。开启时必须配置 `xa_data_source_class_name`、`max_retries =
0`，且数据库和驱动都要支持 XA；该模式不支持定时 flush。
### generate_sink_sql [boolean]
-根据要写入的数据库表结构生成 sql 语句
+为 `true` 时，根据上游 schema 和 RowKind 自动生成写入语句。需要配置 `database`，通常还要配置
`table`，并且不能配置 `query`。默认值为 `false`，此时必须配置 `query`。
### xa_data_source_class_name [string]
@@ -373,17 +487,20 @@ AWS IAM 认证中所需要的secret_access_key。 该参考仅适用于 dialect=
### region [String]
Amazon Aurora DSQL 所在的区域。 该参考仅适用于 dialect="dsql"
-## tips
+## Exactly-once 前置条件
+
+`is_exactly_once = true` 时，JDBC Sink 使用 XA 事务。启用前请确认：
+
+- 设置 `max_retries = 0`，并为已安装的驱动配置正确的 `xa_data_source_class_name`。
+- PostgreSQL 必须允许 prepared transaction。把 `max_prepared_transactions`
设置为能够容纳预期并发事务的正数；如果数据库要求，修改后需重启 PostgreSQL。
+- MySQL Server 和 Connector/J 的组合必须支持 Sink 使用的 XA 操作。执行 XA recovery 的账号还可能需要
`XA_RECOVER_ADMIN` 权限，请按所用 MySQL 版本的要求配置。
+- 不要配置 `sink.flush.interval`，XA 事务边界由 checkpoint 控制。
-在 is_exactly_once = "true" 的情况下，使用 XA 事务。这需要数据库支持，有些数据库需要一些设置：<br/>
-1 postgres 需要设置 `max_prepared_transactions > 1` 例如 `ALTER SYSTEM set
max_prepared_transactions to 10` <br/>
-2 mysql 版本需要 >= `8.0.29` 并且非 root 用户需要授予 `XA_RECOVER_ADMIN` 权限。例如:将 test_db.*
上的 XA_RECOVER_ADMIN
-授予 `'user1'@'%'`<br/>
-3 mysql可以尝试在url中添加 `rewriteBatchedStatements=true` 参数以获得更好的性能<br/>
+对于非 XA 的 MySQL 批量任务，可以尝试在 JDBC URL 中加入 `rewriteBatchedStatements=true`
提升吞吐；实际效果应结合驱动版本和业务负载验证。
-## 附录
+## 驱动参考
-附录参数仅提供参考
+下表仅作为起点；驱动制品和版本应以数据库厂商的兼容矩阵为准。
| 数据源 | driver | url
| xa_data_source_class_name
| maven
|
|------------|----------------------------------------------|--------------------------------------------------------------------|----------------------------------------------------|----------------------------------------------------------------------------------------------------|
@@ -394,7 +511,7 @@ Amazon Aurora DSQL 所在的区域。 该参考仅适用于 dialect="dsql"
| SQL Server | com.microsoft.sqlserver.jdbc.SQLServerDriver |
jdbc:sqlserver://localhost:1433 |
com.microsoft.sqlserver.jdbc.SQLServerXADataSource |
https://mvnrepository.com/artifact/com.microsoft.sqlserver/mssql-jdbc
|
| Oracle | oracle.jdbc.OracleDriver |
jdbc:oracle:thin:@localhost:1521/xepdb1 |
oracle.jdbc.xa.OracleXADataSource |
https://mvnrepository.com/artifact/com.oracle.database.jdbc/ojdbc8
|
| sqlite | org.sqlite.JDBC |
jdbc:sqlite:test.db | /
|
https://mvnrepository.com/artifact/org.xerial/sqlite-jdbc
|
-| GBase8a | com.gbase.jdbc.Driver |
jdbc:gbase://e2e_gbase8aDb:5258/test | /
|
https://cdn.gbase.cn/products/30/p5CiVwXBKQYIUGN8ecHvk/gbase-connector-java-9.5.0.7-build1-bin.jar
|
+| GBase8a | com.gbase.jdbc.Driver |
jdbc:gbase://localhost:5258/test | /
|
https://cdn.gbase.cn/products/30/p5CiVwXBKQYIUGN8ecHvk/gbase-connector-java-9.5.0.7-build1-bin.jar
|
| StarRocks | com.mysql.cj.jdbc.Driver |
jdbc:mysql://localhost:3306/test | /
|
https://mvnrepository.com/artifact/mysql/mysql-connector-java
|
| db2 | com.ibm.db2.jcc.DB2Driver |
jdbc:db2://localhost:50000/testdb |
com.ibm.db2.jcc.DB2XADataSource |
https://mvnrepository.com/artifact/com.ibm.db2.jcc/db2jcc/db2jcc4
|
| saphana | com.sap.db.jdbc.Driver |
jdbc:sap://localhost:39015 | /
|
https://mvnrepository.com/artifact/com.sap.cloud.db.jdbc/ngdbc
|
@@ -410,45 +527,45 @@ Amazon Aurora DSQL 所在的区域。 该参考仅适用于 dialect="dsql"
| Dsql | org.postgresql.Driver |
jdbc:postgresql://Amazon Aurora DSQL Cluster Endpoint:5432/postgres |
org.postgresql.xa.PGXADataSource |
https://mvnrepository.com/artifact/org.postgresql/postgresql
|
| YashanDB | com.yashandb.jdbc.Driver |
jdbc:yasdb://localhost:1688/SYS | /
|
https://mvnrepository.com/artifact/com.yashandb/yashandb-jdbc
|
-## 示例
+## 常用模式
-简单示例
+### 自定义 SQL
-```
+```hocon
jdbc {
url = "jdbc:mysql://localhost:3306/test"
driver = "com.mysql.cj.jdbc.Driver"
- user = "root"
+ username = "root"
password = "123456"
query = "insert into test_table(name,age) values(?,?)"
}
```
-精确一次 (Exactly-once)
+### 自定义 SQL 的 Exactly-once
通过设置 `is_exactly_once` 开启精确一次语义
-```
+```hocon
jdbc {
url = "jdbc:mysql://localhost:3306/test"
driver = "com.mysql.cj.jdbc.Driver"
max_retries = 0
- user = "root"
+ username = "root"
password = "123456"
query = "insert into test_table(name,age) values(?,?)"
- is_exactly_once = "true"
+ is_exactly_once = true
xa_data_source_class_name = "com.mysql.cj.jdbc.MysqlXADataSource"
}
```
-定时刷新 (Timer flush)
+### Zeta 定时刷新
-在 `env` 块中配置 `sink.flush.interval` 即可启用定时刷新。JDBC sink 自动支持定时刷新——当设置了
`sink.flush.interval` 时，引擎会定期向记录流中注入 `FlushSignal` 信号，sink 收到该信号后，无论
`batch_size` 是否已满，都会立即将缓冲中的所有数据刷入数据库。
+该引擎级能力仅由 Zeta 支持。Spark 和 Flink 不会注入 `FlushSignal`，因此在这两个引擎中配置
`sink.flush.interval` 不能启用定时刷新。在 Zeta 中，可以在 `env` 块配置
`sink.flush.interval`；引擎会定期向记录流注入 `FlushSignal`，JDBC Sink 收到后会立即刷出全部缓冲数据，无论是否达到
`batch_size`。
:::tip
@@ -467,79 +584,118 @@ sink {
jdbc {
url = "jdbc:mysql://localhost:3306/test"
driver = "com.mysql.cj.jdbc.Driver"
- user = "root"
+ username = "root"
password = "123456"
database = "sink_database"
table = "sink_table"
+ generate_sink_sql = true
primary_keys = ["id"]
batch_size = 10000
}
}
```
-变更数据捕获 (Change data capture) 事件
+### 变更数据捕获事件
jdbc 接收 CDC 示例
-```
+```hocon
sink {
jdbc {
url = "jdbc:mysql://localhost:3306"
driver = "com.mysql.cj.jdbc.Driver"
- user = "root"
+ username = "root"
password = "123456"
-
+ generate_sink_sql = true
database = "sink_database"
table = "sink_table"
- primary_keys = ["key1", "key2", ...]
+ primary_keys = ["key1", "key2"]
}
}
```
-配置表生成策略
+### 自动创建不存在的目标表
通过设置 `schema_save_mode` 配置为 `CREATE_SCHEMA_WHEN_NOT_EXIST` 来支持不存在表时创建表
-```
+```hocon
sink {
jdbc {
url = "jdbc:mysql://localhost:3306"
driver = "com.mysql.cj.jdbc.Driver"
- user = "root"
+ username = "root"
password = "123456"
-
+ generate_sink_sql = true
database = "sink_database"
table = "sink_table"
- primary_keys = ["key1", "key2", ...]
+ primary_keys = ["key1", "key2"]
schema_save_mode = "CREATE_SCHEMA_WHEN_NOT_EXIST"
- data_save_mode="APPEND_DATA"
+ data_save_mode = "APPEND_DATA"
}
}
```
-支持Postgres 9.5及以下版本的 CDC 示例
+### PostgreSQL 9.5 及以下版本 CDC 兼容模式
Postgres 9.5及以下版本，通过设置 `compatible_mode` 配置为 `postgresLow` 来支持 Postgres CDC 操作
-```
+```hocon
sink {
jdbc {
url = "jdbc:postgresql://localhost:5432"
driver = "org.postgresql.Driver"
- user = "root"
+ username = "root"
password = "123456"
- compatible_mode="postgresLow"
+ compatible_mode = "postgresLow"
database = "sink_database"
table = "sink_table"
generate_sink_sql = true
- primary_keys = ["key1", "key2", ...]
+ primary_keys = ["key1", "key2"]
}
}
```
+### 多表写入
-#### Dsql 示例
+#### MySQL CDC Source
+
+```hocon
+env {
+ parallelism = 1
+ job.mode = "STREAMING"
+ checkpoint.interval = 5000
+}
+
+source {
+ Mysql-CDC {
+ url = "jdbc:mysql://127.0.0.1:3306/seatunnel"
+ username = "root"
+ password = "******"
+
+ table-names = ["seatunnel.role", "seatunnel.user", "galileo.Bucket"]
+ }
+}
+
+transform {
+}
+
+sink {
+ jdbc {
+ url = "jdbc:mysql://localhost:3306"
+ driver = "com.mysql.cj.jdbc.Driver"
+ username = "root"
+ password = "123456"
+ generate_sink_sql = true
+
+ database = "${database_name}_test"
+ table = "${table_name}_test"
+ primary_keys = ["${primary_key}"]
+ }
+}
+```
+
+#### JDBC Source
```hocon
env {
@@ -551,7 +707,51 @@ source {
Jdbc {
driver = oracle.jdbc.driver.OracleDriver
url = "jdbc:oracle:thin:@localhost:1521/XE"
- user = testUser
+ username = testUser
+ password = testPassword
+
+ table_list = [
+ {
+ table_path = "TESTSCHEMA.TABLE_1"
+ },
+ {
+ table_path = "TESTSCHEMA.TABLE_2"
+ }
+ ]
+ }
+}
+
+transform {
+}
+
+sink {
+ jdbc {
+ url = "jdbc:mysql://localhost:3306"
+ driver = "com.mysql.cj.jdbc.Driver"
+ username = "root"
+ password = "123456"
+ generate_sink_sql = true
+
+ database = "${schema_name}_test"
+ table = "${table_name}_test"
+ primary_keys = ["${primary_key}"]
+ }
+}
+```
+
+#### Amazon Aurora DSQL
+
+```hocon
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
+
+source {
+ Jdbc {
+ driver = oracle.jdbc.driver.OracleDriver
+ url = "jdbc:oracle:thin:@localhost:1521/XE"
+ username = testUser
password = testPassword
table_list = [
@@ -587,7 +787,7 @@ sink {
}
```
-## 常见问题
+## 故障排查
### JDBC Sink 支持自动建表吗？
@@ -609,10 +809,13 @@ sink {
jdbc {
url = "jdbc:mysql://localhost:3306/mydb"
driver = "com.mysql.cj.jdbc.Driver"
- user = "root"
+ username = "root"
password = "password"
+ max_retries = 0
is_exactly_once = true
xa_data_source_class_name = "com.mysql.cj.jdbc.MysqlXADataSource"
+ generate_sink_sql = true
+ database = "mydb"
table = "target_table"
primary_keys = ["id"]
}
@@ -630,9 +833,13 @@ SeaTunnel 只有在最终拿到了主键/唯一键信息时，才会进入 upser
```hocon
sink {
jdbc {
- url = "jdbc:mysql://localhost:3306/mydb"
- driver = "com.mysql.cj.jdbc.Driver"
- ...
+ url = "jdbc:postgresql://localhost:5432/sales"
+ driver = "org.postgresql.Driver"
+ username = "postgres"
+ password = "password"
+ generate_sink_sql = true
+ database = "sales"
+ table = "public.orders"
primary_keys = ["id"]
}
}
@@ -668,7 +875,7 @@ sink {
### 为什么提示 JDBC 驱动未找到？
-SeaTunnel 出于许可证原因不内置所有 JDBC 驱动，需手动将驱动 JAR 放入 `$SEATUNNEL_HOME/lib/`。常用驱动：
+SeaTunnel 不内置所有 JDBC 驱动。Spark 和 Flink 需要把 JAR 放到每个执行节点的
`${SEATUNNEL_HOME}/plugins/Jdbc/lib/`；Zeta 需要放到每个 SeaTunnel 节点的
`${SEATUNNEL_HOME}/lib/`，然后重启受影响的进程。常见驱动文件名包括：
- MySQL：`mysql-connector-j-8.x.x.jar`
- PostgreSQL：`postgresql-42.x.x.jar`
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/JdbcAutoGenerateSQLIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/JdbcAutoGenerateSQLIT.java
index 389c50345c..e69d2e5a14 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/JdbcAutoGenerateSQLIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/JdbcAutoGenerateSQLIT.java
@@ -37,10 +37,12 @@ import org.testcontainers.utility.DockerLoggerFactory;
import lombok.extern.slf4j.Slf4j;
import java.io.IOException;
+import java.math.BigDecimal;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.sql.Statement;
+import java.util.Arrays;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.stream.Stream;
@@ -120,6 +122,51 @@ public class JdbcAutoGenerateSQLIT extends TestSuiteBase
implements TestResource
Assertions.assertEquals(0, execResult.getExitCode());
}
+ /** Verifies that the PostgreSQL quick start writes the exact rows
documented for users. */
+ @TestTemplate
+ public void testDocumentedPostgresQuickStart(TestContainer container)
+ throws IOException, InterruptedException {
+ truncateOrdersTable();
+ Container.ExecResult execResult =
+ container.executeJob("/jdbc_sink_postgres_quick_start.conf");
+ Assertions.assertEquals(0, execResult.getExitCode());
+
+ List<List<Object>> result =
+ querySql(
+ "select id, customer_name, amount from orders order by
id",
+ () -> {
+ try {
+ return DriverManager.getConnection(
+ postgreSQLContainer.getJdbcUrl(),
+ postgreSQLContainer.getUsername(),
+ postgreSQLContainer.getPassword());
+ } catch (SQLException e) {
+ throw new RuntimeException(e);
+ }
+ });
+ Assertions.assertEquals(3, result.size());
+ Assertions.assertEquals(
+ Arrays.asList(1L, "Alice", new BigDecimal("120.50")),
result.get(0));
+ Assertions.assertEquals(Arrays.asList(2L, "Bob", new
BigDecimal("80.00")), result.get(1));
+ Assertions.assertEquals(Arrays.asList(3L, "Carol", new
BigDecimal("42.00")), result.get(2));
+ }
+
+ /**
+ * Clears rows from earlier TestTemplate invocations so every engine must
prove its own write.
+ */
+ private void truncateOrdersTable() {
+ try (Connection connection =
+ DriverManager.getConnection(
+ postgreSQLContainer.getJdbcUrl(),
+ postgreSQLContainer.getUsername(),
+ postgreSQLContainer.getPassword());
+ Statement statement = connection.createStatement()) {
+ statement.execute("truncate table orders");
+ } catch (SQLException e) {
+ throw new RuntimeException("Truncating PostgreSql orders table
failed!", e);
+ }
+ }
+
private void initializeJdbcTable() {
try (Connection connection =
DriverManager.getConnection(
@@ -135,6 +182,11 @@ public class JdbcAutoGenerateSQLIT extends TestSuiteBase
implements TestResource
+ "timestamp_tz TIMESTAMPTZ \n"
+ ")";
statement.execute(sink);
+ statement.execute(
+ "create table if not exists orders("
+ + "id BIGINT PRIMARY KEY,"
+ + "customer_name VARCHAR(100) NOT NULL,"
+ + "amount DECIMAL(10, 2) NOT NULL)");
} catch (SQLException e) {
throw new RuntimeException("Initializing PostgreSql table
failed!", e);
}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/resources/jdbc_sink_postgres_quick_start.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/resources/jdbc_sink_postgres_quick_start.conf
new file mode 100644
index 0000000000..1a90032dca
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/resources/jdbc_sink_postgres_quick_start.conf
@@ -0,0 +1,54 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
+
+source {
+ FakeSource {
+ row.num = 3
+ schema = {
+ fields {
+ id = bigint
+ customer_name = string
+ amount = "decimal(10, 2)"
+ }
+ }
+ rows = [
+ { kind = INSERT, fields = [1, "Alice", 120.50] }
+ { kind = INSERT, fields = [2, "Bob", 80.00] }
+ { kind = INSERT, fields = [3, "Carol", 42.00] }
+ ]
+ }
+}
+
+sink {
+ Jdbc {
+ url = "jdbc:postgresql://postgresql:5432/test?loggerLevel=OFF"
+ driver = "org.postgresql.Driver"
+ username = "test"
+ password = "test"
+ generate_sink_sql = true
+ database = "test"
+ table = "public.orders"
+ primary_keys = ["id"]
+ schema_save_mode = "ERROR_WHEN_SCHEMA_NOT_EXIST"
+ data_save_mode = "APPEND_DATA"
+ }
+}
