@@ -2454,3 +2454,145 @@ def test_dynamic_column_addition_user_config():
24542454 db_pid = get_db_replicator_pid (cfg , "test_replication" )
24552455 if db_pid :
24562456 kill_process (db_pid )
2457+
2458+
2459+ def test_ignore_deletes ():
2460+ # Create a temporary config file with ignore_deletes=True
2461+ with tempfile .NamedTemporaryFile (mode = 'w' , suffix = '.yaml' , delete = False ) as temp_config_file :
2462+ config_file = temp_config_file .name
2463+
2464+ # Read the original config
2465+ with open (CONFIG_FILE , 'r' ) as original_config :
2466+ config_data = yaml .safe_load (original_config )
2467+
2468+ # Add ignore_deletes=True
2469+ config_data ['ignore_deletes' ] = True
2470+
2471+ # Write to the temp file
2472+ yaml .dump (config_data , temp_config_file )
2473+
2474+ try :
2475+ cfg = config .Settings ()
2476+ cfg .load (config_file )
2477+
2478+ # Verify the ignore_deletes option was set
2479+ assert cfg .ignore_deletes is True
2480+
2481+ mysql = mysql_api .MySQLApi (
2482+ database = None ,
2483+ mysql_settings = cfg .mysql ,
2484+ )
2485+
2486+ ch = clickhouse_api .ClickhouseApi (
2487+ database = TEST_DB_NAME ,
2488+ clickhouse_settings = cfg .clickhouse ,
2489+ )
2490+
2491+ prepare_env (cfg , mysql , ch )
2492+
2493+ # Create a table with a composite primary key
2494+ mysql .execute (f'''
2495+ CREATE TABLE `{ TEST_TABLE_NAME }
2496+ departments int(11) NOT NULL,
2497+ termine int(11) NOT NULL,
2498+ data varchar(255) NOT NULL,
2499+ PRIMARY KEY (departments,termine)
2500+ )
2501+ ''' )
2502+
2503+ # Insert initial records
2504+ mysql .execute (f"INSERT INTO `{ TEST_TABLE_NAME } , commit = True )
2505+ mysql .execute (f"INSERT INTO `{ TEST_TABLE_NAME } , commit = True )
2506+ mysql .execute (f"INSERT INTO `{ TEST_TABLE_NAME } , commit = True )
2507+
2508+ # Run the replicator with ignore_deletes=True
2509+ run_all_runner = RunAllRunner (cfg_file = config_file )
2510+ run_all_runner .run ()
2511+
2512+ # Wait for replication to complete
2513+ assert_wait (lambda : TEST_DB_NAME in ch .get_databases ())
2514+ ch .execute_command (f'USE `{ TEST_DB_NAME } )
2515+ assert_wait (lambda : TEST_TABLE_NAME in ch .get_tables ())
2516+ assert_wait (lambda : len (ch .select (TEST_TABLE_NAME )) == 3 )
2517+
2518+ # Delete some records from MySQL
2519+ mysql .execute (f"DELETE FROM `{ TEST_TABLE_NAME } , commit = True )
2520+ mysql .execute (f"DELETE FROM `{ TEST_TABLE_NAME } , commit = True )
2521+
2522+ # Wait a moment to ensure replication processes the events
2523+ time .sleep (5 )
2524+
2525+ # Verify records are NOT deleted in ClickHouse (since ignore_deletes=True)
2526+ # The count should still be 3
2527+ assert len (ch .select (TEST_TABLE_NAME )) == 3 , "Deletions were processed despite ignore_deletes=True"
2528+
2529+ # Insert a new record and verify it's added
2530+ mysql .execute (f"INSERT INTO `{ TEST_TABLE_NAME } , commit = True )
2531+ assert_wait (lambda : len (ch .select (TEST_TABLE_NAME )) == 4 )
2532+
2533+ # Verify the new record is correctly added
2534+ result = ch .select (TEST_TABLE_NAME , where = "departments=70 AND termine=80" )
2535+ assert len (result ) == 1
2536+ assert result [0 ]['data' ] == 'data4'
2537+
2538+ # Clean up
2539+ run_all_runner .stop ()
2540+
2541+ # Verify no errors occurred
2542+ assert_wait (lambda : 'stopping db_replicator' in read_logs (TEST_DB_NAME ))
2543+ assert ('Traceback' not in read_logs (TEST_DB_NAME ))
2544+
2545+ # Additional tests for persistence after restart
2546+
2547+ # 1. Remove all entries from table in MySQL
2548+ mysql .execute (f"DELETE FROM `{ TEST_TABLE_NAME } , commit = True )
2549+
2550+ # Add a new row in MySQL before starting the replicator
2551+ mysql .execute (f"INSERT INTO `{ TEST_TABLE_NAME } , commit = True )
2552+
2553+ # 2. Wait 5 seconds
2554+ time .sleep (5 )
2555+
2556+ # 3. Remove binlog directory (similar to prepare_env, but without removing tables)
2557+ if os .path .exists (cfg .binlog_replicator .data_dir ):
2558+ shutil .rmtree (cfg .binlog_replicator .data_dir )
2559+ os .mkdir (cfg .binlog_replicator .data_dir )
2560+
2561+
2562+ # 4. Create and run a new runner
2563+ new_runner = RunAllRunner (cfg_file = config_file )
2564+ new_runner .run ()
2565+
2566+ # 5. Ensure it has all the previous data (should still be 4 records from before + 1 new offline record)
2567+ assert_wait (lambda : TEST_DB_NAME in ch .get_databases ())
2568+ ch .execute_command (f'USE `{ TEST_DB_NAME } )
2569+ assert_wait (lambda : TEST_TABLE_NAME in ch .get_tables ())
2570+ assert_wait (lambda : len (ch .select (TEST_TABLE_NAME )) == 5 )
2571+
2572+ # Verify we still have all the old data
2573+ assert len (ch .select (TEST_TABLE_NAME , where = "departments=10 AND termine=20" )) == 1
2574+ assert len (ch .select (TEST_TABLE_NAME , where = "departments=30 AND termine=40" )) == 1
2575+ assert len (ch .select (TEST_TABLE_NAME , where = "departments=50 AND termine=60" )) == 1
2576+ assert len (ch .select (TEST_TABLE_NAME , where = "departments=70 AND termine=80" )) == 1
2577+
2578+ # Verify the offline data was replicated
2579+ assert len (ch .select (TEST_TABLE_NAME , where = "departments=110 AND termine=120" )) == 1
2580+ offline_data = ch .select (TEST_TABLE_NAME , where = "departments=110 AND termine=120" )[0 ]
2581+ assert offline_data ['data' ] == 'offline_data'
2582+
2583+ # 6. Insert new data and verify it gets added to existing data
2584+ mysql .execute (f"INSERT INTO `{ TEST_TABLE_NAME } , commit = True )
2585+ assert_wait (lambda : len (ch .select (TEST_TABLE_NAME )) == 6 )
2586+
2587+ # Verify the combined old and new data
2588+ result = ch .select (TEST_TABLE_NAME , where = "departments=90 AND termine=100" )
2589+ assert len (result ) == 1
2590+ assert result [0 ]['data' ] == 'data5'
2591+
2592+ # Make sure we have all 6 records (4 original + 1 offline + 1 new one)
2593+ assert len (ch .select (TEST_TABLE_NAME )) == 6
2594+
2595+ new_runner .stop ()
2596+ finally :
2597+ # Clean up the temporary config file
2598+ os .unlink (config_file )
0 commit comments