This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] 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 3ac2d1cec6 [Fix][Connector-V2] Preserve NULL transitions in DB2
upserts (#13037)
3ac2d1cec6 is described below
commit 3ac2d1cec6aa7b3741c4dacdd23020f778e23c02
Author: JacobZheng0927 <[email protected]>
AuthorDate: Fri Oct 30 05:38:46 2026 +0000
[Fix][Connector-V2] Preserve NULL transitions in DB2 upserts (#13037)
---
docs/en/connectors/sink/Jdbc.md | 4 +
docs/zh/connectors/sink/Jdbc.md | 4 +
.../jdbc/internal/dialect/db2/DB2Dialect.java | 8 +-
.../jdbc/internal/dialect/db2/DB2DialectTest.java | 2 +-
.../connectors/seatunnel/jdbc/JdbcDb2UpsertIT.java | 141 +++++++++++++++++++++
5 files changed, 154 insertions(+), 5 deletions(-)
diff --git a/docs/en/connectors/sink/Jdbc.md b/docs/en/connectors/sink/Jdbc.md
index fbebfdfbd4..7f81690643 100644
--- a/docs/en/connectors/sink/Jdbc.md
+++ b/docs/en/connectors/sink/Jdbc.md
@@ -8,6 +8,10 @@ The JDBC Sink connector writes SeaTunnel rows to databases
through a JDBC driver
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.
+### DB2 upsert NULL handling
+
+Generated DB2 upserts update matched rows when an incoming field changes,
including NULL-to-value and value-to-NULL transitions. Equal values, including
two NULLs, do not trigger an unnecessary matched update.
+
## MySQL temporal values
MySQL `DATETIME` maps to SeaTunnel `TIMESTAMP` (`LocalDateTime`). The MySQL
JDBC sink binds its local date/time fields directly, preserving fractional
seconds supported by the target column even when the worker JVM and JDBC
connection time zones differ. MySQL `TIMESTAMP` maps to `TIMESTAMP_TZ` and
retains its instant-based handling.
diff --git a/docs/zh/connectors/sink/Jdbc.md b/docs/zh/connectors/sink/Jdbc.md
index d826866884..9910178e15 100644
--- a/docs/zh/connectors/sink/Jdbc.md
+++ b/docs/zh/connectors/sink/Jdbc.md
@@ -8,6 +8,10 @@ JDBC Sink 通过数据库厂商提供的 JDBC 驱动写入数据。它支持批
第一次配置 JDBC Sink 时，请先阅读[选择写入模式](#选择写入模式)和[快速入门](#快速入门postgresql)，再按需查阅后面的完整参数说明。
+### DB2 upsert 的 NULL 处理
+
+生成的 DB2 upsert 会在传入字段发生变化时更新匹配行，包括 NULL 变为非 NULL 和非 NULL 变为 NULL。相等值（包括两端均为
NULL）不会触发多余的匹配更新。
+
## MySQL 时间值
MySQL `DATETIME` 映射为 SeaTunnel `TIMESTAMP`（`LocalDateTime`）。MySQL JDBC Sink
直接绑定本地日期和时间字段，即使 Worker JVM 与 JDBC 连接使用不同的时区，也会保留目标列支持的小数秒精度。MySQL `TIMESTAMP`
映射为 `TIMESTAMP_TZ`，仍按时间点处理。
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/db2/DB2Dialect.java
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/db2/DB2Dialect.java
index 80ec6709e2..cbb324c457 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/db2/DB2Dialect.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/db2/DB2Dialect.java
@@ -107,10 +107,10 @@ public class DB2Dialect implements JdbcDialect {
Arrays.stream(fieldNames)
.map(
field ->
- "target."
- + quoteIdentifier(field)
- + " <> source."
- + quoteIdentifier(field))
+ String.format(
+ "DECODE(%s, %s, 0, 1) = 1",
+ "target." +
quoteIdentifier(field),
+ "source." +
quoteIdentifier(field)))
.collect(Collectors.joining(" OR "));
// Generate UPDATE SET clause
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/db2/DB2DialectTest.java
b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/db2/DB2DialectTest.java
index 253577dafa..5ff477efe7 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/db2/DB2DialectTest.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/db2/DB2DialectTest.java
@@ -42,7 +42,7 @@ public class DB2DialectTest {
// DB2 JDBC URLs select the database, so schema-qualified table names
must stay
// schema.table in DML statements.
Assertions.assertEquals(
- "MERGE INTO \"test_schema\".\"role\" AS target USING (VALUES
(?, ?, ?, ?, ?, ?, ?)) AS source (\"id\", \"type\", \"role_name\",
\"description\", \"create_time\", \"update_time\", \"id_2\") ON target.\"id\" =
source.\"id\" AND target.\"id_2\" = source.\"id_2\" WHEN MATCHED AND
(target.\"id\" <> source.\"id\" OR target.\"type\" <> source.\"type\" OR
target.\"role_name\" <> source.\"role_name\" OR target.\"description\" <>
source.\"description\" OR target.\"create_time\" <> [...]
+ "MERGE INTO \"test_schema\".\"role\" AS target USING (VALUES
(?, ?, ?, ?, ?, ?, ?)) AS source (\"id\", \"type\", \"role_name\",
\"description\", \"create_time\", \"update_time\", \"id_2\") ON target.\"id\" =
source.\"id\" AND target.\"id_2\" = source.\"id_2\" WHEN MATCHED AND
(DECODE(target.\"id\", source.\"id\", 0, 1) = 1 OR DECODE(target.\"type\",
source.\"type\", 0, 1) = 1 OR DECODE(target.\"role_name\",
source.\"role_name\", 0, 1) = 1 OR DECODE(target.\"description\", [...]
upsertSql);
// test insert sql
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/JdbcDb2UpsertIT.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/JdbcDb2UpsertIT.java
index 8de745e120..7eccf240d0 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/JdbcDb2UpsertIT.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/JdbcDb2UpsertIT.java
@@ -20,16 +20,26 @@ package org.apache.seatunnel.connectors.seatunnel.jdbc;
import org.apache.seatunnel.shade.com.google.common.collect.Lists;
+import org.apache.seatunnel.api.table.catalog.PhysicalColumn;
+import org.apache.seatunnel.api.table.catalog.TableSchema;
+import org.apache.seatunnel.api.table.type.BasicType;
+import org.apache.seatunnel.api.table.type.LocalTimeType;
+import org.apache.seatunnel.api.table.type.PrimitiveByteArrayType;
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import
org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.db2.DB2Dialect;
import org.apache.seatunnel.e2e.common.container.TestContainer;
import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestTemplate;
import org.testcontainers.containers.Container;
import java.io.IOException;
+import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
+import java.time.LocalDate;
import java.util.ArrayList;
import java.util.List;
@@ -114,6 +124,137 @@ public class JdbcDb2UpsertIT extends JdbcDb2IT {
}
}
+ @Test
+ void testPreparedMergePreservesNullTransitionsAndSkipsUnchangedRows()
throws Exception {
+ DB2Dialect dialect = new DB2Dialect();
+ String table = dialect.tableIdentifier(DB2_DATABASE,
"NULL_SAFE_UPSERT");
+ String trigger = dialect.tableIdentifier(DB2_DATABASE,
"NULL_SAFE_UPSERT_TRIGGER");
+ try (Statement ddl = connection.createStatement()) {
+ ddl.execute(
+ "CREATE TABLE "
+ + table
+ + " (ID INT NOT NULL PRIMARY KEY, TXT VARCHAR(20),
"
+ + "NUM INT, FLAG BOOLEAN, BIN VARBINARY(16),
DAY_VALUE DATE, UPDATES INT DEFAULT 0)");
+ try {
+ ddl.execute(
+ "CREATE TRIGGER "
+ + trigger
+ + " BEFORE UPDATE ON "
+ + table
+ + " REFERENCING OLD AS O NEW AS N FOR EACH ROW
"
+ + "BEGIN ATOMIC SET N.UPDATES = O.UPDATES + 1;
END");
+ TableSchema schema =
+ TableSchema.builder()
+ .column(
+ PhysicalColumn.of(
+ "ID", BasicType.INT_TYPE, 0L,
false, null, null))
+ .column(
+ PhysicalColumn.of(
+ "TXT",
+ BasicType.STRING_TYPE,
+ 20L,
+ true,
+ null,
+ null))
+ .column(
+ PhysicalColumn.of(
+ "NUM", BasicType.INT_TYPE, 0L,
true, null, null))
+ .column(
+ PhysicalColumn.of(
+ "FLAG",
+ BasicType.BOOLEAN_TYPE,
+ 0L,
+ true,
+ null,
+ null))
+ .column(
+ PhysicalColumn.of(
+ "BIN",
+
PrimitiveByteArrayType.INSTANCE,
+ 16L,
+ true,
+ null,
+ null))
+ .column(
+ PhysicalColumn.of(
+ "DAY_VALUE",
+ LocalTimeType.LOCAL_DATE_TYPE,
+ 0L,
+ true,
+ null,
+ null))
+ .build();
+ String merge =
+ dialect.getUpsertStatement(
+ DB2_DATABASE,
+ "NULL_SAFE_UPSERT",
+ schema.getFieldNames(),
+ new String[] {"ID"})
+ .get();
+ Object[] row = {1, null, null, null, null, null};
+ Object[] values = {"value", 42, true, new byte[] {1, 2},
LocalDate.of(2024, 1, 2)};
+ int updates = 0;
+ try (PreparedStatement statement =
connection.prepareStatement(merge)) {
+ // First execution binds NULL markers, exercising native
type inference.
+ mergeRow(dialect, schema, statement, row);
+ assertMergeRow(table, row, updates);
+ for (int i = 1; i < row.length; i++) {
+ row[i] = values[i - 1];
+ mergeRow(dialect, schema, statement, row);
+ assertMergeRow(table, row, ++updates);
+ mergeRow(dialect, schema, statement, row);
+ assertMergeRow(table, row, updates);
+ row[i] = null;
+ mergeRow(dialect, schema, statement, row);
+ assertMergeRow(table, row, ++updates);
+ mergeRow(dialect, schema, statement, row);
+ assertMergeRow(table, row, updates);
+ }
+ row[1] = "old";
+ mergeRow(dialect, schema, statement, row);
+ row[1] = "new";
+ mergeRow(dialect, schema, statement, row);
+ assertMergeRow(table, row, updates + 2);
+ row[0] = 2;
+ mergeRow(dialect, schema, statement, row);
+ assertMergeRow(table, row, 0);
+ }
+ } finally {
+ ddl.execute("DROP TABLE " + table);
+ connection.commit();
+ }
+ }
+ }
+
+ private void mergeRow(
+ DB2Dialect dialect, TableSchema schema, PreparedStatement
statement, Object[] row)
+ throws SQLException {
+ dialect.getRowConverter().toExternal(schema, null, new
SeaTunnelRow(row), statement);
+ statement.executeUpdate();
+ connection.commit();
+ }
+
+ private void assertMergeRow(String table, Object[] row, int updates)
throws SQLException {
+ try (Statement statement = connection.createStatement();
+ ResultSet result =
+ statement.executeQuery(
+ "SELECT ID, TXT, NUM, FLAG, BIN, DAY_VALUE,
UPDATES FROM "
+ + table
+ + " WHERE ID = "
+ + row[0])) {
+ Assertions.assertTrue(result.next());
+ Assertions.assertEquals(row[1], result.getString(2));
+ Assertions.assertEquals(row[2], result.getObject(3));
+ Assertions.assertEquals(row[3], result.getObject(4));
+ Assertions.assertArrayEquals((byte[]) row[4], result.getBytes(5));
+ Assertions.assertEquals(
+ row[5], result.getDate(6) == null ? null :
result.getDate(6).toLocalDate());
+ Assertions.assertEquals(
+ updates, result.getInt(7), "Unchanged values must not fire
UPDATE triggers");
+ Assertions.assertFalse(result.next());
+ }
+ }
+
private List<List<Object>> query(String sql) {
try (Statement statement = connection.createStatement();
ResultSet resultSet = statement.executeQuery(sql)) {
