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-12808-50256f12c3092203f05ea98c30aea83a55b4266f in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit eceeb0abad2ef598330da06fdfcb1f2450c8bd6b Author: Ricky Makhija <[email protected]> AuthorDate: Mon Oct 26 05:23:26 2026 +0000 [Fix][Connector-V2][Jdbc] Write the net effect of each key when reducing the sink batch (#12808) --- .../BufferReducedBatchStatementExecutor.java | 24 +++++++++++-- .../BufferReducedBatchStatementExecutorTest.java | 40 ++++++++++++++++++++++ 2 files changed, 61 insertions(+), 3 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 845b114475..3f0bff4735 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 @@ -29,7 +29,9 @@ import lombok.RequiredArgsConstructor; import java.sql.Connection; import java.sql.SQLException; +import java.util.HashMap; import java.util.LinkedHashMap; +import java.util.Map; import java.util.function.Function; @RequiredArgsConstructor @@ -48,6 +50,11 @@ public class BufferReducedBatchStatementExecutor @NonNull private final LinkedHashMap<Object, Triple<Boolean, SeaTunnelRow, SeaTunnelRow>> buffer = new LinkedHashMap<>(); + // Whether each buffered key existed before this batch, judged from its first change. Only the + // last change per key is kept, so an INSERT followed by an UPDATE must still be written as an + // INSERT, and a DELETE followed by an INSERT as an UPDATE. + @NonNull private final Map<Object, Boolean> existedBeforeBatch = new HashMap<>(); + private boolean flushFailed; @Override @@ -64,9 +71,11 @@ public class BufferReducedBatchStatementExecutor } SeaTunnelRow key = keyExtractor.apply(record); + Object bufferKey = bufferKey(key); + existedBeforeBatch.putIfAbsent(bufferKey, !RowKind.INSERT.equals(record.getRowKind())); boolean changeFlag = changeFlag(record.getRowKind()); SeaTunnelRow value = valueTransform.apply(record); - buffer.put(bufferKey(key), Triple.of(changeFlag, key, value)); + buffer.put(bufferKey, Triple.of(changeFlag, key, value)); } /** @@ -88,13 +97,20 @@ public class BufferReducedBatchStatementExecutor public void executeBatch() throws SQLException { try { Boolean preChangeFlag = null; - for (Triple<Boolean, SeaTunnelRow, SeaTunnelRow> bufferedRow : buffer.values()) { + for (Map.Entry<Object, Triple<Boolean, SeaTunnelRow, SeaTunnelRow>> entry : + buffer.entrySet()) { + Triple<Boolean, SeaTunnelRow, SeaTunnelRow> bufferedRow = entry.getValue(); Boolean currentChangeFlag = bufferedRow.getLeft(); if (currentChangeFlag) { if (preChangeFlag != null && !preChangeFlag) { deleteExecutor.executeBatch(); } - upsertExecutor.addToBatch(bufferedRow.getRight()); + SeaTunnelRow value = bufferedRow.getRight().copy(); + value.setRowKind( + existedBeforeBatch.getOrDefault(entry.getKey(), true) + ? RowKind.UPDATE_AFTER + : RowKind.INSERT); + upsertExecutor.addToBatch(value); } else { if (preChangeFlag != null && preChangeFlag) { upsertExecutor.executeBatch(); @@ -112,6 +128,7 @@ public class BufferReducedBatchStatementExecutor } } buffer.clear(); + existedBeforeBatch.clear(); flushFailed = false; } catch (SQLException e) { flushFailed = true; @@ -134,6 +151,7 @@ public class BufferReducedBatchStatementExecutor @Override public void clearBatch() throws SQLException { buffer.clear(); + existedBeforeBatch.clear(); upsertExecutor.clearBatch(); deleteExecutor.clearBatch(); } 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 b3c3e437c7..54cf97793a 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 @@ -27,6 +27,7 @@ import java.sql.Connection; import java.sql.SQLException; import java.util.ArrayList; import java.util.Arrays; +import java.util.Collections; import java.util.List; import java.util.function.Function; @@ -73,6 +74,45 @@ public class BufferReducedBatchStatementExecutorTest extends BufferExecutorTest Function.identity()); } + @Test + void testReducedRowKeepsTheNetEffectOfTheBatch() throws SQLException { + List<SeaTunnelRow> upserts = new ArrayList<>(); + List<SeaTunnelRow> deletes = new ArrayList<>(); + JdbcBatchStatementExecutor<SeaTunnelRow> executor = newKeyedExecutor(upserts, deletes); + // Created then updated in one batch: still has to be inserted. + executor.addToBatch(row(RowKind.INSERT, 1, 1, "a")); + executor.addToBatch(row(RowKind.UPDATE_BEFORE, 1, 1, "a")); + executor.addToBatch(row(RowKind.UPDATE_AFTER, 1, 1, "b")); + // Existing row updated twice. + executor.addToBatch(row(RowKind.UPDATE_AFTER, 2, 2, "x")); + executor.addToBatch(row(RowKind.UPDATE_AFTER, 2, 2, "y")); + // Existing row deleted and re-inserted: the row is still there, so update it. + executor.addToBatch(row(RowKind.DELETE, 3, 3, "p")); + executor.addToBatch(row(RowKind.INSERT, 3, 3, "q")); + // Existing row updated then deleted. + executor.addToBatch(row(RowKind.UPDATE_AFTER, 4, 4, "m")); + executor.addToBatch(row(RowKind.DELETE, 4, 4, "m")); + executor.executeBatch(); + + Assertions.assertEquals( + Arrays.asList("+I[1, 1, b]", "+U[2, 2, y]", "+U[3, 3, q]"), describe(upserts)); + Assertions.assertEquals(Collections.singletonList("+I[4]"), describe(deletes)); + + // A new batch judges existence afresh. + upserts.clear(); + executor.addToBatch(row(RowKind.UPDATE_AFTER, 1, 1, "c")); + executor.executeBatch(); + Assertions.assertEquals(Collections.singletonList("+U[1, 1, c]"), describe(upserts)); + } + + private static List<String> describe(List<SeaTunnelRow> rows) { + List<String> described = new ArrayList<>(); + for (SeaTunnelRow row : rows) { + described.add(row.getRowKind().shortString() + Arrays.toString(row.getFields())); + } + return described; + } + @Test void testRowsWithSameKeyAreReduced() throws SQLException { List<SeaTunnelRow> upserts = new ArrayList<>();
