MySQL: Track session-specific system variables changes

- Add new mysql variable `mysql-session_track_variables`.
- Configure `session_track_system_variables` and `session_track_state_change`
  on backend connections if the mysql variable is enabled.
- Utilize notifications from backend servers to capture system variable changes
  that cannot be handled by `MySQL_Set_Stmt_Parser`
- Update both client and server variable maps based on backend responses.
- TAP test to verify this patch.

Signed-off-by: Wazir Ahmed <[email protected]>

Author: wazir-ahmed

include/MySQL_Session.h

@@ -225,13 +225,16 @@ class MySQL_Session: public Base_Session<MySQL_Session, MySQL_Data_Stream, MySQL
 	bool handler_again___verify_ldap_user_variable();
 	bool handler_again___verify_backend_autocommit();
 	bool handler_again___verify_backend_session_track_gtids();
+	bool handler_again___verify_backend_session_track_variables();
 	bool handler_again___verify_backend_multi_statement();
 	bool handler_again___verify_backend_user_schema();
 	bool handler_again___verify_multiple_variables(MySQL_Connection *);
 	bool handler_again___status_SETTING_INIT_CONNECT(int *);
 	bool handler_again___status_SETTING_LDAP_USER_VARIABLE(int *);
 	bool handler_again___status_SETTING_SQL_MODE(int *);
 	bool handler_again___status_SETTING_SESSION_TRACK_GTIDS(int *);
+	bool handler_again___status_SETTING_SESSION_TRACK_VARIABLES(int *);
+	bool handler_again___status_SETTING_SESSION_TRACK_STATE(int *);
 	bool handler_again___status_CHANGING_CHARSET(int *_rc);
 	bool handler_again___status_CHANGING_SCHEMA(int *);
 	bool handler_again___status_CONNECTING_SERVER(int *);
@@ -280,6 +283,7 @@ class MySQL_Session: public Base_Session<MySQL_Session, MySQL_Data_Stream, MySQL
 	int RunQuery(MySQL_Data_Stream *myds, MySQL_Connection *myconn);
 	void handler___status_WAITING_CLIENT_DATA();
 	void handler_rc0_Process_GTID(MySQL_Connection *myconn);
+	void handler_rc0_Process_Variables(MySQL_Connection *myconn);
 	void handler_rc0_RefreshActiveTransactions(MySQL_Connection* myconn);
 	void handler___status_WAITING_CLIENT_DATA___STATE_SLEEP___MYSQL_COM_INIT_DB_replace_CLICKHOUSE(PtrSize_t& pkt);
 	void handler___status_WAITING_CLIENT_DATA___STATE_SLEEP___MYSQL_COM_QUERY___not_mysql(PtrSize_t& pkt);

include/MySQL_Thread.h

@@ -326,6 +326,13 @@ struct th_metrics_map_idx {
 	};
 };
 
+struct session_track_variables {
+	enum mode {
+		DISABLED = 0,
+		ENABLED
+	};
+};
+
 /**
  * @brief Structure holding the data for a Client_Host_Cache entry.
  */
@@ -579,6 +586,7 @@ class MySQL_Threads_Handler
 		int processlist_max_query_length;
 
 		bool ignore_min_gtid_annotations;
+		int session_track_variables;
 	} variables;
 	struct {
 		unsigned int mirror_sessions_current;

include/mysql_connection.h

@@ -89,6 +89,8 @@ class MySQL_Connection {
 		char *ldap_user_variable;
 		char *ldap_user_variable_value;
 		bool session_track_gtids_sent;
+		bool session_track_variables_sent;
+		bool session_track_state_sent;
 		bool ldap_user_variable_sent;
 		uint8_t protocol_version;
 		int8_t last_set_autocommit;
@@ -262,6 +264,7 @@ class MySQL_Connection {
 	void reset();
 
 	bool get_gtid(char *buff, uint64_t *trx_id);
+	bool get_variables(std::unordered_map<std::string, std::string>&);
 	void reduce_auto_increment_delay_token() { if (auto_increment_delay_token) auto_increment_delay_token--; };
 
 	bool match_ff_req_options(const MySQL_Connection *c);

include/proxysql_structs.h

@@ -311,6 +311,8 @@ enum session_status {
 	SETTING_NEXT_TRANSACTION_READ,
 	PROCESSING_EXTENDED_QUERY_SYNC,
 	RESYNCHRONIZING_CONNECTION,
+	SETTING_SESSION_TRACK_VARIABLES,
+	SETTING_SESSION_TRACK_STATE,
 	session_status___NONE // special marker
 };
 
@@ -1304,6 +1306,7 @@ __thread int mysql_thread___client_host_error_counts;
 __thread int mysql_thread___handle_warnings;
 __thread int mysql_thread___evaluate_replication_lag_on_servers_load;
 __thread bool mysql_thread___ignore_min_gtid_annotations;
+__thread int mysql_thread___session_track_variables;
 
 /* variables used for Query Cache */
 __thread int mysql_thread___query_cache_size_MB;
@@ -1607,6 +1610,7 @@ extern __thread int mysql_thread___client_host_error_counts;
 extern __thread int mysql_thread___handle_warnings;
 extern __thread int mysql_thread___evaluate_replication_lag_on_servers_load;
 extern __thread bool mysql_thread___ignore_min_gtid_annotations;
+extern __thread int mysql_thread___session_track_variables;
 
 /* variables used for Query Cache */
 extern __thread int mysql_thread___query_cache_size_MB;

lib/MySQL_Session.cpp

@@ -1953,6 +1953,25 @@ bool MySQL_Session::handler_again___verify_backend_session_track_gtids() {
 	return ret;
 }
 
+bool MySQL_Session::handler_again___verify_backend_session_track_variables() {
+	if (mysql_thread___session_track_variables == session_track_variables::DISABLED) {
+		return false;
+	}
+
+	if (mybe->server_myds->myconn->options.session_track_variables_sent == false) {
+		mybe->server_myds->myconn->options.session_track_variables_sent = true;
+		set_previous_status_mode3();
+		NEXT_IMMEDIATE_NEW(SETTING_SESSION_TRACK_VARIABLES);
+	}
+
+	if (mybe->server_myds->myconn->options.session_track_state_sent == false) {
+		mybe->server_myds->myconn->options.session_track_state_sent = true;
+		set_previous_status_mode3();
+		NEXT_IMMEDIATE_NEW(SETTING_SESSION_TRACK_STATE);
+	}
+
+	return false;
+}
 
 bool MySQL_Session::handler_again___verify_multiple_variables(MySQL_Connection* myconn) {
 	for (auto i = 0; i < SQL_NAME_LAST_LOW_WM; i++) {
@@ -2753,6 +2772,20 @@ bool MySQL_Session::handler_again___status_SETTING_SESSION_TRACK_GTIDS(int *_rc)
 	return ret;
 }
 
+bool MySQL_Session::handler_again___status_SETTING_SESSION_TRACK_VARIABLES(int *_rc) {
+	bool ret=false;
+	assert(mybe->server_myds->myconn);
+	ret = handler_again___status_SETTING_GENERIC_VARIABLE(_rc, (char *)"session_track_system_variables", "*", false);
+	return ret;
+}
+
+bool MySQL_Session::handler_again___status_SETTING_SESSION_TRACK_STATE(int *_rc) {
+	bool ret=false;
+	assert(mybe->server_myds->myconn);
+	ret = handler_again___status_SETTING_GENERIC_VARIABLE(_rc, (char *)"session_track_state_change", "ON", false);
+	return ret;
+}
+
 bool MySQL_Session::handler_again___status_CHANGING_SCHEMA(int *_rc) {
 	bool ret=false;
 	//fprintf(stderr,"CHANGING_SCHEMA\n");
@@ -4862,6 +4895,41 @@ void MySQL_Session::handler_rc0_Process_GTID(MySQL_Connection *myconn) {
 	}
 }
 
+void MySQL_Session::handler_rc0_Process_Variables(MySQL_Connection *myconn) {
+	std::unordered_map<string, string> var_map;
+
+	if(myconn->get_variables(var_map)) {
+		std::string variable;
+		std::string value;
+
+		for (int idx = 0 ; idx < SQL_NAME_LAST_HIGH_WM ; idx++) {
+			variable = mysql_tracked_variables[idx].set_variable_name;
+
+			auto itr = var_map.find(variable);
+			if(itr != var_map.end()) {
+				value = itr->second;
+				proxy_debug(PROXY_DEBUG_MYSQL_CONNECTION, 7, "Session=%p, backend=%p. Notification for session_track_system_variables: variable=%s, value=%s\n", this, this->mybe, variable.c_str(), value.c_str());
+
+				const MARIADB_CHARSET_INFO *ci = NULL;
+				if (variable == "character_set_results" || variable == "character_set_connection" ||
+						variable == "character_set_client" || variable == "character_set_database") {
+					ci = proxysql_find_charset_name(value.c_str());
+				}
+				else if (variable == "collation_connection") {
+					ci = proxysql_find_charset_collate(value.c_str());
+				}
+
+				if (ci) {
+					value = std::to_string(ci->nr);
+				}
+
+				mysql_variables.client_set_value(this, idx, value);
+				mysql_variables.server_set_value(this, idx, value.c_str());
+			}
+		}
+	}
+}
+
 void MySQL_Session::handler_KillConnectionIfNeeded() {
 	if ( // two conditions
 		// If the server connection is in a non-idle state (ASYNC_IDLE), and the current time is greater than or equal to mybe->server_myds->wait_until
@@ -5059,6 +5127,10 @@ int MySQL_Session::handler() {
 									goto handler_again;
 								}
 
+								if (handler_again___verify_backend_session_track_variables()) {
+									goto handler_again;
+								}
+
 								// Optimize network traffic when we can use 'SET NAMES'
 								if (verify_set_names(this)) {
 									goto handler_again;
@@ -5139,6 +5211,8 @@ int MySQL_Session::handler() {
 
 					handler_rc0_Process_GTID(myconn);
 
+					handler_rc0_Process_Variables(myconn);
+
 					// if we are locked on hostgroup, the value of autocommit is copied from the backend connection
 					// see bug #3549
 					if (locked_on_hostgroup >= 0) {
@@ -5435,6 +5509,12 @@ bool MySQL_Session::handler_again___multiple_statuses(int *rc) {
 		case SETTING_SESSION_TRACK_GTIDS:
 			ret = handler_again___status_SETTING_SESSION_TRACK_GTIDS(rc);
 			break;
+		case SETTING_SESSION_TRACK_VARIABLES:
+			ret = handler_again___status_SETTING_SESSION_TRACK_VARIABLES(rc);
+			break;
+		case SETTING_SESSION_TRACK_STATE:
+			ret = handler_again___status_SETTING_SESSION_TRACK_STATE(rc);
+			break;
 		case SETTING_SET_NAMES:
 			ret = handler_again___status_CHANGING_CHARSET(rc);
 			break;

lib/MySQL_Thread.cpp

@@ -511,6 +511,7 @@ static char * mysql_thread_variables_names[]= {
 	(char *)"proxy_protocol_networks",
 	(char *)"protocol_compression_level",
 	(char *)"ignore_min_gtid_annotations",
+	(char *)"session_track_variables",
 	NULL
 };
 
@@ -1151,6 +1152,8 @@ MySQL_Threads_Handler::MySQL_Threads_Handler() {
 	variables.data_packets_history_size=0;
 	variables.protocol_compression_level=3;
 	variables.ignore_min_gtid_annotations=false;
+	variables.session_track_variables=session_track_variables::ENABLED;
+
 	// status variables
 	status_variables.mirror_sessions_current=0;
 	__global_MySQL_Thread_Variables_version=1;
@@ -2330,7 +2333,7 @@ char ** MySQL_Threads_Handler::get_variables_list() {
 		VariablesPointers_int["eventslog_format"] = make_tuple(&variables.eventslog_format, 0, 0, true);
 		VariablesPointers_int["wait_timeout"]     = make_tuple(&variables.wait_timeout,     0, 0, true);
 		VariablesPointers_int["data_packets_history_size"] = make_tuple(&variables.data_packets_history_size, 0, 0, true);
-
+		VariablesPointers_int["session_track_variables"]   = make_tuple(&variables.session_track_variables,   0, 1, false);
 	}
 
 
@@ -4277,6 +4280,7 @@ void MySQL_Thread::refresh_variables() {
 	REFRESH_VARIABLE_INT(handle_warnings);
 	REFRESH_VARIABLE_INT(evaluate_replication_lag_on_servers_load);
 	REFRESH_VARIABLE_BOOL(ignore_min_gtid_annotations);
+	REFRESH_VARIABLE_INT(session_track_variables);
 #ifdef DEBUG
 	REFRESH_VARIABLE_BOOL(session_debug);
 #endif /* DEBUG */

lib/mysql_connection.cpp

@@ -452,6 +452,8 @@ MySQL_Connection::MySQL_Connection() {
 	options.init_connect_sent=false;
 	options.session_track_gtids = NULL;
 	options.session_track_gtids_sent = false;
+	options.session_track_variables_sent = false;
+	options.session_track_state_sent = false;
 	options.ldap_user_variable=NULL;
 	options.ldap_user_variable_value=NULL;
 	options.ldap_user_variable_sent=false;
@@ -3082,6 +3084,8 @@ void MySQL_Connection::reset() {
 		options.session_track_gtids = NULL;
 		options.session_track_gtids_sent = false;
 	}
+	options.session_track_variables_sent = false;
+	options.session_track_state_sent = false;
 }
 
 bool MySQL_Connection::get_gtid(char *buff, uint64_t *trx_id) {
@@ -3116,6 +3120,48 @@ bool MySQL_Connection::get_gtid(char *buff, uint64_t *trx_id) {
 	return ret;
 }
 
+bool MySQL_Connection::get_variables(std::unordered_map<string, string>& variables) {
+	bool ret = false;
+
+	if ((mysql != nullptr)
+		&& (mysql->net.last_errno == 0)
+		&& (mysql->server_status & SERVER_SESSION_STATE_CHANGED)) {
+		// when there is no error and status changed
+		const char *data;
+		size_t length;
+
+		if (mysql_session_track_get_first(mysql, SESSION_TRACK_SYSTEM_VARIABLES, &data, &length) == 0) {
+			string var_name(data, length);
+			string val;
+
+			// get_first() returns a variable_name
+			// get_next() will return the value
+			bool expect_value = true;
+
+			while (mysql_session_track_get_next(mysql, SESSION_TRACK_SYSTEM_VARIABLES, &data, &length) == 0) {
+				if (expect_value) {
+					val = string(data, length);
+					variables[var_name] = val;
+					// got a value in this iteration
+					// in the next iteration, we have to expect a variable_name
+					expect_value = false;
+				} else {
+					var_name = string(data, length);
+					// got a variable_name in this iteration
+					// in the next iteration, we have to expect the value of this variable
+					expect_value = true;
+				}
+			}
+
+			// update counters
+			// __sync_fetch_and_add(&myds->sess->thread->status_variables.stvar[st_var_gtid_session_collected],1);
+			ret = true;
+		}
+	}
+
+	return ret;
+}
+
 void MySQL_Connection::set_ssl_params(MYSQL *mysql, MySQLServers_SslParams *ssl_params) {
 	if (ssl_params == NULL) {
 		mysql_ssl_set(mysql,

test/tap/groups/groups.json

@@ -229,5 +229,6 @@
   "test_ssl_fast_forward-2_libmysql-t": [ "default-g4", "mysql-auto_increment_delay_multiplex=0-g4", "mysql-multiplexing=false-g4", "mysql-query_digests=0-g4", "mysql-query_digests_keep_comment=1-g4" ],
   "test_ssl_fast_forward-3_libmariadb-t": [ "default-g4", "mysql-auto_increment_delay_multiplex=0-g4", "mysql-multiplexing=false-g4", "mysql-query_digests=0-g4", "mysql-query_digests_keep_comment=1-g4" ],
   "test_ssl_fast_forward-3_libmysql-t": [ "default-g4", "mysql-auto_increment_delay_multiplex=0-g4", "mysql-multiplexing=false-g4", "mysql-query_digests=0-g4", "mysql-query_digests_keep_comment=1-g4" ],
-  "test_ignore_min_gtid-t": [ "default-g4", "mysql-auto_increment_delay_multiplex=0-g4", "mysql-multiplexing=false-g4", "mysql-query_digests=0-g4", "mysql-query_digests_keep_comment=1-g4" ]
+  "test_ignore_min_gtid-t": [ "default-g4", "mysql-auto_increment_delay_multiplex=0-g4", "mysql-multiplexing=false-g4", "mysql-query_digests=0-g4", "mysql-query_digests_keep_comment=1-g4" ],
+  "mysql-track_system_variables-t" : [ "default-g4","mysql-auto_increment_delay_multiplex=0-g4","mysql-multiplexing=false-g4","mysql-query_digests=0-g4","mysql-query_digests_keep_comment=1-g4" ]
 }