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..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
@@ -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 (!beforeInserted) {
+                                insertPreparedStatement(entry.getKey().right, ps, record.getBeforeModifiedFields(), record, record.getBeforeStruct(), true);
+                            } else {
+                                insertPreparedStatement(entry.getKey().right, ps, record.getAfterModifiedFields(), record, record.getAfterStruct(), false);
+                            }
+                            beforeInserted = !beforeInserted;
+                        } 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");
                     }