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-12892-9cd5d0368a4c6ccb72ea040edf315bcc1328e481 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 941276923f634df3efd7115592b9cef2eddd2dba Author: FatMan <[email protected]> AuthorDate: Tue Nov 3 06:41:20 2026 +0000 [Test][Connector-V2] Cover composite Jdbc sink keys with a NULL component (#12892) Co-authored-by: chen <[email protected]> --- .../BufferReducedBatchStatementExecutorTest.java | 26 ++++++++++++++++++++++ 1 file changed, 26 insertions(+) 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 54cf97793a..0a23505dee 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 @@ -164,6 +164,32 @@ public class BufferReducedBatchStatementExecutorTest extends BufferExecutorTest Assertions.assertEquals(1, deletes.get(0).getArity()); } + @Test + void testCompositeKeyWithNullComponentIsNotReduced() throws SQLException { + List<SeaTunnelRow> upserts = new ArrayList<>(); + // Composite key (code, name): only rows whose whole key is non-null are reduced + JdbcBatchStatementExecutor<SeaTunnelRow> executor = + new BufferReducedBatchStatementExecutor( + new RecordingExecutor(upserts), + new RecordingExecutor(new ArrayList<>()), + row -> new SeaTunnelRow(new Object[] {row.getField(1), row.getField(2)}), + Function.identity()); + + executor.addToBatch(row(RowKind.INSERT, 1, 7, null)); + executor.addToBatch(row(RowKind.INSERT, 2, 7, null)); + executor.addToBatch(row(RowKind.INSERT, 3, 7, "a")); + executor.addToBatch(row(RowKind.UPDATE_AFTER, 4, 7, "a")); + executor.executeBatch(); + + Assertions.assertEquals(3, upserts.size()); + Assertions.assertEquals( + Arrays.asList(1, 2, 4), + Arrays.asList( + upserts.get(0).getField(0), + upserts.get(1).getField(0), + upserts.get(2).getField(0))); + } + /** Executor keyed by the second field (a nullable unique key column). */ private static JdbcBatchStatementExecutor<SeaTunnelRow> newKeyedExecutor( List<SeaTunnelRow> upserts, List<SeaTunnelRow> deletes) {
