This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-13037-e805677f731be93d254cd844355fd09c89225041 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
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)) {
