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-12803-51f553cb45b4b2b37b87f7f654352bc0276daf6b in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 64e92b555b574ab9aa2bd4f7fb40714758d1660a Author: FatMan <[email protected]> AuthorDate: Sat Oct 17 06:41:24 2026 +0000 [Fix][Connector-V2] Do not merge Jdbc sink rows whose key contains NULL (#12803) Co-authored-by: chen <[email protected]> --- .../BufferReducedBatchStatementExecutor.java | 37 ++++++--- .../BufferReducedBatchStatementExecutorTest.java | 95 ++++++++++++++++++++++ 2 files changed, 122 insertions(+), 10 deletions(-) diff --git a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/executor/BufferReducedBatchStatementExecutor.java b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/executor/BufferReducedBatchStatementExecutor.java index 6ef9112ec3..845b114475 100644 --- a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/executor/BufferReducedBatchStatementExecutor.java +++ b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/executor/BufferReducedBatchStatementExecutor.java @@ -17,7 +17,7 @@ package org.apache.seatunnel.connectors.seatunnel.jdbc.internal.executor; -import org.apache.seatunnel.shade.org.apache.commons.lang3.tuple.Pair; +import org.apache.seatunnel.shade.org.apache.commons.lang3.tuple.Triple; import org.apache.seatunnel.api.table.type.RowKind; import org.apache.seatunnel.api.table.type.SeaTunnelRow; @@ -30,8 +30,6 @@ import lombok.RequiredArgsConstructor; import java.sql.Connection; import java.sql.SQLException; import java.util.LinkedHashMap; -import java.util.Map; -import java.util.Set; import java.util.function.Function; @RequiredArgsConstructor @@ -42,7 +40,12 @@ public class BufferReducedBatchStatementExecutor @NonNull private final Function<SeaTunnelRow, SeaTunnelRow> keyExtractor; @NonNull private final Function<SeaTunnelRow, SeaTunnelRow> valueTransform; - @NonNull private final LinkedHashMap<SeaTunnelRow, Pair<Boolean, SeaTunnelRow>> buffer = + /** + * Buffered rows by key, so that only the last change of a key is written. The value holds the + * change flag, the key row and the value row. A key with a NULL field is buffered under a + * unique identity instead, see {@link #bufferKey(SeaTunnelRow)}. + */ + @NonNull private final LinkedHashMap<Object, Triple<Boolean, SeaTunnelRow, SeaTunnelRow>> buffer = new LinkedHashMap<>(); private boolean flushFailed; @@ -63,26 +66,40 @@ public class BufferReducedBatchStatementExecutor SeaTunnelRow key = keyExtractor.apply(record); boolean changeFlag = changeFlag(record.getRowKind()); SeaTunnelRow value = valueTransform.apply(record); - buffer.put(key, Pair.of(changeFlag, value)); + buffer.put(bufferKey(key), Triple.of(changeFlag, key, value)); + } + + /** + * Returns the key a row is buffered under. In SQL a NULL key value never equals another value, + * so the database treats rows with a NULL key field (possible for a key taken from a nullable + * unique key) as different rows, and an upsert or delete by that key never matches them. Such + * rows get a unique identity so that they are not merged into one row within a batch. + */ + private static Object bufferKey(SeaTunnelRow key) { + for (Object field : key.getFields()) { + if (field == null) { + return new Object(); + } + } + return key; } @Override public void executeBatch() throws SQLException { try { Boolean preChangeFlag = null; - Set<Map.Entry<SeaTunnelRow, Pair<Boolean, SeaTunnelRow>>> entrySet = buffer.entrySet(); - for (Map.Entry<SeaTunnelRow, Pair<Boolean, SeaTunnelRow>> entry : entrySet) { - Boolean currentChangeFlag = entry.getValue().getKey(); + for (Triple<Boolean, SeaTunnelRow, SeaTunnelRow> bufferedRow : buffer.values()) { + Boolean currentChangeFlag = bufferedRow.getLeft(); if (currentChangeFlag) { if (preChangeFlag != null && !preChangeFlag) { deleteExecutor.executeBatch(); } - upsertExecutor.addToBatch(entry.getValue().getValue()); + upsertExecutor.addToBatch(bufferedRow.getRight()); } else { if (preChangeFlag != null && preChangeFlag) { upsertExecutor.executeBatch(); } - deleteExecutor.addToBatch(entry.getKey()); + deleteExecutor.addToBatch(bufferedRow.getMiddle()); } preChangeFlag = currentChangeFlag; } diff --git a/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/executor/BufferReducedBatchStatementExecutorTest.java b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/executor/BufferReducedBatchStatementExecutorTest.java index 3f72235389..b3c3e437c7 100644 --- a/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/executor/BufferReducedBatchStatementExecutorTest.java +++ b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/executor/BufferReducedBatchStatementExecutorTest.java @@ -17,10 +17,16 @@ package org.apache.seatunnel.connectors.seatunnel.jdbc.internal.executor; +import org.apache.seatunnel.api.table.type.RowKind; import org.apache.seatunnel.api.table.type.SeaTunnelRow; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + import java.sql.Connection; import java.sql.SQLException; +import java.util.ArrayList; +import java.util.Arrays; import java.util.List; import java.util.function.Function; @@ -66,4 +72,93 @@ public class BufferReducedBatchStatementExecutorTest extends BufferExecutorTest Function.identity(), Function.identity()); } + + @Test + void testRowsWithSameKeyAreReduced() throws SQLException { + List<SeaTunnelRow> upserts = new ArrayList<>(); + JdbcBatchStatementExecutor<SeaTunnelRow> executor = + newKeyedExecutor(upserts, new ArrayList<>()); + + executor.addToBatch(row(RowKind.INSERT, 1, 1, "a")); + executor.addToBatch(row(RowKind.UPDATE_AFTER, 2, 1, "b")); + executor.executeBatch(); + + Assertions.assertEquals(1, upserts.size()); + Assertions.assertEquals("b", upserts.get(0).getField(2)); + } + + @Test + void testRowsWithNullKeyAreNotReduced() throws SQLException { + List<SeaTunnelRow> upserts = new ArrayList<>(); + JdbcBatchStatementExecutor<SeaTunnelRow> executor = + newKeyedExecutor(upserts, new ArrayList<>()); + + executor.addToBatch(row(RowKind.INSERT, 1, null, "a")); + executor.addToBatch(row(RowKind.INSERT, 2, null, "b")); + executor.addToBatch(row(RowKind.INSERT, 3, 7, "c")); + executor.executeBatch(); + + // NULL never equals NULL in SQL, so the database keeps both rows with a NULL key + Assertions.assertEquals(3, upserts.size()); + Assertions.assertEquals( + Arrays.asList(1, 2, 3), + Arrays.asList( + upserts.get(0).getField(0), + upserts.get(1).getField(0), + upserts.get(2).getField(0))); + } + + @Test + void testDeleteWithNullKeyKeepsOrderWithUpserts() throws SQLException { + List<SeaTunnelRow> upserts = new ArrayList<>(); + List<SeaTunnelRow> deletes = new ArrayList<>(); + JdbcBatchStatementExecutor<SeaTunnelRow> executor = newKeyedExecutor(upserts, deletes); + + executor.addToBatch(row(RowKind.INSERT, 1, null, "a")); + executor.addToBatch(row(RowKind.DELETE, 1, null, "a")); + executor.executeBatch(); + + Assertions.assertEquals(1, upserts.size()); + Assertions.assertEquals(1, deletes.size()); + // The delete executor receives the extracted key, not the whole row + Assertions.assertEquals(1, deletes.get(0).getArity()); + } + + /** Executor keyed by the second field (a nullable unique key column). */ + private static JdbcBatchStatementExecutor<SeaTunnelRow> newKeyedExecutor( + List<SeaTunnelRow> upserts, List<SeaTunnelRow> deletes) { + return new BufferReducedBatchStatementExecutor( + new RecordingExecutor(upserts), + new RecordingExecutor(deletes), + row -> new SeaTunnelRow(new Object[] {row.getField(1)}), + Function.identity()); + } + + private static SeaTunnelRow row(RowKind rowKind, Integer id, Integer code, String name) { + SeaTunnelRow row = new SeaTunnelRow(new Object[] {id, code, name}); + row.setRowKind(rowKind); + return row; + } + + private static class RecordingExecutor implements JdbcBatchStatementExecutor<SeaTunnelRow> { + private final List<SeaTunnelRow> recorder; + + RecordingExecutor(List<SeaTunnelRow> recorder) { + this.recorder = recorder; + } + + @Override + public void prepareStatements(Connection connection) {} + + @Override + public void addToBatch(SeaTunnelRow record) { + recorder.add(record); + } + + @Override + public void executeBatch() {} + + @Override + public void closeStatements() {} + } }
