Fixed resume of initial replication with ignore_deletes option (#173)

Author: bakwc

mysql_ch_replicator/config.py

@@ -107,6 +107,7 @@ class Settings:
     DEFAULT_OPTIMIZE_INTERVAL = 86400
     DEFAULT_CHECK_DB_UPDATED_INTERVAL = 120
     DEFAULT_AUTO_RESTART_INTERVAL = 3600
+    DEFAULT_INITIAL_REPLICATION_BATCH_SIZE = 50000
 
     def __init__(self):
         self.mysql = MysqlSettings()
@@ -131,6 +132,7 @@ def __init__(self):
         self.initial_replication_threads = 0
         self.ignore_deletes = False
         self.mysql_timezone = 'UTC'
+        self.initial_replication_batch_size = 50000
 
     def load(self, settings_file):
         data = open(settings_file, 'r').read()
@@ -158,6 +160,7 @@ def load(self, settings_file):
         self.initial_replication_threads = data.pop('initial_replication_threads', 0)
         self.ignore_deletes = data.pop('ignore_deletes', False)
         self.mysql_timezone = data.pop('mysql_timezone', 'UTC')
+        self.initial_replication_batch_size = data.pop('initial_replication_batch_size', Settings.DEFAULT_INITIAL_REPLICATION_BATCH_SIZE)
 
         indexes = data.pop('indexes', [])
         for index in indexes:

mysql_ch_replicator/db_replicator.py

@@ -88,13 +88,14 @@ def remove(self):
 
 class DbReplicator:
     def __init__(self, config: Settings, database: str, target_database: str = None, initial_only: bool = False, 
-                 worker_id: int = None, total_workers: int = None, table: str = None):
+                 worker_id: int = None, total_workers: int = None, table: str = None, initial_replication_test_fail_records: int = None):
         self.config = config
         self.database = database
         self.worker_id = worker_id
         self.total_workers = total_workers
         self.settings_file = config.settings_file
         self.single_table = table  # Store the single table to process
+        self.initial_replication_test_fail_records = initial_replication_test_fail_records  # Test flag for early exit
 
         # use same as source database by default
         self.target_database = database
@@ -143,6 +144,11 @@ def __init__(self, config: Settings, database: str, target_database: str = None,
         self.target_database_tmp = self.target_database + '_tmp'
         if self.is_parallel_worker:
             self.target_database_tmp = self.target_database
+        
+        # If ignore_deletes is enabled, we replicate directly into the target DB
+        # This must be set here to ensure consistency between first run and resume
+        if self.config.ignore_deletes:
+            self.target_database_tmp = self.target_database
 
         self.mysql_api = MySQLApi(
             database=self.database,
@@ -205,7 +211,6 @@ def run(self):
             if self.config.ignore_deletes:
                 logger.info(f'using existing database (ignore_deletes=True)')
                 self.clickhouse_api.database = self.target_database
-                self.target_database_tmp = self.target_database
 
                 # Create database if it doesn't exist
                 if self.target_database not in self.clickhouse_api.get_databases():

mysql_ch_replicator/db_replicator_initial.py

@@ -20,7 +20,6 @@
 
 class DbReplicatorInitial:
 
-    INITIAL_REPLICATION_BATCH_SIZE = 50000
     SAVE_STATE_INTERVAL = 10
     BINLOG_TOUCH_INTERVAL = 120
 
@@ -180,7 +179,7 @@ def perform_initial_replication_table(self, table_name):
             records = self.replicator.mysql_api.get_records(
                 table_name=table_name,
                 order_by=primary_keys,
-                limit=self.INITIAL_REPLICATION_BATCH_SIZE,
+                limit=self.replicator.config.initial_replication_batch_size,
                 start_value=query_start_values,
                 worker_id=self.replicator.worker_id,
                 total_workers=self.replicator.total_workers,
@@ -207,6 +206,16 @@ def perform_initial_replication_table(self, table_name):
             self.prevent_binlog_removal()
 
             stats_number_of_records += len(records)
+            
+            # Test flag: Exit early if we've replicated enough records for testing
+            if (self.replicator.initial_replication_test_fail_records is not None and 
+                stats_number_of_records >= self.replicator.initial_replication_test_fail_records):
+                logger.info(
+                    f'TEST MODE: Exiting initial replication after {stats_number_of_records} records '
+                    f'(limit: {self.replicator.initial_replication_test_fail_records})'
+                )
+                return
+            
             curr_time = time.time()
             if curr_time - last_stats_dump_time >= 60.0:
                 last_stats_dump_time = curr_time

mysql_ch_replicator/main.py

@@ -109,6 +109,7 @@ def run_db_replicator(args, config: Settings):
         worker_id=args.worker_id,
         total_workers=args.total_workers,
         table=args.table,
+        initial_replication_test_fail_records=getattr(args, 'initial_replication_test_fail_records', None),
     )
     db_replicator.run()
 
@@ -169,6 +170,10 @@ def main():
         "--table", type=str, default=None,
         help="Specific table to process (used with --worker_id for parallel processing of a single table)",
     )
+    parser.add_argument(
+        "--initial-replication-test-fail-records", type=int, default=None,
+        help="FOR TESTING ONLY: Exit initial replication after processing this many records",
+    )
     args = parser.parse_args()
 
     config = Settings()

test_mysql_ch_replicator.py

@@ -1179,10 +1179,8 @@ def test_json():
 
 
 def test_string_primary_key(monkeypatch):
-    monkeypatch.setattr(DbReplicatorInitial, 'INITIAL_REPLICATION_BATCH_SIZE', 1)
-
     cfg = config.Settings()
-    cfg.load(CONFIG_FILE)
+    cfg.load('tests_config_string_primary_key.yaml')
 
     mysql = mysql_api.MySQLApi(
         database=None,
@@ -1217,9 +1215,9 @@ def test_string_primary_key(monkeypatch):
         commit=True,
     )
 
-    binlog_replicator_runner = BinlogReplicatorRunner()
+    binlog_replicator_runner = BinlogReplicatorRunner(cfg_file='tests_config_string_primary_key.yaml')
     binlog_replicator_runner.run()
-    db_replicator_runner = DbReplicatorRunner(TEST_DB_NAME)
+    db_replicator_runner = DbReplicatorRunner(TEST_DB_NAME, cfg_file='tests_config_string_primary_key.yaml')
     db_replicator_runner.run()
 
     assert_wait(lambda: TEST_DB_NAME in ch.get_databases())
@@ -1241,10 +1239,8 @@ def test_string_primary_key(monkeypatch):
 
 
 def test_if_exists_if_not_exists(monkeypatch):
-    monkeypatch.setattr(DbReplicatorInitial, 'INITIAL_REPLICATION_BATCH_SIZE', 1)
-
     cfg = config.Settings()
-    cfg.load(CONFIG_FILE)
+    cfg.load('tests_config_string_primary_key.yaml')
 
     mysql = mysql_api.MySQLApi(
         database=None,
@@ -1258,9 +1254,9 @@ def test_if_exists_if_not_exists(monkeypatch):
 
     prepare_env(cfg, mysql, ch)
 
-    binlog_replicator_runner = BinlogReplicatorRunner()
+    binlog_replicator_runner = BinlogReplicatorRunner(cfg_file='tests_config_string_primary_key.yaml')
     binlog_replicator_runner.run()
-    db_replicator_runner = DbReplicatorRunner(TEST_DB_NAME)
+    db_replicator_runner = DbReplicatorRunner(TEST_DB_NAME, cfg_file='tests_config_string_primary_key.yaml')
     db_replicator_runner.run()
 
     assert_wait(lambda: TEST_DB_NAME in ch.get_databases())
@@ -1282,10 +1278,8 @@ def test_if_exists_if_not_exists(monkeypatch):
 
 
 def test_percona_migration(monkeypatch):
-    monkeypatch.setattr(DbReplicatorInitial, 'INITIAL_REPLICATION_BATCH_SIZE', 1)
-
     cfg = config.Settings()
-    cfg.load(CONFIG_FILE)
+    cfg.load('tests_config_string_primary_key.yaml')
 
     mysql = mysql_api.MySQLApi(
         database=None,
@@ -1310,9 +1304,9 @@ def test_percona_migration(monkeypatch):
         commit=True,
     )
 
-    binlog_replicator_runner = BinlogReplicatorRunner()
+    binlog_replicator_runner = BinlogReplicatorRunner(cfg_file='tests_config_string_primary_key.yaml')
     binlog_replicator_runner.run()
-    db_replicator_runner = DbReplicatorRunner(TEST_DB_NAME)
+    db_replicator_runner = DbReplicatorRunner(TEST_DB_NAME, cfg_file='tests_config_string_primary_key.yaml')
     db_replicator_runner.run()
 
     assert_wait(lambda: TEST_DB_NAME in ch.get_databases())
@@ -1360,10 +1354,8 @@ def test_percona_migration(monkeypatch):
 
 
 def test_add_column_first_after_and_drop_column(monkeypatch):
-    monkeypatch.setattr(DbReplicatorInitial, 'INITIAL_REPLICATION_BATCH_SIZE', 1)
-
     cfg = config.Settings()
-    cfg.load(CONFIG_FILE)
+    cfg.load('tests_config_string_primary_key.yaml')
 
     mysql = mysql_api.MySQLApi(
         database=None,
@@ -1388,9 +1380,9 @@ def test_add_column_first_after_and_drop_column(monkeypatch):
         commit=True,
     )
 
-    binlog_replicator_runner = BinlogReplicatorRunner()
+    binlog_replicator_runner = BinlogReplicatorRunner(cfg_file='tests_config_string_primary_key.yaml')
     binlog_replicator_runner.run()
-    db_replicator_runner = DbReplicatorRunner(TEST_DB_NAME)
+    db_replicator_runner = DbReplicatorRunner(TEST_DB_NAME, cfg_file='tests_config_string_primary_key.yaml')
     db_replicator_runner.run()
 
     assert_wait(lambda: TEST_DB_NAME in ch.get_databases())
@@ -2912,3 +2904,124 @@ def test_timezone_conversion():
     finally:
         # Clean up temporary config file
         os.unlink(temp_config_file)
+
+def test_resume_initial_replication_with_ignore_deletes():
+    """
+    Test that resuming initial replication works correctly with ignore_deletes=True.
+    
+    This reproduces the bug from https://github.com/bakwc/mysql_ch_replicator/issues/172
+    where resuming initial replication would fail with "Database sirocco_tmp does not exist"
+    when ignore_deletes=True because the code would try to use the _tmp database instead
+    of the target database directly.
+    """
+    # Create a temporary config file with ignore_deletes=True
+    with tempfile.NamedTemporaryFile(mode='w', suffix='.yaml', delete=False) as temp_config_file:
+        config_file = temp_config_file.name
+        
+        # Read the original config
+        with open(CONFIG_FILE, 'r') as original_config:
+            config_data = yaml.safe_load(original_config)
+        
+        # Add ignore_deletes=True
+        config_data['ignore_deletes'] = True
+        
+        # Set initial_replication_batch_size to 1 for testing
+        config_data['initial_replication_batch_size'] = 1
+        
+        # Write to the temp file
+        yaml.dump(config_data, temp_config_file)
+
+    try:
+        cfg = config.Settings()
+        cfg.load(config_file)
+        
+        # Verify the ignore_deletes option was set
+        assert cfg.ignore_deletes is True
+
+        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)
+
+        # Create a table with many records to ensure initial replication takes time
+        mysql.execute(f'''
+        CREATE TABLE `{TEST_TABLE_NAME}` (
+            id int NOT NULL AUTO_INCREMENT,
+            name varchar(255),
+            data varchar(1000),
+            PRIMARY KEY (id)
+        )
+        ''')
+
+        # Insert many records to make initial replication take longer
+        for i in range(100):
+            mysql.execute(
+                f"INSERT INTO `{TEST_TABLE_NAME}` (name, data) VALUES ('test_{i}', 'data_{i}');",
+                commit=True
+            )
+
+        # Start binlog replicator
+        binlog_replicator_runner = BinlogReplicatorRunner(cfg_file=config_file)
+        binlog_replicator_runner.run()
+
+        # Start db replicator for initial replication with test flag to exit early
+        db_replicator_runner = DbReplicatorRunner(TEST_DB_NAME, cfg_file=config_file, 
+                                                 additional_arguments='--initial-replication-test-fail-records 30')
+        db_replicator_runner.run()
+        
+        # Wait for initial replication to start
+        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())
+        
+        # Wait for some records to be replicated but not all (should hit the 30 record limit)
+        assert_wait(lambda: len(ch.select(TEST_TABLE_NAME)) > 0)
+        
+        # The db replicator should have stopped automatically due to the test flag
+        # But we still call stop() to ensure proper cleanup
+        db_replicator_runner.stop()
+        
+        # Verify the state is still PERFORMING_INITIAL_REPLICATION
+        state_path = os.path.join(cfg.binlog_replicator.data_dir, TEST_DB_NAME, 'state.pckl')
+        state = DbReplicatorState(state_path)
+        assert state.status.value == 2  # PERFORMING_INITIAL_REPLICATION
+        
+        # Add more records while replication is stopped
+        for i in range(100, 150):
+            mysql.execute(
+                f"INSERT INTO `{TEST_TABLE_NAME}` (name, data) VALUES ('test_{i}', 'data_{i}');",
+                commit=True
+            )
+
+        # Verify that sirocco_tmp database does NOT exist (it should use sirocco directly)
+        assert f"{TEST_DB_NAME}_tmp" not in ch.get_databases(), "Temporary database should not exist with ignore_deletes=True"
+        
+        # Resume initial replication - this should NOT fail with "Database sirocco_tmp does not exist"
+        db_replicator_runner_2 = DbReplicatorRunner(TEST_DB_NAME, cfg_file=config_file)
+        db_replicator_runner_2.run()
+        
+        # Wait for all records to be replicated (100 original + 50 extra = 150)
+        assert_wait(lambda: len(ch.select(TEST_TABLE_NAME)) == 150, max_wait_time=30)
+        
+        # Verify the replication completed successfully
+        records = ch.select(TEST_TABLE_NAME)
+        assert len(records) == 150, f"Expected 150 records, got {len(records)}"
+        
+        # Verify we can continue with realtime replication
+        mysql.execute(f"INSERT INTO `{TEST_TABLE_NAME}` (name, data) VALUES ('realtime_test', 'realtime_data');", commit=True)
+        assert_wait(lambda: len(ch.select(TEST_TABLE_NAME)) == 151)
+        
+        # Clean up
+        db_replicator_runner_2.stop()
+        binlog_replicator_runner.stop()
+        
+    finally:
+        # Clean up temp config file
+        os.unlink(config_file)

tests_config_string_primary_key.yaml

@@ -0,0 +1,36 @@
+mysql:
+  host: 'localhost'
+  port: 9306
+  user: 'root'
+  password: 'admin'
+
+clickhouse:
+  host: 'localhost'
+  port: 9123
+  user: 'default'
+  password: 'admin'
+
+binlog_replicator:
+  data_dir: '/app/binlog/'
+  records_per_file: 100000
+  binlog_retention_period: 43200  # 12 hours in seconds
+
+databases: '*test*'
+log_level: 'debug'
+optimize_interval: 3
+check_db_updated_interval: 3
+initial_replication_batch_size: 1
+
+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'