Merge pull request #4280 from sysown/v2.x_transaction_isolation_level…

…_support

Adding SET TRANSACTION ISOLATION LEVEL * support

include/MySQL_Variables.h

@@ -39,11 +39,13 @@ class MySQL_Variables {
 
 	bool client_set_value(MySQL_Session* session, int idx, const std::string& value);
 	bool client_set_hash_and_value(MySQL_Session* session, int idx, const std::string& value, uint32_t hash);
+	void client_reset_value(MySQL_Session* session, int idx);
 	const char* client_get_value(MySQL_Session* session, int idx) const;
 	uint32_t client_get_hash(MySQL_Session* session, int idx) const;
 
 	void server_set_value(MySQL_Session* session, int idx, const char* value);
 	void server_set_hash_and_value(MySQL_Session* session, int idx, const char* value, uint32_t hash);
+	void server_reset_value(MySQL_Session* session, int idx);
 	const char* server_get_value(MySQL_Session* session, int idx) const;
 	inline uint32_t server_get_hash(MySQL_Session* session, int idx) const;
 

include/gen_utils.h

@@ -332,3 +332,5 @@ std::string trim(const std::string& s);
  * @return An 'unique_ptr' holding the resulting 'SQLite3_result'.
  */
 std::unique_ptr<SQLite3_result> get_SQLite3_resulset(MYSQL_RES* resultset);
+
+std::vector<std::string> split_string(const std::string& str, char delimiter);

include/proxysql_structs.h

@@ -221,6 +221,8 @@ enum mysql_variable_name {
 	SQL_TIME_ZONE,
 	SQL_TIMESTAMP,
 	SQL_TMP_TABLE_SIZE,
+	SQL_NEXT_ISOLATION_LEVEL,
+	SQL_NEXT_TRANSACTION_READ,
 	SQL_UNIQUE_CHECKS,
 	SQL_WSREP_OSU_METHOD,
 	SQL_NAME_LAST_HIGH_WM,
@@ -254,6 +256,8 @@ enum session_status {
 	SETTING_MULTIPLE_VARIABLES,
 	SETTING_SET_NAMES,
 	SHOW_WARNINGS,
+	SETTING_NEXT_ISOLATION_LEVEL,
+	SETTING_NEXT_TRANSACTION_READ,
 	session_status___NONE // special marker
 };
 
@@ -1188,6 +1192,8 @@ mysql_variable_st mysql_tracked_variables[] {
 	{ SQL_TIME_ZONE,                  SETTING_VARIABLE, true,  false, false, false, (char *)"time_zone",                  NULL, (char *)"SYSTEM" , false} ,
 	{ SQL_TIMESTAMP,                  SETTING_VARIABLE, false, false, true,  false, (char *)"timestamp",                  NULL, (char *)"" , false} ,
 	{ SQL_TMP_TABLE_SIZE,             SETTING_VARIABLE, false, false, true,  false, (char *)"tmp_table_size",             NULL, (char *)"" , false} ,
+	{ SQL_NEXT_ISOLATION_LEVEL, SETTING_NEXT_ISOLATION_LEVEL, false, true,  false, false, (char *)"transaction isolation level", (char *)"next_isolation_level", (char *)"READ COMMITTED" , false} ,
+	{ SQL_NEXT_TRANSACTION_READ, SETTING_NEXT_TRANSACTION_READ, false, true,  false, false, (char *)"transaction read", (char *)"next_transaction_read", (char *)"WRITE" , false} ,
 	{ SQL_UNIQUE_CHECKS,              SETTING_VARIABLE, true,  false, false, true,  (char *)"unique_checks",              NULL, (char *)"" , false} ,
 	{ SQL_WSREP_OSU_METHOD,           SETTING_VARIABLE, true,  false, false, false, (char *)"wsrep_osu_method",           NULL, (char *)"" , false} ,
 

lib/MySQL_Session.cpp

@@ -2546,10 +2546,16 @@ bool MySQL_Session::handler_again___status_SETTING_GENERIC_VARIABLE(int *_rc, co
 			char *sv = mybe->server_myds->myconn->mysql->server_version;
 			if (strncmp(sv,(char *)"8",1)==0) {
 				sprintf(query,q,"transaction_isolation", var_value);
-			}
-			else {
+			} else {
 				sprintf(query,q,"tx_isolation", var_value);
 			}
+		} else if (strncasecmp("tx_read_only", var_name, 12) == 0) {
+			char* sv = mybe->server_myds->myconn->mysql->server_version;
+			if (strncmp(sv, (char *)"8", 1) == 0) {
+				sprintf(query,q,"transaction_read_only", var_value);
+			} else {
+				sprintf(query,q,"tx_read_only", var_value);
+			}
 		} else if (strncasecmp("aurora_read_replica_read_committed", var_name, 34) == 0) {
 			// If aurora_read_replica_read_committed is set, isolation level is
 			// internally reset so that it will be set again.
@@ -2582,6 +2588,15 @@ bool MySQL_Session::handler_again___status_SETTING_GENERIC_VARIABLE(int *_rc, co
 		myds->DSS = STATE_MARIADB_GENERIC;
 		st=previous_status.top();
 		previous_status.pop();
+
+		if (strcasecmp("transaction isolation level", var_name) == 0) {
+			mysql_variables.server_reset_value(this, SQL_NEXT_ISOLATION_LEVEL);
+			mysql_variables.client_reset_value(this, SQL_NEXT_ISOLATION_LEVEL);
+		} else if (strcasecmp("transaction read", var_name) == 0) {
+			mysql_variables.server_reset_value(this, SQL_NEXT_TRANSACTION_READ);
+			mysql_variables.client_reset_value(this, SQL_NEXT_TRANSACTION_READ);
+		}
+
 		NEXT_IMMEDIATE_NEW(st);
 	} else {
 		if (rc==-1) {
@@ -5143,6 +5158,8 @@ int MySQL_Session::handler() {
 		case SETTING_TRANSACTION_READ:
 		case SETTING_CHARSET:
 		case SETTING_VARIABLE:
+		case SETTING_NEXT_ISOLATION_LEVEL:
+		case SETTING_NEXT_TRANSACTION_READ:
 			{
 				int rc = 0;
 				if (mysql_variables.update_variable(this, status, rc)) {
@@ -6432,6 +6449,33 @@ bool MySQL_Session::handler___status_WAITING_CLIENT_DATA___STATE_SLEEP___MYSQL_C
 								return false;
 							proxy_debug(PROXY_DEBUG_MYSQL_COM, 8, "Changing connection TX ISOLATION to %s\n", value1.c_str());
 						}
+					} else if (var == "tx_read_only") {
+						std::string value1 = *values;
+						proxy_debug(PROXY_DEBUG_MYSQL_COM, 5, "Processing SET tx_read_only value %s\n", value1.c_str());
+
+						if (
+							(value1 == "0") ||
+							(strcasecmp(value1.c_str(), "false")==0) ||
+							(strcasecmp(value1.c_str(), "off")==0)
+						) {
+							value1 = "WRITE";
+						} else if (
+							(value1 == "1") ||
+							(strcasecmp(value1.c_str(), "true")==0) ||
+							(strcasecmp(value1.c_str(), "on")==0)
+						) {
+							value1 = "ONLY";
+						} else {
+							//proxy_warning("Unknown tx_read_only value \"%s\"\n", value1.c_str());
+							unable_to_parse_set_statement(lock_hostgroup);
+							return false;
+						}
+						uint32_t read_only_int=SpookyHash::Hash32(value1.c_str(),value1.length(),10);
+						if (mysql_variables.client_get_hash(this, SQL_TRANSACTION_READ) != read_only_int) {
+							if (!mysql_variables.client_set_value(this, SQL_TRANSACTION_READ, value1.c_str()))
+								return false;
+							proxy_debug(PROXY_DEBUG_MYSQL_COM, 8, "Changing connection TX ACCESS MODE to READ %s\n", value1.c_str());
+						}
 					} else if (std::find(mysql_variables.ignore_vars.begin(), mysql_variables.ignore_vars.end(), var) != mysql_variables.ignore_vars.end()) {
 						// this is a variable we parse but ignore
 						// see MySQL_Variables::MySQL_Variables() for a list of ignored variables
@@ -6550,27 +6594,51 @@ bool MySQL_Session::handler___status_WAITING_CLIENT_DATA___STATE_SLEEP___MYSQL_C
 			} else if (match_regexes && match_regexes[2]->match(dig)) {
 				SetParser parser(nq);
 				std::map<std::string, std::vector<std::string>> set = parser.parse2();
+
 				for(auto it = std::begin(set); it != std::end(set); ++it) {
-					std::string var = it->first;
-					auto values = std::begin(it->second);
-					proxy_debug(PROXY_DEBUG_MYSQL_COM, 5, "Processing SET variable %s\n", var.c_str());
-					if (var == "isolation level") {
-						std::string value1 = *values;
-						proxy_debug(PROXY_DEBUG_MYSQL_COM, 5, "Processing SET SESSION TRANSACTION ISOLATION LEVEL value %s\n", value1.c_str());
-						uint32_t isolation_level_int=SpookyHash::Hash32(value1.c_str(),value1.length(),10);
-						if (mysql_variables.client_get_hash(this, SQL_ISOLATION_LEVEL) != isolation_level_int) {
-							if (!mysql_variables.client_set_value(this, SQL_ISOLATION_LEVEL, value1.c_str()))
-								return false;
-							proxy_debug(PROXY_DEBUG_MYSQL_COM, 8, "Changing connection TRANSACTION ISOLATION LEVEL to %s\n", value1.c_str());
+
+					const std::vector<std::string>& val = split_string(it->first, ':');
+
+					if (val.size() == 2) {
+
+						const auto values = std::begin(it->second);
+						const std::string& var = val[1];
+
+						enum mysql_variable_name isolation_level_val;
+						enum mysql_variable_name transaction_read_val;
+
+						if (val[0] == "session") {
+							isolation_level_val = SQL_ISOLATION_LEVEL;
+							transaction_read_val = SQL_TRANSACTION_READ;
+						} else {
+							isolation_level_val = SQL_NEXT_ISOLATION_LEVEL;
+							transaction_read_val = SQL_NEXT_TRANSACTION_READ;
 						}
-					} else if (var == "read") {
-						std::string value1 = *values;
-						proxy_debug(PROXY_DEBUG_MYSQL_COM, 5, "Processing SET SESSION TRANSACTION READ value %s\n", value1.c_str());
-						uint32_t transaction_read_int=SpookyHash::Hash32(value1.c_str(),value1.length(),10);
-						if (mysql_variables.client_get_hash(this, SQL_TRANSACTION_READ) != transaction_read_int) {
-							if (!mysql_variables.client_set_value(this, SQL_TRANSACTION_READ, value1.c_str()))
-								return false;
-							proxy_debug(PROXY_DEBUG_MYSQL_COM, 8, "Changing connection TRANSACTION READ to %s\n", value1.c_str());
+
+						proxy_debug(PROXY_DEBUG_MYSQL_COM, 5, "Processing SET variable %s\n", var.c_str());
+						if (var == "isolation level") {
+							const std::string& value1 = *values;
+							proxy_debug(PROXY_DEBUG_MYSQL_COM, 5, "Processing SET %s TRANSACTION ISOLATION LEVEL value %s\n", val[0].c_str(), value1.c_str());
+							const uint32_t isolation_level_int = SpookyHash::Hash32(value1.c_str(), value1.length(), 10);
+							if (mysql_variables.client_get_hash(this, isolation_level_val) != isolation_level_int) {
+								if (!mysql_variables.client_set_value(this, isolation_level_val, value1.c_str()))
+									return false;
+
+								proxy_debug(PROXY_DEBUG_MYSQL_COM, 8, "Changing connection TRANSACTION ISOLATION LEVEL to %s\n", value1.c_str());
+							}
+						} else if (var == "read") {
+							const std::string& value1 = *values;
+							proxy_debug(PROXY_DEBUG_MYSQL_COM, 5, "Processing SET %s TRANSACTION READ value %s\n", val[0].c_str(), value1.c_str());
+							const uint32_t transaction_read_int = SpookyHash::Hash32(value1.c_str(), value1.length(), 10);
+							if (mysql_variables.client_get_hash(this, transaction_read_val) != transaction_read_int) {
+								if (!mysql_variables.client_set_value(this, transaction_read_val, value1.c_str()))
+									return false;
+
+								proxy_debug(PROXY_DEBUG_MYSQL_COM, 8, "Changing connection TRANSACTION READ to %s\n", value1.c_str());
+							}
+						} else {
+							unable_to_parse_set_statement(lock_hostgroup);
+							return false;
 						}
 					} else {
 						unable_to_parse_set_statement(lock_hostgroup);

lib/MySQL_Thread.cpp

@@ -2975,7 +2975,7 @@ bool MySQL_Thread::init() {
 	match_regexes[0] = NULL; // NOTE: historically we used match_regexes[0] for SET SQL_LOG_BIN . Not anymore
 
 	std::stringstream ss;
-	ss << "^SET (|SESSION |@@|@@session.|@@local.)`?(" << mysql_variables.variables_regexp << "SESSION_TRACK_GTIDS|TX_ISOLATION)`?( *)(:|)=( *)";
+	ss << "^SET (|SESSION |@@|@@session.|@@local.)`?(" << mysql_variables.variables_regexp << "SESSION_TRACK_GTIDS|TX_ISOLATION|TX_READ_ONLY|TRANSACTION_ISOLATION|TRANSACTION_READ_ONLY)`?( *)(:|)=( *)";
 	match_regexes[1]=new Session_Regex((char *)ss.str().c_str());
 
 	match_regexes[2]=new Session_Regex((char *)"^SET(?: +)(|SESSION +)TRANSACTION(?: +)(?:(?:(ISOLATION(?: +)LEVEL)(?: +)(REPEATABLE(?: +)READ|READ(?: +)COMMITTED|READ(?: +)UNCOMMITTED|SERIALIZABLE))|(?:(READ)(?: +)(WRITE|ONLY)))");

lib/MySQL_Variables.cpp

@@ -94,6 +94,24 @@ bool MySQL_Variables::client_set_hash_and_value(MySQL_Session* session, int idx,
 	return true;
 }
 
+void MySQL_Variables::client_reset_value(MySQL_Session* session, int idx) {
+	if (!session || !session->client_myds || !session->client_myds->myconn) {
+		proxy_warning("Session validation failed\n");
+		return;
+	}
+
+	MySQL_Connection *client_conn = session->client_myds->myconn;
+
+	if (client_conn->var_hash[idx] != 0) {
+		client_conn->var_hash[idx] = 0;
+		if (client_conn->variables[idx].value) {
+			free(client_conn->variables[idx].value);
+			client_conn->variables[idx].value = NULL;
+		}
+		// we now regererate dynamic_variables_idx
+		client_conn->reorder_dynamic_variables_idx();
+	}
+}
 void MySQL_Variables::server_set_hash_and_value(MySQL_Session* session, int idx, const char* value, uint32_t hash) {
 	if (!session || !session->mybe || !session->mybe->server_myds || !session->mybe->server_myds->myconn || !value) {
 		proxy_warning("Session validation failed\n");
@@ -222,6 +240,25 @@ void MySQL_Variables::server_set_value(MySQL_Session* session, int idx, const ch
 	session->mybe->server_myds->myconn->reorder_dynamic_variables_idx();
 }
 
+void MySQL_Variables::server_reset_value(MySQL_Session* session, int idx) {
+	assert(session);
+	assert(session->mybe);
+	assert(session->mybe->server_myds);
+	assert(session->mybe->server_myds->myconn);
+
+	MySQL_Connection *backend_conn = session->mybe->server_myds->myconn;
+
+	if (backend_conn->var_hash[idx] != 0) {
+		backend_conn->var_hash[idx] = 0;
+		if (backend_conn->variables[idx].value) {
+			free(backend_conn->variables[idx].value);
+			backend_conn->variables[idx].value = NULL;
+		}
+		// we now regererate dynamic_variables_idx
+		backend_conn->reorder_dynamic_variables_idx();
+	}
+}
+
 const char* MySQL_Variables::server_get_value(MySQL_Session* session, int idx) const {
 	assert(session);
 	assert(session->mybe);

lib/gen_utils.cpp

@@ -1,6 +1,6 @@
 #include <vector>
 #include <memory>
-
+#include <sstream>
 #include "gen_utils.h"
 
 
@@ -264,4 +264,16 @@ std::unique_ptr<SQLite3_result> get_SQLite3_resulset(MYSQL_RES* resultset) {
 	mysql_data_seek(resultset, 0);
 
 	return sqlite_result;
+} 
+
+std::vector<std::string> split_string(const std::string& str, char delimiter) {
+	std::vector<std::string> tokens {};
+	std::string token {};
+	std::istringstream tokenStream(str);
+
+	while (std::getline(tokenStream, token, delimiter)) {
+		tokens.push_back(token);
+	}
+
+	return tokens;
 }

lib/set_parser.cpp

@@ -104,7 +104,12 @@ VALGRIND_ENABLE_ERROR_REPORTING;
 			}
 		} else if (value4 != "") {
 			// VARIABLE
-		value5.erase(value5.find_last_not_of(" \n\r\t,")+1);
+			if (strcasecmp("transaction_isolation", value4.c_str()) == 0) {
+				value4 = "tx_isolation";
+			} else if (strcasecmp("transaction_read_only", value4.c_str()) == 0) {
+				value4 = "tx_read_only";
+			}
+			value5.erase(value5.find_last_not_of(" \n\r\t,")+1);
 			key = value4;
 			if (value5 == "''" || value5 == "\"\"") {
 				op.push_back("");
@@ -356,6 +361,11 @@ VALGRIND_ENABLE_ERROR_REPORTING;
 			}
 		} else if (value4 != "") {
 			// VARIABLE
+			if (strcasecmp("transaction_isolation", value4.c_str()) == 0) {
+				value4 = "tx_isolation";
+			} else if (strcasecmp("transaction_read_only", value4.c_str()) == 0) {
+				value4 = "tx_read_only";
+			}
 			value5.erase(value5.find_last_not_of(" \n\r\t,")+1);
 			key = value4;
 			if (value5 == "''" || value5 == "\"\"") {
@@ -415,17 +425,17 @@ std::map<std::string,std::vector<std::string>> SetParser::parse2() {
 		proxy_debug(PROXY_DEBUG_MYSQL_QUERY_PROCESSOR, 4, "SET parsing: v1='%s' , v2='%s' , v3='%s' , v4='%s' , v5='%s'\n", value1.c_str(), value2.c_str(), value3.c_str(), value4.c_str(), value5.c_str());
 #endif // DEBUG
 		std::string key;
-		if (value1 != "") { // session is specified
+		//if (value1 != "") { // session is specified
 			if (value2 != "") { // isolation level
-				key = value2;
+				key = value1 + ":" + value2;
 				std::transform(value3.begin(), value3.end(), value3.begin(), ::toupper);
 				op.push_back(value3);
 			} else {
-				key = value4;
+				key = value1 + ":" + value4;
 				std::transform(value5.begin(), value5.end(), value5.begin(), ::toupper);
 				op.push_back(value5);
 			}
-		}
+		//}
 		std::transform(key.begin(), key.end(), key.begin(), ::tolower);
 		result[key] = op;
 	}