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-12848-934276b2f735349c166569de58b490e21cc8319b in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 2bc9d0eab71168dba624ed784d7ab5e89ab9bab0 Author: Goutam Adwant <[email protected]> AuthorDate: Mon Oct 19 12:34:38 2026 +0000 [Fix][Connector-V2] Run the final Databend CDC merge when the writer closes (#12848) --- .../databend/sink/DatabendSinkWriter.java | 8 ++- .../databend/sink/DatabendSinkWriterTest.java | 66 ++++++++++++++++++++++ 2 files changed, 72 insertions(+), 2 deletions(-) diff --git a/seatunnel-connectors-v2/connector-databend/src/main/java/org/apache/seatunnel/connectors/seatunnel/databend/sink/DatabendSinkWriter.java b/seatunnel-connectors-v2/connector-databend/src/main/java/org/apache/seatunnel/connectors/seatunnel/databend/sink/DatabendSinkWriter.java index 57ff26a377..9b245b28af 100644 --- a/seatunnel-connectors-v2/connector-databend/src/main/java/org/apache/seatunnel/connectors/seatunnel/databend/sink/DatabendSinkWriter.java +++ b/seatunnel-connectors-v2/connector-databend/src/main/java/org/apache/seatunnel/connectors/seatunnel/databend/sink/DatabendSinkWriter.java @@ -88,6 +88,9 @@ public class DatabendSinkWriter private String streamName; private String targetTableName; private PreparedStatement cdcPreparedStatement; + // Set once a CDC batch reaches the raw table, cleared after it is merged into the target table. + // executeBatch() resets batchCount, so batchCount cannot tell whether a merge is still needed. + private boolean hasUnmergedCdcRows = false; private String conflictKey; private boolean enableDelete; @@ -224,7 +227,7 @@ public class DatabendSinkWriter } private void performMerge() { - if (batchCount <= 0) { + if (!hasUnmergedCdcRows) { log.debug("No data to merge, skipping"); return; } @@ -235,7 +238,7 @@ public class DatabendSinkWriter try (Statement stmt = connection.createStatement()) { stmt.execute(mergeSql); log.info("Merge operation completed successfully"); - batchCount = 0; // Reset batch count after successful merge + hasUnmergedCdcRows = false; } catch (SQLException e) { log.error("Failed to execute merge operation: {}", e.getMessage(), e); } @@ -738,6 +741,7 @@ public class DatabendSinkWriter log.info( "CDC batch executed successfully, total affected rows: {}", totalAffected); + hasUnmergedCdcRows = true; verifyRawTableData(rawTableName, sinkTablePath.getDatabaseName()); } else { int[] results = preparedStatement.executeBatch(); diff --git a/seatunnel-connectors-v2/connector-databend/src/test/java/org/apache/seatunnel/connectors/seatunnel/databend/sink/DatabendSinkWriterTest.java b/seatunnel-connectors-v2/connector-databend/src/test/java/org/apache/seatunnel/connectors/seatunnel/databend/sink/DatabendSinkWriterTest.java index b36352b2f1..b3ab305244 100644 --- a/seatunnel-connectors-v2/connector-databend/src/test/java/org/apache/seatunnel/connectors/seatunnel/databend/sink/DatabendSinkWriterTest.java +++ b/seatunnel-connectors-v2/connector-databend/src/test/java/org/apache/seatunnel/connectors/seatunnel/databend/sink/DatabendSinkWriterTest.java @@ -30,11 +30,20 @@ import org.junit.jupiter.api.Test; import java.lang.reflect.Field; import java.lang.reflect.InvocationTargetException; import java.lang.reflect.Method; +import java.sql.Connection; +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.sql.Statement; import java.util.Collections; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.startsWith; +import static org.mockito.Mockito.doCallRealMethod; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; public class DatabendSinkWriterTest { @@ -277,6 +286,63 @@ public class DatabendSinkWriterTest { org.junit.jupiter.api.Assertions.assertFalse(mergeSql.contains("raw_data:score as score")); } + @Test + public void testCloseMergesCdcRowsAlreadyFlushedToRawTable() throws Exception { + Statement statement = mock(Statement.class); + DatabendSinkWriter sinkWriter = mockCdcWriter(statement); + + // write() flushes a batch once batch_size is reached, which resets the batch count. + setPrivateField(sinkWriter, "batchCount", 1); + Method executeBatch = DatabendSinkWriter.class.getDeclaredMethod("executeBatch"); + executeBatch.setAccessible(true); + executeBatch.invoke(sinkWriter); + + sinkWriter.close(); + + verify(statement).execute(startsWith("MERGE INTO test_db.target_table")); + } + + @Test + public void testCloseSkipsMergeWhenNoCdcRowsWereFlushed() throws Exception { + Statement statement = mock(Statement.class); + DatabendSinkWriter sinkWriter = mockCdcWriter(statement); + + sinkWriter.close(); + + verify(statement, never()).execute(startsWith("MERGE INTO")); + } + + private DatabendSinkWriter mockCdcWriter(Statement statement) throws Exception { + Connection connection = mock(Connection.class); + when(connection.createStatement()).thenReturn(statement); + when(connection.getAutoCommit()).thenReturn(true); + ResultSet resultSet = mock(ResultSet.class); + when(statement.executeQuery(anyString())).thenReturn(resultSet); + PreparedStatement cdcStatement = mock(PreparedStatement.class); + when(cdcStatement.executeBatch()).thenReturn(new int[] {1}); + + DatabendSinkWriter sinkWriter = mock(DatabendSinkWriter.class); + setPrivateField(sinkWriter, "connection", connection); + setPrivateField(sinkWriter, "cdcPreparedStatement", cdcStatement); + setPrivateField(sinkWriter, "isCdcMode", true); + setPrivateField(sinkWriter, "sinkTablePath", TablePath.of("test_db", "target_table")); + setPrivateField(sinkWriter, "rawTableName", "raw_table"); + setPrivateField(sinkWriter, "streamName", "cdc_stream"); + setPrivateField(sinkWriter, "targetTableName", "target_table"); + setPrivateField(sinkWriter, "conflictKey", "id"); + setPrivateField( + sinkWriter, + "runtimeRowType", + new SeaTunnelRowType( + new String[] {"id", "name"}, + new org.apache.seatunnel.api.table.type.SeaTunnelDataType[] { + BasicType.INT_TYPE, BasicType.STRING_TYPE + })); + doCallRealMethod().when(sinkWriter).close(); + doCallRealMethod().when(sinkWriter).generateMergeSql(); + return sinkWriter; + } + // Helper method to set private fields using reflection private void setPrivateField(Object target, String fieldName, Object value) throws Exception { Field field = target.getClass().getDeclaredField(fieldName);
