When executing CREATE DATABASE database_name ENGINE = MaterializeMySQL(mysql_host:mysql_port, mysql_database, mysql_user, mysql_password).

• Step 1:
  We need to create the tables existing in MySQL in the ClickHouse and input the data at the same time.

  It does the following(one MySQL session):

  MySQL :> FLUSH TABLES;
  MySQL :> FLUSH TABLES WITH READ LOCK;
  MySQL :> SHOW MASTER STATUS; /* fetch binlog_file binlog_position*/
  MySQL :> SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ;
  MySQL :> START TRANSACTION /*!40100 WITH CONSISTENT SNAPSHOT */;
  MySQL :> SHOW CREATE TABLE ... /* each tables in mysql_database */;
  MySQL :> UNLOCK TABLES;
  
  ClickHouse :> CREATE TABLE dump_table_name(..., `_sign` Int8, `_version` UInt64) ENGINE = ReplacingMergeTree(_version) PARTITION primary_keys, ORDER BY primary_keys[not increment], keys[not increment], unique_keys[not increment], keys[increment], unique_keys[increment], primary_keys[increment]; /* dump table struct */
  ClickHouse :> INSERT INTO dump_table_name SELECT *, 1 AS _sign, 1 AS _version FROM mysql(mysql_host:mysql_port, mysql_database, dump_table_name, mysql_user, mysql_password); /* dump table data */
  
  MySQL :> COMMIT;

• Step 2:
  We need to consume MySQL binlog event in real time to pull real-time data( MySQL Replication Protocol )

  It does the following(one thread):

  • MYSQL_WRITE_ROWS_EVENT : write data to buffer and flush data when (bytes | rows | time > settings)
    • sign_column = 1, version_column = ++version(Initial value: UInt64(2))
  • MYSQL_DELETE_ROWS_EVENT: write data to buffer and flush data when (bytes | rows | time > settings)
    • sign_column = -1, version_column = ++version(Initial value: UInt64(2))
  • MYSQL_UPDATE_ROWS_EVENT: write data to buffer and flush data when (bytes | rows | time > settings)
    • When not modified order by expression columns
      • old data: no insert.
      • new data: sign_column = 1, version_column = ++version(Initial value: UInt64(2))
    • When modify order by expression columns
      • old data: sign_column = -1, version_column = ++version(Initial value: UInt64(2))
      • new data: sign_column = 1, version_column = ++version(Initial value: UInt64(2))
  • MYSQL_QUERY_EVENT: flush data, transform and execute the query
    • CREATE TABLE Query: Same as dump table structure
    • DROP TABLE Query: Same as ClickHouse drop table query
    • RENAME TABLE Query: Same as ClickHouse rename table query
    • ALTER TABLE Query: transform ADD_COLUMN, DROP_COLUMN, RENAME_COLUMN, MODIFY_COLUMN
    • Ignore other query.