Post replication commands (#203)

Author: bakwc

.cursor/rules/rules.mdc

@@ -5,4 +5,4 @@ alwaysApply: true
 ---
 Use following command to run tests:
 
-sudo docker exec -w /app/ -it mysql_ch_replicator-replicator-1 python3 -m pytest -v -s test_mysql_ch_replicator.py -k test_truncate_operation_bug_issue_155
+docker exec -w /app/ -it tests-replicator-1 python3 -m pytest -v -s tests/ -k test_your_test_name

README.md

@@ -275,6 +275,16 @@ mysql_timezone: 'UTC'    # optional, timezone for MySQL timestamp conversion (de
 - `types_mappings` - custom types mapping, eg. you can map char(36) to UUID instead of String, etc.
 - `ignore_deletes` - when set to `true`, DELETE operations in MySQL will be ignored during replication. This creates an append-only model where data is only added, never removed. In this mode, the replicator doesn't create a temporary database and instead replicates directly to the target database.
 - `mysql_timezone` - timezone to use for MySQL timestamp conversion to ClickHouse DateTime64. Default is `'UTC'`. Accepts any valid timezone name (e.g., `'America/New_York'`, `'Europe/London'`, `'Asia/Tokyo'`). This setting ensures proper timezone handling when converting MySQL timestamp fields to ClickHouse DateTime64 with timezone information.
+- `post_initial_replication_commands` - SQL commands to execute in ClickHouse after initial replication completes for each database. Useful for creating materialized views, summary tables, or other database objects. Commands are executed in order, once per database matching the pattern.
+
+```yaml
+post_initial_replication_commands:
+  - databases: '*'
+    commands:
+      - 'CREATE TABLE IF NOT EXISTS summary_table (...) ENGINE = SummingMergeTree() ORDER BY (...)'
+      - 'CREATE MATERIALIZED VIEW IF NOT EXISTS data_mv TO summary_table AS SELECT ...'
+      - 'INSERT INTO summary_table SELECT ... FROM replicated_table'
+```
 
 Few more tables / dbs examples:
 

mysql_ch_replicator/config.py

@@ -48,6 +48,12 @@ class PartitionBy:
     partition_by: str = ''
 
 
+@dataclass
+class PostInitialReplicationCommands:
+    databases: str | list = '*'
+    commands: list = None
+
+
 @dataclass
 class ClickhouseSettings:
     host: str = 'localhost'
@@ -134,6 +140,7 @@ def __init__(self):
         self.check_db_updated_interval = 0
         self.indexes: list[Index] = []
         self.partition_bys: list[PartitionBy] = []
+        self.post_initial_replication_commands: list[PostInitialReplicationCommands] = []
         self.auto_restart_interval = 0
         self.http_host = ''
         self.http_port = 0
@@ -184,6 +191,12 @@ def load(self, settings_file):
                 PartitionBy(**partition_by)
             )
 
+        post_initial_replication_commands = data.pop('post_initial_replication_commands', [])
+        for cmd_config in post_initial_replication_commands:
+            self.post_initial_replication_commands.append(
+                PostInitialReplicationCommands(**cmd_config)
+            )
+        
         assert isinstance(self.databases, str) or isinstance(self.databases, list)
         assert isinstance(self.tables, str) or isinstance(self.tables, list)
         self.binlog_replicator = BinlogReplicatorSettings(**data.pop('binlog_replicator'))
@@ -250,6 +263,15 @@ def get_partition_bys(self, db_name, table_name):
             results.append(partition_by.partition_by)
         return results
 
+    def get_post_initial_replication_commands(self, db_name):
+        results = []
+        for cmd_config in self.post_initial_replication_commands:
+            if not self.is_pattern_matches(db_name, cmd_config.databases):
+                continue
+            if cmd_config.commands:
+                results.extend(cmd_config.commands)
+        return results
+
     def validate(self):
         self.mysql.validate()
         self.clickhouse.validate()

mysql_ch_replicator/db_replicator_initial.py

@@ -123,6 +123,9 @@ def perform_initial_replication(self):
                         f'RENAME DATABASE `{self.replicator.target_database_tmp}` TO `{self.replicator.target_database}`',
                     )
             self.replicator.clickhouse_api.database = self.replicator.target_database
+            
+            # Execute post-initial-replication commands
+            self.execute_post_initial_replication_commands()
         logger.info(f'initial replication - done')
 
     def perform_initial_replication_table(self, table_name):
@@ -425,3 +428,25 @@ def consolidate_worker_record_versions(self, table_name):
                 logger.info(f"Current version {current_version} is already up-to-date with ClickHouse version {max_version}")
         else:
             logger.warning(f"No record version found in ClickHouse for table {table_name}")
+
+    def execute_post_initial_replication_commands(self):
+        """
+        Execute custom commands after initial replication finishes.
+        Commands are configured per database in the config file.
+        """
+        commands = self.replicator.config.get_post_initial_replication_commands(self.replicator.database)
+        
+        if not commands:
+            logger.info(f'No post-initial-replication commands configured for database {self.replicator.database}')
+            return
+        
+        logger.info(f'Executing {len(commands)} post-initial-replication commands for database {self.replicator.database}')
+        
+        self.replicator.clickhouse_api.execute_command(f'USE `{self.replicator.target_database}`')
+        
+        for i, command in enumerate(commands, 1):
+            logger.info(f'Executing command {i}/{len(commands)}: {command[:100]}...')
+            self.replicator.clickhouse_api.execute_command(command)
+            logger.info(f'Command {i}/{len(commands)} executed successfully')
+        
+        logger.info(f'All post-initial-replication commands executed successfully')

tests/test_mysql_ch_replicator.py

@@ -956,3 +956,149 @@ def test_resume_initial_replication_with_ignore_deletes():
     finally:
         # Clean up temp config file
         os.unlink(config_file)
+
+
+def test_post_initial_replication_commands():
+    config_file = 'tests/tests_config_post_commands.yaml'
+
+    cfg = config.Settings()
+    cfg.load(config_file)
+
+    mysql = mysql_api.MySQLApi(
+        database=None,
+        mysql_settings=cfg.mysql,
+    )
+
+    ch = clickhouse_api.ClickhouseApi(
+        database=TEST_DB_NAME,
+        clickhouse_settings=cfg.clickhouse,
+    )
+
+    prepare_env(cfg, mysql, ch)
+
+    mysql.execute(f"""
+CREATE TABLE `{TEST_TABLE_NAME}` (
+    id int(11) NOT NULL AUTO_INCREMENT,
+    event_time DATETIME NOT NULL,
+    event_type VARCHAR(50) NOT NULL,
+    PRIMARY KEY (id)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
+""")
+
+    initial_inserts = [
+        ('2024-01-01', 'type_0'),
+        ('2024-01-02', 'type_1'),
+        ('2024-01-03', 'type_2'),
+        ('2024-01-01', 'type_0'),
+        ('2024-01-02', 'type_1'),
+        ('2024-01-03', 'type_2'),
+        ('2024-01-01', 'type_0'),
+        ('2024-01-02', 'type_1'),
+        ('2024-01-03', 'type_2'),
+        ('2024-01-04', 'type_0'),
+    ]
+    
+    for date, event_type in initial_inserts:
+        mysql.execute(
+            f"INSERT INTO `{TEST_TABLE_NAME}` (event_time, event_type) VALUES ('{date} 10:00:00', '{event_type}');",
+            commit=True
+        )
+
+    binlog_replicator_runner = BinlogReplicatorRunner(cfg_file=config_file)
+    binlog_replicator_runner.run()
+
+    db_replicator_runner = DbReplicatorRunner(TEST_DB_NAME, cfg_file=config_file)
+    db_replicator_runner.run()
+
+    assert_wait(lambda: TEST_DB_NAME in ch.get_databases())
+    ch.execute_command(f'USE `{TEST_DB_NAME}`')
+    assert_wait(lambda: TEST_TABLE_NAME in ch.get_tables())
+    assert_wait(lambda: len(ch.select(TEST_TABLE_NAME)) == 10)
+
+    assert_wait(lambda: 'events_per_day' in ch.get_tables(), max_wait_time=15)
+    assert_wait(lambda: 'events_mv' in ch.get_tables(), max_wait_time=15)
+    
+    ch.execute_command('OPTIMIZE TABLE events_per_day FINAL')
+    
+    def check_initial_data_ready():
+        records = ch.select('events_per_day ORDER BY event_date, event_type')
+        return len(records) > 0
+    
+    assert_wait(check_initial_data_ready, max_wait_time=10)
+    
+    events_per_day_records = ch.select('events_per_day ORDER BY event_date, event_type')
+    assert len(events_per_day_records) > 0, "events_per_day should have aggregated data from initial replication backfill"
+    
+    aggregated_data = {}
+    for record in events_per_day_records:
+        date_str = str(record['event_date'])
+        event_type = record['event_type']
+        count = record['total_events']
+        key = (date_str, event_type)
+        aggregated_data[key] = count
+    
+    assert ('2024-01-01', 'type_0') in aggregated_data, "Should have 2024-01-01 + type_0"
+    assert aggregated_data[('2024-01-01', 'type_0')] == 3, f"Expected 3 events for 2024-01-01 + type_0, got {aggregated_data[('2024-01-01', 'type_0')]}"
+    
+    assert ('2024-01-02', 'type_1') in aggregated_data, "Should have 2024-01-02 + type_1"
+    assert aggregated_data[('2024-01-02', 'type_1')] == 3, f"Expected 3 events for 2024-01-02 + type_1, got {aggregated_data[('2024-01-02', 'type_1')]}"
+    
+    assert ('2024-01-03', 'type_2') in aggregated_data, "Should have 2024-01-03 + type_2"
+    assert aggregated_data[('2024-01-03', 'type_2')] == 3, f"Expected 3 events for 2024-01-03 + type_2, got {aggregated_data[('2024-01-03', 'type_2')]}"
+    
+    assert ('2024-01-04', 'type_0') in aggregated_data, "Should have 2024-01-04 + type_0"
+    assert aggregated_data[('2024-01-04', 'type_0')] == 1, f"Expected 1 event for 2024-01-04 + type_0, got {aggregated_data[('2024-01-04', 'type_0')]}"
+    
+    realtime_inserts = [
+        ('2024-01-05', 'type_new', 3),
+        ('2024-01-06', 'type_new', 2),
+        ('2024-01-01', 'type_0', 2),
+    ]
+    
+    for date, event_type, count in realtime_inserts:
+        for _ in range(count):
+            mysql.execute(
+                f"INSERT INTO `{TEST_TABLE_NAME}` (event_time, event_type) VALUES ('{date} 12:00:00', '{event_type}');",
+                commit=True
+            )
+    
+    assert_wait(lambda: len(ch.select(TEST_TABLE_NAME)) == 17)
+    
+    ch.execute_command('OPTIMIZE TABLE events_per_day FINAL')
+    
+    def check_realtime_aggregated():
+        records = ch.select('events_per_day')
+        agg = {}
+        for r in records:
+            key = (str(r['event_date']), r['event_type'])
+            agg[key] = r['total_events']
+        return (
+            agg.get(('2024-01-05', 'type_new'), 0) == 3 and
+            agg.get(('2024-01-06', 'type_new'), 0) == 2 and
+            agg.get(('2024-01-01', 'type_0'), 0) == 5
+        )
+    
+    assert_wait(check_realtime_aggregated, max_wait_time=15)
+    
+    ch.execute_command('OPTIMIZE TABLE events_per_day FINAL')
+    
+    def check_final_aggregated():
+        final_records = ch.select('events_per_day ORDER BY event_date, event_type')
+        final_aggregated_data = {}
+        for record in final_records:
+            date_str = str(record['event_date'])
+            event_type = record['event_type']
+            count = record['total_events']
+            key = (date_str, event_type)
+            final_aggregated_data[key] = count
+        
+        return (
+            final_aggregated_data.get(('2024-01-05', 'type_new')) == 3 and
+            final_aggregated_data.get(('2024-01-06', 'type_new')) == 2 and
+            final_aggregated_data.get(('2024-01-01', 'type_0')) == 5
+        )
+    
+    assert_wait(check_final_aggregated, max_wait_time=15)
+
+    db_replicator_runner.stop()
+    binlog_replicator_runner.stop()

tests/tests_config_post_commands.yaml

@@ -0,0 +1,44 @@
+mysql:
+  host: 'localhost'
+  port: 9306
+  user: 'root'
+  password: 'admin'
+
+clickhouse:
+  host: 'localhost'
+  port: 9123
+  user: 'default'
+  password: 'admin'
+  erase_batch_size: 2
+
+binlog_replicator:
+  data_dir: '/app/binlog/'
+  records_per_file: 100000
+  binlog_retention_period: 43200
+
+databases: '*test*'
+log_level: 'info'
+optimize_interval: 3
+check_db_updated_interval: 3
+
+target_databases:
+  replication-test_db_2: replication-destination
+
+indexes:
+  - databases: '*'
+    tables: ['group']
+    index: 'INDEX name_idx name TYPE ngrambf_v1(5, 65536, 4, 0) GRANULARITY 1'
+
+http_host: 'localhost'
+http_port: 9128
+
+types_mapping:
+  'char(36)': 'UUID'
+
+post_initial_replication_commands:
+  - databases: '*'
+    commands:
+      - 'CREATE TABLE IF NOT EXISTS events_per_day (event_date Date, event_type String, total_events UInt64) ENGINE = SummingMergeTree() ORDER BY (event_date, event_type)'
+      - 'CREATE MATERIALIZED VIEW IF NOT EXISTS events_mv TO events_per_day AS SELECT toDate(event_time) AS event_date, event_type, count() AS total_events FROM test_table GROUP BY event_date, event_type'
+      - 'INSERT INTO events_per_day SELECT toDate(event_time) AS event_date, event_type, count() AS total_events FROM test_table GROUP BY event_date, event_type'
+