From 3f559e24ca695c326a17f26f1578108125a71187 Mon Sep 17 00:00:00 2001
From: Raffaellorr <38311006+Raffaellorr@users.noreply.github.com>
Date: Fri, 9 Sep 2022 20:38:47 +0800
Subject: [PATCH 1/2] Fix COLLAPSING_MERGE_TREE `Update` statement

It would insert an empty statement, which covered the `before` statement. Now I insert a `before` statement first and then insert an `after` statement.
---
 .../clickhouse/sink/connector/db/DbWriter.java        | 11 +++++++++--
 1 file changed, 9 insertions(+), 2 deletions(-)

diff --git a/src/main/java/com/altinity/clickhouse/sink/connector/db/DbWriter.java b/src/main/java/com/altinity/clickhouse/sink/connector/db/DbWriter.java
index 5545fcc4f..a3d49616d 100644
--- a/src/main/java/com/altinity/clickhouse/sink/connector/db/DbWriter.java
+++ b/src/main/java/com/altinity/clickhouse/sink/connector/db/DbWriter.java
@@ -401,6 +401,7 @@ public BlockMetaData addToPreparedStatementBatch(String topicName, Map<MutablePa
             try (PreparedStatement ps = this.conn.prepareStatement(insertQuery)) {
 
                 List<ClickHouseStruct> recordsList = entry.getValue();
+                boolean beforeInserted = false;
                 for (ClickHouseStruct record : recordsList) {
                     try {
                         bmd.update(record);
@@ -418,9 +419,15 @@ public BlockMetaData addToPreparedStatementBatch(String topicName, Map<MutablePa
                         insertPreparedStatement(entry.getKey().right, ps, record.getAfterModifiedFields(), record, record.getAfterStruct(), false);
                     } else if(CdcRecordState.CDC_RECORD_STATE_BOTH == getCdcSectionBasedOnOperation(record.getCdcOperation()))  {
                         if(this.engine != null && this.engine.getEngine().equalsIgnoreCase(DBMetadata.TABLE_ENGINE.COLLAPSING_MERGE_TREE.getEngine())) {
-                            insertPreparedStatement(entry.getKey().right, ps, record.getBeforeModifiedFields(), record, record.getBeforeStruct(), true);
+                            if (!beforeInseted) {
+                                insertPreparedStatement(entry.getKey().right, ps, record.getBeforeModifiedFields(), record, record.getBeforeStruct(), true);
+                            } else {
+                                insertPreparedStatement(entry.getKey().right, ps, record.getAfterModifiedFields(), record, record.getAfterStruct(), false);
+                            }
+                            beforeInseted = !beforeInseted;
+                        } else {
+                            insertPreparedStatement(entry.getKey().right, ps, record.getAfterModifiedFields(), record, record.getAfterStruct(), false);
                         }
-                        insertPreparedStatement(entry.getKey().right, ps, record.getAfterModifiedFields(), record, record.getAfterStruct(), false);
                     } else {
                         log.error("INVALID CDC RECORD STATE");
                     }

From bc2f768a55ac6fc010dc20e55042ff3497ca5aa2 Mon Sep 17 00:00:00 2001
From: Raffaellorr <38311006+Raffaellorr@users.noreply.github.com>
Date: Mon, 12 Sep 2022 09:50:48 +0800
Subject: [PATCH 2/2] fix typo `beforeInserted`

---
 .../com/altinity/clickhouse/sink/connector/db/DbWriter.java   | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/src/main/java/com/altinity/clickhouse/sink/connector/db/DbWriter.java b/src/main/java/com/altinity/clickhouse/sink/connector/db/DbWriter.java
index a3d49616d..32e80ad5d 100644
--- a/src/main/java/com/altinity/clickhouse/sink/connector/db/DbWriter.java
+++ b/src/main/java/com/altinity/clickhouse/sink/connector/db/DbWriter.java
@@ -419,12 +419,12 @@ public BlockMetaData addToPreparedStatementBatch(String topicName, Map<MutablePa
                         insertPreparedStatement(entry.getKey().right, ps, record.getAfterModifiedFields(), record, record.getAfterStruct(), false);
                     } else if(CdcRecordState.CDC_RECORD_STATE_BOTH == getCdcSectionBasedOnOperation(record.getCdcOperation()))  {
                         if(this.engine != null && this.engine.getEngine().equalsIgnoreCase(DBMetadata.TABLE_ENGINE.COLLAPSING_MERGE_TREE.getEngine())) {
-                            if (!beforeInseted) {
+                            if (!beforeInserted) {
                                 insertPreparedStatement(entry.getKey().right, ps, record.getBeforeModifiedFields(), record, record.getBeforeStruct(), true);
                             } else {
                                 insertPreparedStatement(entry.getKey().right, ps, record.getAfterModifiedFields(), record, record.getAfterStruct(), false);
                             }
-                            beforeInseted = !beforeInseted;
+                            beforeInserted = !beforeInserted;
                         } else {
                             insertPreparedStatement(entry.getKey().right, ps, record.getAfterModifiedFields(), record, record.getAfterStruct(), false);
                         }