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 eceeb0abad [Fix][Connector-V2][Jdbc] Write the net effect of each key
when reducing the sink batch (#12808)
eceeb0abad is described below
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<>();
