Tests for optimizer

Author: bakwc

mysql_ch_replicator/clickhouse_api.py

@@ -176,10 +176,12 @@ def drop_database(self, db_name):
     def create_database(self, db_name):
         self.cursor.execute(f'CREATE DATABASE {db_name}')
 
-    def select(self, table_name, where=None):
+    def select(self, table_name, where=None, final=None):
         query = f'SELECT * FROM {table_name}'
         if where:
             query += f' WHERE {where}'
+        if final is not None:
+            query += f' SETTINGS final = {int(final)};'
         result = self.client.query(query)
         rows = result.result_rows
         columns = result.column_names

mysql_ch_replicator/db_optimizer.py

@@ -3,9 +3,10 @@
 import time
 from logging import getLogger
 
-from config import Settings
-from mysql_api import MySQLApi
-from clickhouse_api import ClickhouseApi
+from .config import Settings
+from .mysql_api import MySQLApi
+from .clickhouse_api import ClickhouseApi
+from .utils import GracefulKiller
 
 
 logger = getLogger(__name__)
@@ -40,7 +41,7 @@ class DbOptimizer:
     def __init__(self, config: Settings):
         self.state = State(os.path.join(
             config.binlog_replicator.data_dir,
-            ''
+            'db_optimizer.bin',
         ))
         self.config = config
         self.mysql_api = MySQLApi(
@@ -72,6 +73,7 @@ def optimize_table(self, db_name, table_name):
             f'OPTIMIZE TABLE {db_name}.{table_name} FINAL SETTINGS mutations_sync = 2'
         )
         logger.info('Optimize finished')
+        self.state.last_process_time[db_name] = time.time()
 
     def optimize_database(self, db_name):
         self.mysql_api.set_database(db_name)
@@ -87,15 +89,15 @@ def optimize_database(self, db_name):
             self.optimize_table(db_name, table)
 
     def run(self):
-        while True:
-            time.sleep(999)
-
+        logger.info('running optimizer')
+        killer = GracefulKiller()
         try:
-            while True:
+            while not killer.kill_now:
                 db_to_optimize = self.select_db_to_optimize()
                 if db_to_optimize is None:
                     time.sleep(min(120, self.config.optimize_interval))
                     continue
                 self.optimize_database(db_name=db_to_optimize)
         except Exception as e:
             logger.error(f'error {e}', exc_info=True)
+        logger.info('optimizer stopped')

mysql_ch_replicator/main.py

@@ -133,7 +133,7 @@ def main():
     parser.add_argument(
         "mode", help="run mode",
         type=str,
-        choices=["run_all", "binlog_replicator", "db_replicator", "monitoring"])
+        choices=["run_all", "binlog_replicator", "db_replicator", "monitoring", "db_optimizer"])
     parser.add_argument("--config", help="config file path", default='config.yaml', type=str)
     parser.add_argument("--db", help="source database(s) name", type=str)
     parser.add_argument("--target_db", help="target database(s) name, if not set will be same as source", type=str)

test_mysql_ch_replicator.py

@@ -344,8 +344,19 @@ def test_runner():
     assert_wait(lambda: len(ch.select(TEST_TABLE_NAME)) == 3)
 
     mysql.execute(f"UPDATE {TEST_TABLE_NAME} SET age=66 WHERE name='Ivan'", commit=True)
-    time.sleep(4)
-    assert_wait(lambda: len(ch.select(TEST_TABLE_NAME)) == 3)
+    assert_wait(lambda: ch.select(TEST_TABLE_NAME, "name='Ivan'")[0]['age'] == 66)
+
+    mysql.execute(f"UPDATE {TEST_TABLE_NAME} SET age=77 WHERE name='Ivan'", commit=True)
+    assert_wait(lambda: ch.select(TEST_TABLE_NAME, "name='Ivan'")[0]['age'] == 77)
+
+    mysql.execute(f"UPDATE {TEST_TABLE_NAME} SET age=88 WHERE name='Ivan'", commit=True)
+    assert_wait(lambda: ch.select(TEST_TABLE_NAME, "name='Ivan'")[0]['age'] == 88)
+
+    mysql.execute(f"INSERT INTO {TEST_TABLE_NAME} (name, age) VALUES ('Vlad', 99);", commit=True)
+
+    assert_wait(lambda: len(ch.select(TEST_TABLE_NAME)) == 4)
+
+    assert_wait(lambda: len(ch.select(TEST_TABLE_NAME, final=False)) == 4)
 
     run_all_runner.stop()
 

tests_config.yaml

@@ -17,3 +17,4 @@ binlog_replicator:
 
 databases: '*test*'
 log_level: 'debug'
+optimize_interval: 3