MySQL: Track session-specific system variables changes

- Configure `session_track_system_variables` and `session_track_state_change`
  in all backend connections.
- 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_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
 };
 

lib/MySQL_Session.cpp

@@ -1953,6 +1953,21 @@ bool MySQL_Session::handler_again___verify_backend_session_track_gtids() {
 	return ret;
 }
 
+bool MySQL_Session::handler_again___verify_backend_session_track_variables() {
+	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 +2768,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 +4891,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 +5123,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 +5207,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 +5505,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_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" ]
 }

test/tap/tests/mysql-track_system_variables-t.cpp

@@ -0,0 +1,87 @@
+/**
+ * @file test_track_system_variables-t.cpp
+ * @brief This test verifies that ProxySQL properly tracks session-specific
+ *   system variables across the backend connections.
+ */
+
+#include <stdio.h>
+#include <stdlib.h>
+#include "json.hpp"
+#include "mysql.h"
+#include "tap.h"
+#include "command_line.h"
+#include "utils.h"
+
+using nlohmann::json;
+
+int main(int argc, char** argv) {
+	CommandLine cl;
+	if (cl.getEnv()) {
+		diag("Failed to get the required environmental variables.");
+		return exit_status();
+	}
+
+	plan(1);
+
+	MYSQL* proxy = init_mysql_conn(cl.host, cl.port, cl.username, cl.password, true);
+	if (!proxy) {
+		fprintf(stderr, "File %s, line %d, Error: %s\n", __FILE__, __LINE__, mysql_error(proxy));
+		return exit_status();
+	}
+
+	MYSQL_QUERY_T(proxy, "CREATE DATABASE IF NOT EXISTS test");
+	MYSQL_QUERY_T(proxy, "SELECT 1");
+	MYSQL_RES* reset_result = mysql_store_result(proxy);
+	mysql_free_result(reset_result);
+
+	MYSQL_QUERY_T(proxy, "DROP PROCEDURE IF EXISTS test.set_innodb_lock_wait_timeout");
+	const char* create_proc =
+		"CREATE PROCEDURE test.set_innodb_lock_wait_timeout() "
+		"BEGIN "
+		"  SET innodb_lock_wait_timeout = CAST(FLOOR(50 + (RAND() * 100)) AS UNSIGNED); "
+		"END";
+
+	MYSQL_QUERY_T(proxy, create_proc);
+	MYSQL_QUERY_T(proxy, "CALL test.set_innodb_lock_wait_timeout()");
+
+	int set_value = -1;
+	MYSQL_QUERY_T(proxy, "SELECT @@innodb_lock_wait_timeout");
+	MYSQL_RES* result = mysql_store_result(proxy);
+	if (result) {
+		MYSQL_ROW row = mysql_fetch_row(result);
+		if (row) {
+			set_value = atoi(row[0]);
+		}
+		mysql_free_result(result);
+	}
+
+	MYSQL_QUERY(proxy, "PROXYSQL INTERNAL SESSION");
+	result = mysql_store_result(proxy);
+	MYSQL_ROW row = mysql_fetch_row(result);
+	auto j_session = nlohmann::json::parse(row[0]);
+	mysql_free_result(result);
+
+	int backend_value = -1;
+	int client_value = -1;
+	if (j_session.contains("backends")) {
+		for (auto& backend : j_session["backends"]) {
+			if (backend != nullptr && backend.contains("conn")) {
+				if (backend["conn"].contains("innodb_lock_wait_timeout")) {
+					backend_value = std::stoi(backend["conn"]["innodb_lock_wait_timeout"].get<std::string>());
+					break;
+				}
+			}
+		}
+	}
+	if (j_session.contains("conn")) {
+		if (j_session["conn"].contains("innodb_lock_wait_timeout")) {
+			client_value = std::stoi(j_session["conn"]["innodb_lock_wait_timeout"].get<std::string>());
+		}
+	}
+
+	ok(set_value == backend_value && set_value == client_value,
+		"Match session innodb_lock_wait_timeout value with backend & client variable list. Expected: %d, Backend: %d, Client: %d", set_value, backend_value, client_value);
+
+	mysql_close(proxy);
+	return exit_status();
+}

test/tap/tests/test_binlog_reader-t.cpp

@@ -263,6 +263,12 @@ int main(int argc, char** argv) {
 	vector<mysql_res_row> reader_1_read {};
 	vector<mysql_res_row> reader_2_read {};
 
+	// Create backend connections before clearing the stats.
+	// This is to exclude queries sent during connection setup phrase.
+	MYSQL_QUERY_T(proxysql_mysql, "DO 1");
+	MYSQL_QUERY_T(proxysql_mysql, "SELECT 1");
+	mysql_free_result(mysql_store_result(proxysql_mysql));
+
 	// Reset connection pool stats
 	int rc = mysql_query(proxysql_admin, "SELECT * FROM stats.stats_mysql_connection_pool_reset");
 	if (rc != EXIT_SUCCESS) { goto cleanup; }
@@ -279,10 +285,6 @@ int main(int argc, char** argv) {
 		rc = perform_update(proxysql_mysql, NUM_ROWS);
 		if (rc != EXIT_SUCCESS) { goto cleanup; }
 
-		MYSQL_RES* my_res = mysql_store_result(proxysql_admin);
-		vector<mysql_res_row> pre_select_rows = extract_mysql_rows(my_res);
-		mysql_free_result(my_res);
-
 		int r_row = rand() % NUM_ROWS;
 		if (r_row == 0) { r_row = 1; }