Merge pull request #7717 from MaxFedotov/support_mysql_federated

support clickhouse as mysql federated server

dbms/programs/server/MySQLHandler.cpp

@@ -15,6 +15,7 @@
 #include <IO/ReadBufferFromString.h>
 #include <IO/WriteBufferFromPocoSocket.h>
 #include <Storages/IStorage.h>
+#include <boost/algorithm/string/replace.hpp>
 
 #if USE_POCO_NETSSL
 #include <Poco/Net/SecureStreamSocket.h>
@@ -267,30 +268,50 @@ void MySQLHandler::comPing()
     packet_sender->sendPacket(OK_Packet(0x0, client_capability_flags, 0, 0, 0), true);
 }
 
+static bool isFederatedServerSetupCommand(const String & query);
+
 void MySQLHandler::comQuery(ReadBuffer & payload)
+{
+    String query = String(payload.position(), payload.buffer().end());
+
+    // This is a workaround in order to support adding ClickHouse to MySQL using federated server.
+    // As Clickhouse doesn't support these statements, we just send OK packet in response.
+    if (isFederatedServerSetupCommand(query))
+    {
+        packet_sender->sendPacket(OK_Packet(0x00, client_capability_flags, 0, 0, 0), true);
+    }
+    else
     {
         bool with_output = false;
         std::function<void(const String &)> set_content_type = [&with_output](const String &) -> void {
             with_output = true;
         };
 
-    const String query("select ''");
-    ReadBufferFromString empty_select(query);
-
+        String replacement_query = "select ''";
         bool should_replace = false;
+
         // Translate query from MySQL to ClickHouse.
         // This is a temporary workaround until ClickHouse supports the syntax "@@var_name".
-    if (std::string(payload.position(), payload.buffer().end()) == "select @@version_comment limit 1")  // MariaDB client starts session with that query
+        if (query == "select @@version_comment limit 1")  // MariaDB client starts session with that query
         {
             should_replace = true;
         }
+        // This is a workaround in order to support adding ClickHouse to MySQL using federated server.
+        if (0 == strncasecmp("SHOW TABLE STATUS LIKE", query.c_str(), 22))
+        {
+            should_replace = true;
+            replacement_query = boost::replace_all_copy(query, "SHOW TABLE STATUS LIKE ", show_table_status_replacement_query);
+        }
+
+        ReadBufferFromString replacement(replacement_query);
 
         Context query_context = connection_context;
-    executeQuery(should_replace ? empty_select : payload, *out, true, query_context, set_content_type, nullptr);
+        executeQuery(should_replace ? replacement : payload, *out, true, query_context, set_content_type, nullptr);
 
         if (!with_output)
             packet_sender->sendPacket(OK_Packet(0x00, client_capability_flags, 0, 0, 0), true);
     }
+}
 
 void MySQLHandler::authPluginSSL()
 {
@@ -335,4 +356,33 @@ void MySQLHandlerSSL::finishHandshakeSSL(size_t packet_size, char * buf, size_t
 
 #endif
 
+static bool isFederatedServerSetupCommand(const String & query)
+{
+    return 0 == strncasecmp("SET NAMES", query.c_str(), 9) || 0 == strncasecmp("SET character_set_results", query.c_str(), 25)
+        || 0 == strncasecmp("SET FOREIGN_KEY_CHECKS", query.c_str(), 22) || 0 == strncasecmp("SET AUTOCOMMIT", query.c_str(), 14)
+        || 0 == strncasecmp("SET SESSION TRANSACTION ISOLATION LEVEL", query.c_str(), 39);
+}
+
+const String MySQLHandler::show_table_status_replacement_query("SELECT"
+                                                               " name AS Name,"
+                                                               " engine AS Engine,"
+                                                               " '10' AS Version,"
+                                                               " 'Dynamic' AS Row_format,"
+                                                               " 0 AS Rows,"
+                                                               " 0 AS Avg_row_length,"
+                                                               " 0 AS Data_length,"
+                                                               " 0 AS Max_data_length,"
+                                                               " 0 AS Index_length,"
+                                                               " 0 AS Data_free,"
+                                                               " 'NULL' AS Auto_increment,"
+                                                               " metadata_modification_time AS Create_time,"
+                                                               " metadata_modification_time AS Update_time,"
+                                                               " metadata_modification_time AS Check_time,"
+                                                               " 'utf8_bin' AS Collation,"
+                                                               " 'NULL' AS Checksum,"
+                                                               " '' AS Create_options,"
+                                                               " '' AS Comment"
+                                                               " FROM system.tables"
+                                                               " WHERE name LIKE ");
+
 }

dbms/programs/server/MySQLHandler.h

@@ -11,7 +11,6 @@
 
 namespace DB
 {
-
 /// Handler for MySQL wire protocol connections. Allows to connect to ClickHouse using MySQL client.
 class MySQLHandler : public Poco::Net::TCPServerConnection
 {
@@ -59,6 +58,9 @@ protected:
     std::shared_ptr<WriteBuffer> out;
 
     bool secure_connection = false;
+
+private:
+    static const String show_table_status_replacement_query;
 };
 
 #if USE_SSL && USE_POCO_NETSSL

dbms/tests/integration/test_mysql_protocol/clients/mysql/docker_compose.yml

@@ -2,5 +2,7 @@ version: '2.2'
 services:
   mysql1:
     image: mysql:5.7
-    # rewriting default command, because starting server is unnecessary
-    command: sleep infinity
+    restart: always
+    environment:
+      MYSQL_ALLOW_EMPTY_PASSWORD: 1
+    command: --federated --socket /var/run/mysqld/mysqld.sock

dbms/tests/integration/test_mysql_protocol/test.py

@@ -108,6 +108,39 @@ def test_mysql_client(mysql_client, server_address):
 
     assert stdout == '\n'.join(['column', '0', '0', '1', '1', '5', '5', 'tmp_column', '0', '1', ''])
 
+def test_mysql_federated(mysql_client, server_address):
+    node.query('''DROP DATABASE IF EXISTS mysql_federated''', settings={"password": "123"})
+    node.query('''CREATE DATABASE mysql_federated''', settings={"password": "123"})
+    node.query('''CREATE TABLE mysql_federated.test (col UInt32) ENGINE = Log''', settings={"password": "123"})
+    node.query('''INSERT INTO mysql_federated.test VALUES (0), (1), (5)''', settings={"password": "123"})
+
+
+    code, (_, stderr) = mysql_client.exec_run('''
+        mysql
+        -e "DROP SERVER IF EXISTS clickhouse;"
+        -e "CREATE SERVER clickhouse FOREIGN DATA WRAPPER mysql OPTIONS (USER 'default', PASSWORD '123', HOST '{host}', PORT {port}, DATABASE 'mysql_federated');"
+        -e "DROP DATABASE IF EXISTS mysql_federated;"
+        -e "CREATE DATABASE mysql_federated;"
+    '''.format(host=server_address, port=server_port), demux=True)
+
+    assert code == 0
+
+    code, (stdout, stderr) = mysql_client.exec_run('''
+        mysql
+        -e "CREATE TABLE mysql_federated.test(`col` int UNSIGNED) ENGINE=FEDERATED CONNECTION='clickhouse';"
+        -e "SELECT * FROM mysql_federated.test ORDER BY col;"
+    '''.format(host=server_address, port=server_port), demux=True)
+
+    assert stdout == '\n'.join(['col', '0', '1', '5', ''])
+
+    code, (stdout, stderr) = mysql_client.exec_run('''
+        mysql
+        -e "INSERT INTO mysql_federated.test VALUES (0), (1), (5);"
+        -e "SELECT * FROM mysql_federated.test ORDER BY col;"
+    '''.format(host=server_address, port=server_port), demux=True)
+
+    assert stdout == '\n'.join(['col', '0', '0', '1', '1', '5', '5', ''])
+
 
 def test_python_client(server_address):
     client = pymysql.connections.Connection(host=server_address, user='user_with_double_sha1', password='abacaba', database='default', port=server_port)