Merge pull request #52067 from valbok/start-transaction

MaterializedMySQL: Support CREATE TABLE AS SELECT

src/Core/MySQL/MySQLReplication.cpp

@@ -121,6 +121,17 @@ namespace MySQLReplication
         {
             typ = QUERY_SAVEPOINT;
         }
+
+        // https://dev.mysql.com/worklog/task/?id=13355
+        // When doing query "CREATE TABLE xx AS SELECT", the binlog will be
+        // "CREATE TABLE ... START TRANSACTION", the DDL will be failed
+        // so, just ignore the "START TRANSACTION" suffix
+        if (query.ends_with("START TRANSACTION"))
+        {
+            auto pos = query.rfind("START TRANSACTION");
+            if (pos > 0)
+                query.resize(pos);
+        }
     }
 
     void QueryEvent::dump(WriteBuffer & out) const

tests/integration/test_materialized_mysql_database/materialized_with_ddl.py

@@ -2336,3 +2336,32 @@ def named_collections(clickhouse_node, mysql_node, service_name):
     )
     clickhouse_node.query(f"DROP DATABASE IF EXISTS {db}")
     mysql_node.query(f"DROP DATABASE IF EXISTS {db}")
+
+
+def create_table_as_select(clickhouse_node, mysql_node, service_name):
+    db = "create_table_as_select"
+    mysql_node.query(f"DROP DATABASE IF EXISTS {db}")
+    clickhouse_node.query(f"DROP DATABASE IF EXISTS {db}")
+    mysql_node.query(f"CREATE DATABASE {db}")
+    clickhouse_node.query(
+        f"CREATE DATABASE {db} ENGINE = MaterializeMySQL('{service_name}:3306', '{db}', 'root', 'clickhouse')"
+    )
+    mysql_node.query(
+        f"CREATE TABLE {db}.t1(a INT NOT NULL PRIMARY KEY) ENGINE = InnoDB"
+    )
+    mysql_node.query(f"INSERT INTO {db}.t1 VALUES (1)")
+    check_query(
+        clickhouse_node,
+        f"SHOW TABLES FROM {db} FORMAT TSV",
+        "t1\n",
+    )
+
+    mysql_node.query(f"CREATE TABLE {db}.t2(PRIMARY KEY(a)) AS SELECT * FROM {db}.t1")
+    check_query(
+        clickhouse_node,
+        f"SHOW TABLES FROM {db} FORMAT TSV",
+        "t1\nt2\n",
+    )
+
+    clickhouse_node.query(f"DROP DATABASE IF EXISTS {db}")
+    mysql_node.query(f"DROP DATABASE IF EXISTS {db}")

tests/integration/test_materialized_mysql_database/test.py

@@ -529,3 +529,9 @@ def test_named_collections(started_cluster, started_mysql_8_0, clickhouse_node):
     materialized_with_ddl.named_collections(
         clickhouse_node, started_mysql_8_0, "mysql80"
     )
+
+
+def test_create_table_as_select(started_cluster, started_mysql_8_0, clickhouse_node):
+    materialized_with_ddl.create_table_as_select(
+        clickhouse_node, started_mysql_8_0, "mysql80"
+    )