Added performance tests (bakwc#82)

Author: bakwc

README.md

@@ -18,7 +18,7 @@ With a focus on high performance, it utilizes batching heavily and uses C++ exte
 ## Features
 
 - **Real-Time Replication**: Keeps your ClickHouse database in sync with MySQL in real-time.
-- **High Performance**: Utilizes batching and ports slow parts to C++ (e.g., MySQL internal JSON parsing) for optimal performance.
+- **High Performance**: Utilizes batching and ports slow parts to C++ (e.g., MySQL internal JSON parsing) for optimal performance (±20K events / second on a single core).
 - **Supports Migrations/Schema Changes**: Handles adding, altering, and removing tables without breaking the replication process.
 - **Recovery without Downtime**: Allows for preserving old data while performing initial replication, ensuring continuous operation.
 - **Correct Data Removal**: Unlike MaterializedMySQL, `mysql_ch_replicator` ensures physical removal of data.

conftest.py

@@ -0,0 +1,34 @@
+# conftest.py
+import pytest
+
+
+def pytest_addoption(parser):
+    parser.addoption(
+        "--run-optional",
+        action="store_true",
+        default=False,
+        help="Run tests marked as optional",
+    )
+
+
+def pytest_collection_modifyitems(config, items):
+    run_optional = config.getoption("--run-optional")
+    keyword = config.getoption("keyword")  # Retrieves the value passed with -k
+
+    selected_tests = set()
+
+    if keyword:
+        # Collect nodeids of tests that match the -k keyword expression
+        for item in items:
+            if keyword in item.name or keyword in item.nodeid:
+                selected_tests.add(item.nodeid)
+
+    for item in items:
+        if "optional" in item.keywords:
+            if run_optional or item.nodeid in selected_tests:
+                # Do not skip if --run-optional is set or if the test matches the -k expression
+                continue
+            else:
+                # Skip the test
+                skip_marker = pytest.mark.skip(reason="Optional test, use --run-optional to include")
+                item.add_marker(skip_marker)

mysql_ch_replicator/binlog_replicator.py

@@ -473,6 +473,9 @@ def run(self):
 
                     self.data_writer.store_event(log_event)
 
+                    if last_read_count > 1000:
+                        break
+
                 self.update_state_if_required(last_transaction_id)
                 self.clear_old_binlog_if_required()
                 #print("last read count", last_read_count)

pytest.ini

@@ -0,0 +1,3 @@
+[pytest]
+markers =
+    optional: mark test as optional to run by default

test_mysql_ch_replicator.py

@@ -10,7 +10,7 @@
 from mysql_ch_replicator import config
 from mysql_ch_replicator import mysql_api
 from mysql_ch_replicator import clickhouse_api
-from mysql_ch_replicator.binlog_replicator import State as BinlogState
+from mysql_ch_replicator.binlog_replicator import State as BinlogState, FileReader, EventType
 from mysql_ch_replicator.db_replicator import State as DbReplicatorState, DbReplicator
 from mysql_ch_replicator.converter import MysqlToClickhouseConverter
 
@@ -1269,3 +1269,127 @@ def test_parse_mysql_table_structure():
 
     assert structure.table_name == 'user_preferences_portal'
 
+
+def get_last_file(directory, extension='.bin'):
+    max_num = -1
+    last_file = None
+    ext_len = len(extension)
+
+    with os.scandir(directory) as it:
+        for entry in it:
+            if entry.is_file() and entry.name.endswith(extension):
+                # Extract the numerical part by removing the extension
+                num_part = entry.name[:-ext_len]
+                try:
+                    num = int(num_part)
+                    if num > max_num:
+                        max_num = num
+                        last_file = entry.name
+                except ValueError:
+                    # Skip files where the name before extension is not an integer
+                    continue
+    return last_file
+
+
+def get_last_insert_from_binlog(cfg: config.Settings, db_name: str):
+    binlog_dir_path = os.path.join(cfg.binlog_replicator.data_dir, db_name)
+    if not os.path.exists(binlog_dir_path):
+        return None
+    last_file = get_last_file(binlog_dir_path)
+    if last_file is None:
+        return None
+    reader = FileReader(os.path.join(binlog_dir_path, last_file))
+    last_insert = None
+    while True:
+        event = reader.read_next_event()
+        if event is None:
+            break
+        if event.event_type != EventType.ADD_EVENT.value:
+            continue
+        for record in event.records:
+            last_insert = record
+    return last_insert
+
+
+@pytest.mark.optional
+def test_performance_dbreplicator():
+    config_file = 'tests_config_perf.yaml'
+    num_records = 100000
+
+    cfg = config.Settings()
+    cfg.load(config_file)
+
+    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)
+
+    mysql.execute(f'''
+    CREATE TABLE {TEST_TABLE_NAME} (
+        id int NOT NULL AUTO_INCREMENT,
+        name varchar(2048),
+        age int,
+        PRIMARY KEY (id)
+    ); 
+        ''')
+
+    binlog_replicator_runner = BinlogReplicatorRunner(cfg_file=config_file)
+    binlog_replicator_runner.run()
+
+    time.sleep(1)
+
+    mysql.execute(f"INSERT INTO {TEST_TABLE_NAME} (name, age) VALUES ('TEST_VALUE_1', 33);", commit=True)
+
+    def _get_last_insert_name():
+        record = get_last_insert_from_binlog(cfg=cfg, db_name=TEST_DB_NAME)
+        if record is None:
+            return None
+        return record[1].decode('utf-8')
+
+    assert_wait(lambda: _get_last_insert_name() == 'TEST_VALUE_1', retry_interval=0.5)
+
+    binlog_replicator_runner.stop()
+
+    time.sleep(1)
+
+    print("populating mysql data")
+
+    base_value = 'a' * 2000
+
+    for i in range(num_records):
+        if i % 2000 == 0:
+            print(f'populated {i} elements')
+        mysql.execute(
+            f"INSERT INTO {TEST_TABLE_NAME} (name, age) "
+            f"VALUES ('TEST_VALUE_{i}_{base_value}', {i});", commit=i % 20 == 0,
+        )
+
+    mysql.execute(f"INSERT INTO {TEST_TABLE_NAME} (name, age) VALUES ('TEST_VALUE_FINAL', 0);", commit=True)
+
+    print("running db_replicator")
+    t1 = time.time()
+    binlog_replicator_runner = BinlogReplicatorRunner(cfg_file=config_file)
+    binlog_replicator_runner.run()
+
+    assert_wait(lambda: _get_last_insert_name() == 'TEST_VALUE_FINAL', retry_interval=0.5, max_wait_time=1000)
+    t2 = time.time()
+
+    binlog_replicator_runner.stop()
+
+    time_delta = t2 - t1
+    rps = num_records / time_delta
+
+    print('\n\n')
+    print("*****************************")
+    print("records per second:", int(rps))
+    print("total time (seconds):", round(time_delta, 2))
+    print("*****************************")
+    print('\n\n')
+

tests_config_perf.yaml

@@ -0,0 +1,21 @@
+
+mysql:
+  host: 'localhost'
+  port: 9307
+  user: 'root'
+  password: 'admin'
+
+clickhouse:
+  host: 'localhost'
+  port: 9123
+  user: 'default'
+  password: 'admin'
+
+binlog_replicator:
+  data_dir: '/root/binlog/'
+  records_per_file: 1000
+
+databases: '*test*'
+log_level: 'info'
+optimize_interval: 3
+check_db_updated_interval: 3