Enhanced compatibility with native mysql-connector-java(JDBC) (#10021)

* Skip the `/* comments */ SELECT @@variables ...` from mysql-connector-java setup for MySQL Handler #9336

mysql-connector setup query:
/* mysql-connector-java-5.1.38 ( Revision: ${revinfo.commit} ) */SELECT  @@session.auto_increment_increment AS auto_increment_increment, @@character_set_client AS character_set_client, @@character_set_connection AS character_set_connection, @@character_set_results AS character_set_results, @@character_set_server AS character_set_server, @@init_connect AS init_connect, @@interactive_timeout AS interactive_timeout...

ClickHouse side Error:
{} <Error> executeQuery: Code: 62, e.displayText() = DB::Exception: Syntax error: failed at position 74: @@session.auto_increment_increment AS auto_increment_increment, @@character_set_client AS character_set_client, @@character_set_connection AS character_set_conn. Expected one of: CAST, NULL...

Client side Exception:
java.sql.SQLException: Syntax error: failed at position 74: @@session.auto_increment_increment AS auto_increment_increment, @@character_set_client AS character_set_client, @@character_set_connection AS character_set_conn. Expected one of: CAST...

* add repalce 'SHOW VARIABLES' for mysql-connector-java-5.1.34 #9336

* Add java client(JDBC) integration test to test_mysql_protocol

* shift out java tests from dbms

* Update MySQLHandler.cpp

* Update MySQLHandler.cpp

* test_mysql_protocol: add Test.java exit code 1 when expection

Co-authored-by: alexey-milovidov <milovidov@yandex-team.ru>

programs/server/MySQLHandler.cpp

@@ -16,6 +16,7 @@
 #include <IO/WriteBufferFromPocoSocket.h>
 #include <Storages/IStorage.h>
 #include <boost/algorithm/string/replace.hpp>
+#include <regex>
 
 #if USE_POCO_NETSSL
 #include <Poco/Net/SecureStreamSocket.h>
@@ -268,7 +269,8 @@ void MySQLHandler::comPing()
     packet_sender->sendPacket(OK_Packet(0x0, client_capability_flags, 0, 0, 0), true);
 }
 
-static bool isFederatedServerSetupCommand(const String & query);
+static bool isFederatedServerSetupSetCommand(const String & query);
+static bool isFederatedServerSetupSelectVarCommand(const String & query);
 
 void MySQLHandler::comQuery(ReadBuffer & payload)
 {
@@ -276,7 +278,7 @@ void MySQLHandler::comQuery(ReadBuffer & payload)
 
     // 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))
+    if (isFederatedServerSetupSetCommand(query))
     {
         packet_sender->sendPacket(OK_Packet(0x00, client_capability_flags, 0, 0, 0), true);
     }
@@ -288,10 +290,11 @@ void MySQLHandler::comQuery(ReadBuffer & payload)
 
         // Translate query from MySQL to ClickHouse.
         // This is a temporary workaround until ClickHouse supports the syntax "@@var_name".
-        if (query == "select @@version_comment limit 1")  // MariaDB client starts session with that query
+        if (isFederatedServerSetupSelectVarCommand(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))
         {
@@ -358,11 +361,27 @@ void MySQLHandlerSSL::finishHandshakeSSL(size_t packet_size, char * buf, size_t
 
 #endif
 
-static bool isFederatedServerSetupCommand(const String & query)
+static bool isFederatedServerSetupSetCommand(const String & query)
 {
-    return 0 == strncasecmp("SET NAMES", query.c_str(), 9) || 0 == strncasecmp("SET character_set_results", query.c_str(), 25)
+    static const std::regex expr{
-        || 0 == strncasecmp("SET FOREIGN_KEY_CHECKS", query.c_str(), 22) || 0 == strncasecmp("SET AUTOCOMMIT", query.c_str(), 14)
+        "(^(SET NAMES(.*)))"
-        || 0 == strncasecmp("SET SESSION TRANSACTION ISOLATION LEVEL", query.c_str(), 39);
+        "|(^(SET character_set_results(.*)))"
+        "|(^(SET FOREIGN_KEY_CHECKS(.*)))"
+        "|(^(SET AUTOCOMMIT(.*)))"
+        "|(^(SET sql_mode(.*)))"
+        "|(^(SET SESSION TRANSACTION ISOLATION LEVEL(.*)))"
+        , std::regex::icase};
+    return 1 == std::regex_match(query, expr);
+}
+
+static bool isFederatedServerSetupSelectVarCommand(const String & query)
+{
+     static const std::regex expr{
+         "|(^(SELECT @@(.*)))"
+         "|(^((/\\*(.*)\\*/)([ \t]*)(SELECT([ \t]*)@@(.*))))"
+         "|(^((/\\*(.*)\\*/)([ \t]*)(SHOW VARIABLES(.*))))"
+         , std::regex::icase};
+     return 1 == std::regex_match(query, expr);
 }
 
 const String MySQLHandler::show_table_status_replacement_query("SELECT"

tests/integration/test_mysql_protocol/clients/java/0.reference

@@ -0,0 +1,15 @@
+33jdbc
+44ck
+0
+1
+2
+3
+4
+5
+6
+7
+8
+9
+10
+11
+12

tests/integration/test_mysql_protocol/clients/java/Dockerfile

@@ -0,0 +1,18 @@
+FROM ubuntu:18.04
+
+RUN apt-get update && \
+    apt-get install -y software-properties-common build-essential openjdk-8-jdk libmysql-java curl
+
+RUN rm -rf \
+        /var/lib/apt/lists/* \
+        /var/cache/debconf \
+        /tmp/* \
+RUN apt-get clean
+
+ARG ver=5.1.46
+RUN curl -L -o /mysql-connector-java-${ver}.jar https://repo1.maven.org/maven2/mysql/mysql-connector-java/${ver}/mysql-connector-java-${ver}.jar
+ENV CLASSPATH=$CLASSPATH:/mysql-connector-java-${ver}.jar
+
+WORKDIR /jdbc
+COPY Test.java Test.java
+RUN javac Test.java

tests/integration/test_mysql_protocol/clients/java/Test.java

@@ -0,0 +1,76 @@
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+
+class JavaConnectorTest {
+    private static final String CREATE_TABLE_SQL = "CREATE TABLE IF NOT EXISTS default.test1 (age Int32, name String) Engine = Memory";
+    private static final String INSERT_SQL = "INSERT INTO default.test1 VALUES(33, 'jdbc'),(44, 'ck')";
+    private static final String SELECT_SQL = "SELECT * FROM default.test1";
+    private static final String SELECT_NUMBER_SQL = "SELECT * FROM system.numbers LIMIT 13";
+    private static final String DROP_TABLE_SQL = "DROP TABLE default.test1";
+
+    public static void main(String[] args) {
+        int i = 0;
+        String host = "127.0.0.1";
+        String port = "9004";
+        String user = "default";
+        String password = "";
+        String database = "default";
+        while (i < args.length) {
+            switch (args[i]) {
+                case "--host":
+                    host = args[++i];
+                    break;
+                case "--port":
+                    port = args[++i];
+                    break;
+                case "--user":
+                    user = args[++i];
+                    break;
+                case "--password":
+                    password = args[++i];
+                    break;
+                case "--database":
+                    database = args[++i];
+                    break;
+                default:
+                    i++;
+                    break;
+            }
+        }
+
+        String jdbcUrl = String.format("jdbc:mysql://%s:%s/%s?maxAllowedPacket=67108864&useSSL=false", host, port, database);
+
+        Connection conn = null;
+        Statement stmt = null;
+        try {
+            conn = DriverManager.getConnection(jdbcUrl, user, password);
+            stmt = conn.createStatement();
+            stmt.executeUpdate(CREATE_TABLE_SQL);
+            stmt.executeUpdate(INSERT_SQL);
+
+            ResultSet rs = stmt.executeQuery(SELECT_SQL);
+            while (rs.next()) {
+                System.out.print(rs.getString("age"));
+                System.out.print(rs.getString("name"));
+                System.out.println();
+            }
+
+            stmt.executeUpdate(DROP_TABLE_SQL);
+
+            rs = stmt.executeQuery(SELECT_NUMBER_SQL);
+            while (rs.next()) {
+                System.out.print(rs.getString(1));
+                System.out.println();
+            }
+
+            stmt.close();
+            conn.close();
+        } catch (SQLException e) {
+            e.printStackTrace();
+            System.exit(1);
+        }
+    }
+}

tests/integration/test_mysql_protocol/clients/java/docker_compose.yml

@@ -0,0 +1,8 @@
+version: '2.2'
+services:
+  java1:
+    build:
+      context: ./
+      network: host
+    # to keep container running
+    command: sleep infinity

tests/integration/test_mysql_protocol/test.py

@@ -79,6 +79,13 @@ def nodejs_container():
     yield docker.from_env().containers.get(cluster.project_name + '_mysqljs1_1')
 
 
+@pytest.fixture(scope='module')
+def java_container():
+    docker_compose = os.path.join(SCRIPT_DIR, 'clients', 'java', 'docker_compose.yml')
+    subprocess.check_call(['docker-compose', '-p', cluster.project_name, '-f', docker_compose, 'up', '--no-recreate', '-d', '--build'])
+    yield docker.from_env().containers.get(cluster.project_name + '_java1_1')
+
+
 def test_mysql_client(mysql_client, server_address):
     # type: (Container, str) -> None
     code, (stdout, stderr) = mysql_client.exec_run('''
@@ -266,6 +273,21 @@ def test_mysqljs_client(server_address, nodejs_container):
     assert code == 1
 
 
+def test_java_client(server_address, java_container):
+    # type: (str, Container) -> None
+    with open(os.path.join(SCRIPT_DIR, 'clients', 'java', '0.reference')) as fp:
+        reference = fp.read()
+
+    code, (stdout, stderr) = java_container.exec_run('java JavaConnectorTest --host {host} --port {port} --user user_with_empty_password --database '
+                                                       'abc'.format(host=server_address, port=server_port), demux=True)
+    assert code == 1
+
+    code, (stdout, stderr) = java_container.exec_run('java JavaConnectorTest --host {host} --port {port} --user user_with_empty_password --database '
+                                                       'default'.format(host=server_address, port=server_port), demux=True)
+    assert code == 0
+    assert stdout == reference
+
+
 def test_types(server_address):
     client = pymysql.connections.Connection(host=server_address, user='default', password='123', database='default', port=server_port)