Support ALTER CHANGE (#2)

* Fixed state save order (prevent skipping schema migration in case of exception)

* Support for ALTER table CHANGE column

Author: bakwc

docker-compose-tests.yaml

@@ -11,7 +11,10 @@ services:
       - CLICKHOUSE_ADMIN_PASSWORD=admin
       - CLICKHOUSE_ADMIN_USER=default
       - CLICKHOUSE_HTTP_PORT=9123
-    network_mode: host
+    networks:
+      default:
+    ports:
+      - 9123:9123
     volumes:
       - ./tests_override.xml:/bitnami/clickhouse/etc/conf.d/override.xml:ro
 

mysql_ch_replicator/converter.py

@@ -205,7 +205,11 @@ def convert_alter_query(self, mysql_query, db_name):
             if op_name == 'alter':
                 continue
 
-            raise Exception('not implement')
+            if op_name == 'change':
+                self.__convert_alter_table_change_column(db_name, table_name, tokens)
+                continue
+
+            raise Exception(f'operation {op_name} not implement, query: {subquery}')
 
     def __convert_alter_table_add_column(self, db_name, table_name, tokens):
         if len(tokens) < 2:
@@ -306,6 +310,51 @@ def __convert_alter_table_modify_column(self, db_name, table_name, tokens):
         if self.db_replicator:
             self.db_replicator.clickhouse_api.execute_command(query)
 
+    def __convert_alter_table_change_column(self, db_name, table_name, tokens):
+        if len(tokens) < 3:
+            raise Exception('wrong tokens count', tokens)
+
+        if ',' in ' '.join(tokens):
+            raise Exception('add multiple columns not implemented', tokens)
+
+        column_name = strip_sql_name(tokens[0])
+        new_column_name = strip_sql_name(tokens[1])
+        column_type_mysql = tokens[2]
+        column_type_mysql_parameters = ' '.join(tokens[3:])
+
+        column_type_ch = self.convert_field_type(column_type_mysql, column_type_mysql_parameters)
+
+        # update table structure
+        if self.db_replicator:
+            table_structure = self.db_replicator.state.tables_structure[table_name]
+            mysql_table_structure: TableStructure = table_structure[0]
+            ch_table_structure: TableStructure = table_structure[1]
+
+            current_column_type_ch = ch_table_structure.get_field(column_name).field_type
+
+            if current_column_type_ch != column_type_ch:
+
+                mysql_table_structure.update_field(
+                    TableField(name=column_name, field_type=column_type_mysql),
+                )
+
+                ch_table_structure.update_field(
+                    TableField(name=column_name, field_type=column_type_ch),
+                )
+
+                query = f'ALTER TABLE {db_name}.{table_name} MODIFY COLUMN {column_name} {column_type_ch}'
+                self.db_replicator.clickhouse_api.execute_command(query)
+
+            if column_name != new_column_name:
+                curr_field_mysql = mysql_table_structure.get_field(column_name)
+                curr_field_clickhouse = ch_table_structure.get_field(column_name)
+
+                curr_field_mysql.name = new_column_name
+                curr_field_clickhouse.name = new_column_name
+
+                query = f'ALTER TABLE {db_name}.{table_name} RENAME COLUMN {column_name} TO {new_column_name}'
+                self.db_replicator.clickhouse_api.execute_command(query)
+
     def parse_create_table_query(self, mysql_query) -> tuple:
         mysql_table_structure = self.parse_mysql_table_structure(mysql_query)
         ch_table_structure = self.convert_table_structure(mysql_table_structure)

mysql_ch_replicator/db_replicator.py

@@ -274,9 +274,6 @@ def handle_event(self, event: LogEvent):
                 return
 
         logger.debug(f'processing event {event.transaction_id}')
-        self.stats.events_count += 1
-        self.stats.last_transaction = event.transaction_id
-        self.state.last_processed_transaction_non_uploaded = event.transaction_id
 
         event_handlers = {
             EventType.ADD_EVENT.value: self.handle_insert_event,
@@ -286,6 +283,10 @@ def handle_event(self, event: LogEvent):
 
         event_handlers[event.event_type](event)
 
+        self.stats.events_count += 1
+        self.stats.last_transaction = event.transaction_id
+        self.state.last_processed_transaction_non_uploaded = event.transaction_id
+
         self.upload_records_if_required(table_name=event.table_name)
 
         self.save_state_if_required()

mysql_ch_replicator/table_structure.py

@@ -48,3 +48,9 @@ def has_field(self, field_name):
             if field.name == field_name:
                 return True
         return False
+
+    def get_field(self, field_name):
+        for field in self.fields:
+            if field.name == field_name:
+                return field
+        return None

test_mysql_ch_replicator.py

@@ -130,7 +130,19 @@ def test_e2e_regular():
         f"VALUES ('John', 12, 'Doe', 'USA');", commit=True,
     )
 
+    mysql.execute(
+        f"ALTER TABLE {TEST_DB_NAME}.{TEST_TABLE_NAME} "
+        f"CHANGE COLUMN country origin VARCHAR(24) DEFAULT '' NOT NULL",
+    )
+
     assert_wait(lambda: len(ch.select(TEST_TABLE_NAME)) == 5)
+    assert_wait(lambda: ch.select(TEST_TABLE_NAME, where="name='John'")[0].get('origin') == 'USA')
+
+    mysql.execute(
+        f"ALTER TABLE {TEST_DB_NAME}.{TEST_TABLE_NAME} "
+        f"CHANGE COLUMN origin country VARCHAR(24) DEFAULT '' NOT NULL",
+    )
+    assert_wait(lambda: ch.select(TEST_TABLE_NAME, where="name='John'")[0].get('origin') is None)
     assert_wait(lambda: ch.select(TEST_TABLE_NAME, where="name='John'")[0].get('country') == 'USA')
 
     mysql.execute(f"ALTER TABLE {TEST_DB_NAME}.{TEST_TABLE_NAME} DROP COLUMN country")