Erase using batches (#186)

Author: bakwc

mysql_ch_replicator/clickhouse_api.py

@@ -85,6 +85,7 @@ class ClickhouseApi:
     def __init__(self, database: str | None, clickhouse_settings: ClickhouseSettings):
         self.database = database
         self.clickhouse_settings = clickhouse_settings
+        self.erase_batch_size = clickhouse_settings.erase_batch_size
         self.client = clickhouse_connect.get_client(
             host=clickhouse_settings.host,
             port=clickhouse_settings.port,
@@ -248,22 +249,34 @@ def insert(self, table_name, records, table_structure: TableStructure = None):
 
     def erase(self, table_name, field_name, field_values):
         field_name = ','.join(field_name)
-        field_values = ', '.join(f'({v})' for v in field_values)
-        query = DELETE_QUERY.format(**{
-            'db_name': self.database,
-            'table_name': table_name,
-            'field_name': field_name,
-            'field_values': field_values,
-        })
-        t1 = time.time()
-        self.execute_command(query)
-        t2 = time.time()
-        duration = t2 - t1
+        
+        # Batch large deletions to avoid ClickHouse max query size limit
+        field_values_list = list(field_values)
+        
+        total_duration = 0.0
+        total_records = len(field_values_list)
+        
+        for i in range(0, len(field_values_list), self.erase_batch_size):
+            batch = field_values_list[i:i + self.erase_batch_size]
+            batch_field_values = ', '.join(f'({v})' for v in batch)
+            
+            query = DELETE_QUERY.format(**{
+                'db_name': self.database,
+                'table_name': table_name,
+                'field_name': field_name,
+                'field_values': batch_field_values,
+            })
+            
+            t1 = time.time()
+            self.execute_command(query)
+            t2 = time.time()
+            total_duration += (t2 - t1)
+        
         self.stats.on_event(
             table_name=table_name,
-            duration=duration,
+            duration=total_duration,
             is_insert=False,
-            records=len(field_values),
+            records=total_records,
         )
 
     def drop_database(self, db_name):

mysql_ch_replicator/config.py

@@ -52,6 +52,7 @@ class ClickhouseSettings:
     password: str = ''
     connection_timeout: int = 30
     send_receive_timeout: int = 120
+    erase_batch_size: int = 100000  # Number of records to delete per batch
 
     def validate(self):
         if not isinstance(self.host, str):
@@ -78,6 +79,11 @@ def validate(self):
         if self.send_receive_timeout <= 0:
             raise ValueError(f'send_receive_timeout timeout should be at least 1 second')
 
+        if not isinstance(self.erase_batch_size, int):
+            raise ValueError(f'erase_batch_size should be int and not {stype(self.erase_batch_size)}')
+        if self.erase_batch_size <= 0:
+            raise ValueError('erase_batch_size should be positive')
+
 
 @dataclass
 class BinlogReplicatorSettings:

tests_config.yaml

@@ -9,6 +9,7 @@ clickhouse:
   port: 9123
   user: 'default'
   password: 'admin'
+  erase_batch_size: 2
 
 binlog_replicator:
   data_dir: '/app/binlog/'