Added test reproducing crash

Author: bakwc

test_mysql_ch_replicator.py

@@ -2257,3 +2257,99 @@ def test_performance_initial_only_replication():
 
     # Clean up the temporary config file
     os.remove(parallel_config_file)
+
+
+def test_schema_evolution_with_db_mapping():
+    """Test case to reproduce issue where schema evolution doesn't work with database mapping."""
+    # Use the predefined config file with database mapping
+    config_file = "tests_config_db_mapping.yaml"
+    
+    cfg = config.Settings()
+    cfg.load(config_file)
+
+    # Note: Not setting a specific database in MySQL API
+    mysql = mysql_api.MySQLApi(
+        database=None,
+        mysql_settings=cfg.mysql,
+    )
+
+    ch = clickhouse_api.ClickhouseApi(
+        database="mapped_target_db",
+        clickhouse_settings=cfg.clickhouse,
+    )
+
+    prepare_env(cfg, mysql, ch, db_name=TEST_DB_NAME)
+
+    # Create a test table with some columns using fully qualified name
+    mysql.execute(f'''
+CREATE TABLE `{TEST_DB_NAME}`.`{TEST_TABLE_NAME}` (
+  `id` int NOT NULL,
+  `name` varchar(255) NOT NULL,
+  PRIMARY KEY (`id`)); 
+    ''')
+
+    mysql.execute(
+        f"INSERT INTO `{TEST_DB_NAME}`.`{TEST_TABLE_NAME}` (id, name) VALUES (1, 'Original')",
+        commit=True,
+    )
+
+    # Start the replication
+    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()
+
+    # Make sure initial replication works with the database mapping
+    assert_wait(lambda: "mapped_target_db" in ch.get_databases())
+    ch.execute_command(f'USE `mapped_target_db`')
+    assert_wait(lambda: TEST_TABLE_NAME in ch.get_tables())
+    assert_wait(lambda: len(ch.select(TEST_TABLE_NAME)) == 1)
+
+    # Now follow user's exact sequence of operations with fully qualified names
+    # 1. Add new column
+    mysql.execute(f"ALTER TABLE `{TEST_DB_NAME}`.`{TEST_TABLE_NAME}` ADD COLUMN added_new_column char(1)", commit=True)
+    
+    # 2. Rename the column
+    mysql.execute(f"ALTER TABLE `{TEST_DB_NAME}`.`{TEST_TABLE_NAME}` RENAME COLUMN added_new_column TO rename_column_name", commit=True)
+    
+    # 3. Modify column type
+    mysql.execute(f"ALTER TABLE `{TEST_DB_NAME}`.`{TEST_TABLE_NAME}` MODIFY rename_column_name varchar(5)", commit=True)
+    
+    # 4. Insert data using the modified schema
+    mysql.execute(
+        f"INSERT INTO `{TEST_DB_NAME}`.`{TEST_TABLE_NAME}` (id, name, rename_column_name) VALUES (2, 'Second', 'ABCDE')",
+        commit=True,
+    )
+    
+    # 5. Drop the column - this is where the error was reported
+    mysql.execute(f"ALTER TABLE `{TEST_DB_NAME}`.`{TEST_TABLE_NAME}` DROP COLUMN rename_column_name", commit=True)
+    
+    # 6. Add more inserts after schema changes to verify ongoing replication
+    mysql.execute(
+        f"INSERT INTO `{TEST_DB_NAME}`.`{TEST_TABLE_NAME}` (id, name) VALUES (3, 'Third record after drop column')",
+        commit=True,
+    )
+    
+    # Check if all changes were replicated correctly
+    time.sleep(5)  # Allow time for processing the changes
+    result = ch.select(TEST_TABLE_NAME)
+    print(f"ClickHouse table contents: {result}")
+    
+    # Verify all records are present
+    assert_wait(lambda: len(ch.select(TEST_TABLE_NAME)) == 3)
+    
+    # Verify specific records exist
+    records = ch.select(TEST_TABLE_NAME)
+    print(f"Record type: {type(records[0])}")  # Debug the record type
+    
+    # Access by field name 'id' instead of by position
+    record_ids = [record['id'] for record in records]
+    assert 1 in record_ids, "Original record (id=1) not found"
+    assert 3 in record_ids, "New record (id=3) after schema changes not found"
+    
+    # Note: This test will likely fail with "IndexError: list index out of range" 
+    # as reported by the user when using database mapping with schema evolution
+    
+    # Clean up
+    db_replicator_runner.stop()
+    binlog_replicator_runner.stop()

tests_config_db_mapping.yaml

@@ -0,0 +1,28 @@
+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
+
+# This mapping is the key part that causes issues with schema evolution
+target_databases:
+  replication-test_db: mapped_target_db
+
+http_host: 'localhost'
+http_port: 9128