Merge pull request #1922 from ahaha-zhang/v1.4.16

support more mysql system variables to keep mulitplexing enabled

include/MySQL_Thread.h

@@ -362,6 +362,7 @@ class MySQL_Threads_Handler
 		char *default_schema;
 		char *interfaces;
 		char *server_version;
+        char *keep_multiplexing_variables;
 		uint8_t default_charset;
 		bool servers_stats;
 		bool commands_stats;

include/mysql_connection.h

@@ -170,6 +170,7 @@ class MySQL_Connection {
 	bool IsServerOffline();
 	bool IsAutoCommit();
 	bool MultiplexDisabled();
+    bool IsKeepMultiplexEnabledVariables(char *query_digest_text);
 	void ProcessQueryAndSetStatusFlags(char *query_digest_text);
 	void optimize();
 	void close_mysql();

include/proxysql_structs.h

@@ -589,6 +589,7 @@ __thread char *mysql_thread___server_version;
 __thread char *mysql_thread___init_connect;
 __thread char *mysql_thread___default_sql_mode;
 __thread char *mysql_thread___default_time_zone;
+__thread char *mysql_thread___keep_multiplexing_variables;
 __thread int mysql_thread___max_allowed_packet;
 __thread int mysql_thread___throttle_connections_per_sec_to_hostgroup;
 __thread int mysql_thread___max_transaction_time;
@@ -696,6 +697,7 @@ extern ProxySQL_GlobalVariables GloVars;
 extern MySQL_HostGroups_Manager *MyHGM;
 extern __thread char *mysql_thread___default_schema;
 extern __thread char *mysql_thread___server_version;
+extern __thread char *mysql_thread___keep_multiplexing_variables;
 extern __thread char *mysql_thread___init_connect;
 extern __thread char *mysql_thread___default_sql_mode;
 extern __thread char *mysql_thread___default_time_zone;

lib/MySQL_HostGroups_Manager.cpp

@@ -1517,6 +1517,7 @@ MySQL_Connection * MySQL_HostGroups_Manager::get_MyConn_from_pool(unsigned int _
 		conn=mysrvc->ConnectionsFree->get_random_MyConn(sess, ff);
 		if (conn) {
 			mysrvc->ConnectionsUsed->add(conn);
+            conn->auto_increment_delay_token = 0;
 			status.myconnpoll_get_ok++;
 		}
 	}

lib/MySQL_Thread.cpp

@@ -312,6 +312,7 @@ static char * mysql_thread_variables_names[]= {
 	(char *)"connpoll_reset_queue_length",
 	(char *)"stats_time_backend_query",
 	(char *)"stats_time_query_processor",
+    (char *)"keep_multiplexing_variables",
 	NULL
 };
 
@@ -430,6 +431,7 @@ MySQL_Threads_Handler::MySQL_Threads_Handler() {
 	variables.ssl_p2s_cert=NULL;
 	variables.ssl_p2s_key=NULL;
 	variables.ssl_p2s_cipher=NULL;
+    variables.keep_multiplexing_variables=strdup((char *)"tx_isolation,version");
 #ifdef DEBUG
 	variables.session_debug=true;
 #endif /*debug */
@@ -570,6 +572,7 @@ char * MySQL_Threads_Handler::get_variable_string(char *name) {
 	if (!strcasecmp(name,"eventslog_filename")) return strdup(variables.eventslog_filename);
 	if (!strcasecmp(name,"default_schema")) return strdup(variables.default_schema);
 	if (!strcasecmp(name,"interfaces")) return strdup(variables.interfaces);
+    if (!strcasecmp(name,"keep_multiplexing_variables")) return strdup(variables.keep_multiplexing_variables);
 	proxy_error("Not existing variable: %s\n", name); assert(0);
 	return NULL;
 }
@@ -707,6 +710,7 @@ char * MySQL_Threads_Handler::get_variable(char *name) {	// this is the public f
 	if (!strcasecmp(name,"eventslog_filename")) return strdup(variables.eventslog_filename);
 	if (!strcasecmp(name,"default_schema")) return strdup(variables.default_schema);
 	if (!strcasecmp(name,"interfaces")) return strdup(variables.interfaces);
+    if (!strcasecmp(name,"keep_multiplexing_variables")) return strdup(variables.keep_multiplexing_variables);
 	if (!strcasecmp(name,"server_capabilities")) {
 		// FIXME : make it human readable
 		sprintf(intbuf,"%d",variables.server_capabilities);
@@ -1729,6 +1733,15 @@ bool MySQL_Threads_Handler::set_variable(char *name, char *value) {	// this is t
 		}
 		return true;
 	}
+    if (!strcasecmp(name,"keep_multiplexing_variables")) {
+        if (vallen) {
+            free(variables.keep_multiplexing_variables);
+            variables.keep_multiplexing_variables=strdup(value);
+            return true;
+        } else {
+            return false;
+        }
+    }
 
 	// SSL proxy to server variables
 	if (!strcasecmp(name,"ssl_p2s_ca")) {
@@ -3388,6 +3401,7 @@ void MySQL_Thread::refresh_variables() {
 	mysql_thread___eventslog_filename=GloMTH->get_variable_string((char *)"eventslog_filename");
 	GloMyLogger->set_base_filename(); // both filename and filesize are set here
 	if (mysql_thread___default_schema) free(mysql_thread___default_schema);
+    mysql_thread___keep_multiplexing_variables=GloMTH->get_variable_string((char *)"keep_multiplexing_variables");
 	mysql_thread___default_schema=GloMTH->get_variable_string((char *)"default_schema");
 	mysql_thread___server_capabilities=GloMTH->get_variable_uint16((char *)"server_capabilities");
 	mysql_thread___default_charset=GloMTH->get_variable_uint8((char *)"default_charset");

lib/mysql_connection.cpp

@@ -1582,6 +1582,83 @@ bool MySQL_Connection::MultiplexDisabled() {
 	return ret;
 }
 
+bool MySQL_Connection::IsKeepMultiplexEnabledVariables(char *query_digest_text) {
+    if (query_digest_text==NULL) return true;
+
+    char *query_digest_text_filter_select;
+    unsigned long query_digest_text_len=strlen(query_digest_text);
+    if (strncasecmp(query_digest_text,"SELECT ",strlen("SELECT "))==0){
+        query_digest_text_filter_select=(char*)malloc(query_digest_text_len-7+1);
+        memcpy(query_digest_text_filter_select,&query_digest_text[7],query_digest_text_len-7);
+        query_digest_text_filter_select[query_digest_text_len-7]='\0';
+    }
+    //filter @@session. and @@
+    char *match=NULL;
+    while ((match = strcasestr(query_digest_text_filter_select,"@@session."))) {
+        *match = '\0';
+        strcat(query_digest_text_filter_select, match+strlen("@@session."));
+    }
+    while ((match = strcasestr(query_digest_text_filter_select,"@@"))) {
+        *match = '\0';
+        strcat(query_digest_text_filter_select, match+strlen("@@"));
+    }
+
+    std::vector<char*>query_digest_text_filter_select_v;
+    char* query_digest_text_filter_select_tok=strtok(query_digest_text_filter_select, ",");
+    while(query_digest_text_filter_select_tok){
+        //filter "as"/space/alias,such as select @@version as a, @@version b
+        while (1){
+            char c = *query_digest_text_filter_select_tok;
+            if (!isspace(c)){
+                break;
+            }
+            query_digest_text_filter_select_tok++;
+        }
+        char* match_as;
+        match_as=strcasestr(query_digest_text_filter_select_tok," ");
+        if(match_as){
+            query_digest_text_filter_select_tok[match_as-query_digest_text_filter_select_tok]='\0';
+            query_digest_text_filter_select_v.push_back(query_digest_text_filter_select_tok);
+        }else{
+            query_digest_text_filter_select_v.push_back(query_digest_text_filter_select_tok);
+        }
+        query_digest_text_filter_select_tok=strtok(NULL, ",");
+    }
+
+    std::vector<char*>keep_multiplexing_variables_v;
+    char* keep_multiplexing_variables_tmp;
+    unsigned long keep_multiplexing_variables_len=strlen(mysql_thread___keep_multiplexing_variables);
+    keep_multiplexing_variables_tmp=(char*)malloc(keep_multiplexing_variables_len+1);
+    memcpy(keep_multiplexing_variables_tmp, mysql_thread___keep_multiplexing_variables, keep_multiplexing_variables_len);
+    keep_multiplexing_variables_tmp[keep_multiplexing_variables_len]='\0';
+    char* keep_multiplexing_variables_tok=strtok(keep_multiplexing_variables_tmp, " ,");
+    while (keep_multiplexing_variables_tok){
+        keep_multiplexing_variables_v.push_back(keep_multiplexing_variables_tok);
+        keep_multiplexing_variables_tok=strtok(NULL, " ,");
+    }
+
+    for (std::vector<char*>::iterator it=query_digest_text_filter_select_v.begin();it!=query_digest_text_filter_select_v.end();it++){
+        bool is_match=false;
+        for (std::vector<char*>::iterator it1=keep_multiplexing_variables_v.begin();it1!=keep_multiplexing_variables_v.end();it1++){
+            //printf("%s,%s\n",*it,*it1);
+            if (strncasecmp(*it,*it1,strlen(*it1))==0){
+                is_match=true;
+                break;
+            }
+        }
+        if(is_match){
+            is_match=false;
+            continue;
+        }else{
+            free(query_digest_text_filter_select);
+            free(keep_multiplexing_variables_tmp);
+            return false;
+        }
+    }
+    free(query_digest_text_filter_select);
+    free(keep_multiplexing_variables_tmp);
+    return true;
+}
 
 void MySQL_Connection::ProcessQueryAndSetStatusFlags(char *query_digest_text) {
 	if (query_digest_text==NULL) return;
@@ -1603,11 +1680,11 @@ void MySQL_Connection::ProcessQueryAndSetStatusFlags(char *query_digest_text) {
 	}
 	if (get_status_user_variable()==false) { // we search for variables only if not already set
 		if (mul!=2 && index(query_digest_text,'@')) { // mul = 2 has a special meaning : do not disable multiplex for variables in THIS QUERY ONLY
-			if (
-				strncasecmp(query_digest_text,"SELECT @@tx_isolation", strlen("SELECT @@tx_isolation"))
-				&&
-				strncasecmp(query_digest_text,"SELECT @@version", strlen("SELECT @@version"))
-			) {
+			//if (
+			//	strncasecmp(query_digest_text,"SELECT @@tx_isolation", strlen("SELECT @@tx_isolation"))
+			//	&&
+			//	strncasecmp(query_digest_text,"SELECT @@version", strlen("SELECT @@version"))
+            if(!IsKeepMultiplexEnabledVariables(query_digest_text)) {
 				set_status_user_variable(true);
 			}
 		}

test/keepmultiplexingvariables/client1.cpp

@@ -0,0 +1,40 @@
+#include <iostream>
+#include "proxysql.h"
+#include "cpp.h"
+
+Query_Cache *GloQC;
+MySQL_Authentication *GloMyAuth;
+Query_Processor *GloQPro;
+ProxySQL_Admin *GloAdmin;
+MySQL_Threads_Handler *GloMTH;
+
+MySQL_Thread *GloMT;
+
+MySQL_STMT_Manager_v14 *GloMyStmt;
+
+MySQL_Monitor *GloMyMon;
+std::thread *MyMon_thread = NULL;
+
+MySQL_Logger *GloMyLogger;
+
+SQLite3_Server *GloSQLite3Server;
+
+ProxySQL_Cluster *GloProxyCluster = NULL;
+
+ProxySQL_Statistics *GloProxyStats = NULL;
+
+int main(int argc, const char * argv[]) {
+    char* query_digest_text = "select @@session.tx_isolation as a,@@tx_isolation,@@tx_isolation as a,@@version";
+
+    MySQL_Connection conn;
+
+    mysql_thread___keep_multiplexing_variables="tx_isolation,version";
+
+    std::cout << conn.IsKeepMultiplexEnabledVariables(query_digest_text);
+    return 0;
+}
+
+//link and build
+//c++ -c -o client1.o client1.cpp -std=c++11 -I../../include -I../../deps/jemalloc/jemalloc/include/jemalloc -I../../deps/mariadb-client-library/mariadb_client/include -I../../deps/libconfig/libconfig-1.4.9/lib -I../../deps/libdaemon/libdaemon -I../../deps/sqlite3/sqlite3 -I../../deps/clickhouse-cpp/clickhouse-cpp -I../../deps/libmicrohttpd/libmicrohttpd/src/include
+
+//c++ -o client1 client1.o ../../src/obj/proxysql_global.o ../../lib/libproxysql.a ../../deps/libmicrohttpd/libmicrohttpd/src/microhttpd/.libs/libmicrohttpd.a ../../deps/pcre/pcre/.libs/libpcre.a ../../deps/pcre/pcre/.libs/libpcrecpp.a  ../../deps/libdaemon/libdaemon/libdaemon/.libs/libdaemon.a  ../../deps/libconfig/libconfig/lib/.libs/libconfig++.a ../../deps/libconfig/libconfig/lib/.libs/libconfig.a ../../deps/curl/curl/lib/.libs/libcurl.a ../../deps/sqlite3/sqlite3/sqlite3.o -std=c++11 -I../../include -I../../deps/jemalloc/jemalloc/include/jemalloc -I../../deps/mariadb-client-library/mariadb_client/include -I../../deps/libconfig/libconfig-1.4.9/lib -I../../deps/libdaemon/libdaemon -I../../deps/sqlite3/sqlite3 -I../../deps/clickhouse-cpp/clickhouse-cpp -I../../deps/libmicrohttpd/libmicrohttpd/src/include     -L../../lib -L../../deps/jemalloc/jemalloc/lib -L../../deps/libconfig/libconfig-1.4.9/lib/.libs -L../../deps/re2/re2/obj -L../../deps/mariadb-client-library/mariadb_client/libmariadb -L../../deps/libdaemon/libdaemon/libdaemon/.libs -L../../deps/pcre/pcre/.libs -L../../deps/libmicrohttpd/libmicrohttpd/src/microhttpd/.libs  -L/usr/local/opt/openssl/lib  -lssl -lre2 -lmariadbclient -lpthread -lm -lz -liconv -lcrypto