Merge pull request #282 from Altinity/ignore_deletes

Added logic to ignore deletes
Altinity · Aug 3, 2023 · bfa9bd4 · bfa9bd4
2 parents dae79fa + ad3dee2
commit bfa9bd4
Show file tree

Hide file tree

Showing 4 changed files with 15 additions and 3 deletions.
diff --git a/sink-connector-lightweight/docker/mysqld.cnf b/sink-connector-lightweight/docker/mysqld.cnf
@@ -1,10 +1,11 @@
 [mysqld]
 max_connections=100000
 
+default_authentication_plugin=mysql_native_password
 
 gtid-mode = on
 enforce-gtid-consistency = true
 # for loading files
 local_infile = on
 # to support tables without PK
-sql_generate_invisible_primary_key=1
+sql_generate_invisible_primary_key=1
diff --git a/...r/src/main/java/com/altinity/clickhouse/sink/connector/ClickHouseSinkConnectorConfig.java b/...r/src/main/java/com/altinity/clickhouse/sink/connector/ClickHouseSinkConnectorConfig.java
@@ -353,7 +353,16 @@ static ConfigDef newConfigDef() {
                         3,
                         ConfigDef.Width.NONE,
                         ClickHouseSinkConnectorConfigVariables.THREAD_POOL_SIZE.toString())
-
+                .define(
+                        ClickHouseSinkConnectorConfigVariables.IGNORE_DELETE.toString(),
+                        Type.BOOLEAN,
+                        false,
+                        Importance.HIGH,
+                        "If true, Deletes are ignored are not persisted to ClickHouse.",
+                        CONFIG_GROUP_CONNECTOR_CONFIG,
+                        3,
+                        ConfigDef.Width.NONE,
+                        ClickHouseSinkConnectorConfigVariables.IGNORE_DELETE.toString())
                 // ToDo: Add JVM Proxy
                 ;
     }

diff --git a/...n/java/com/altinity/clickhouse/sink/connector/ClickHouseSinkConnectorConfigVariables.java b/...n/java/com/altinity/clickhouse/sink/connector/ClickHouseSinkConnectorConfigVariables.java
@@ -3,6 +3,7 @@
 
 public enum ClickHouseSinkConnectorConfigVariables {
 
+    IGNORE_DELETE("ignore_delete"),
     THREAD_POOL_SIZE("thread.pool.size"),
     BUFFER_COUNT("buffer.count"),
     DEDUPLICATION_POLICY("deduplication.policy"),

diff --git a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/db/DbWriter.java b/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/db/DbWriter.java
@@ -659,7 +659,8 @@ public void insertPreparedStatement(Map<String, Integer> columnNameToIndexMap, P
             }
             // Sign column to mark deletes in ReplacingMergeTree
             if(this.replacingMergeTreeDeleteColumn != null && this.columnNameToDataTypeMap.containsKey(replacingMergeTreeDeleteColumn)) {
-                if(columnNameToIndexMap.containsKey(replacingMergeTreeDeleteColumn)) {
+                if(columnNameToIndexMap.containsKey(replacingMergeTreeDeleteColumn) &&
+                        this.config.getBoolean(ClickHouseSinkConnectorConfigVariables.IGNORE_DELETE.toString()) == false) {
                     if (record.getCdcOperation().getOperation().equalsIgnoreCase(ClickHouseConverter.CDC_OPERATION.DELETE.getOperation())) {
                         if(replacingMergeTreeWithIsDeletedColumn)
                             ps.setInt(columnNameToIndexMap.get(replacingMergeTreeDeleteColumn), 1);