Added warnings handling

include/MySQL_Session.h

@@ -274,6 +274,7 @@ class MySQL_Session
 	int to_process;
 	int pending_connect;
 	enum proxysql_session_type session_type;
+	int warning_in_hg;
 
 	// bool
 	bool autocommit;
@@ -379,6 +380,7 @@ class MySQL_Session
 	bool has_any_backend();
 	void detected_broken_connection(const char *file, unsigned int line, const char *func, const char *action, MySQL_Connection *myconn, int myerr, const char *message, bool verbose=false);
 	void generate_status_one_hostgroup(int hid, std::string& s);
+	void reset_warning_hostgroup_flag_and_release_connection();
 	friend void SQLite3_Server_session_handler(MySQL_Session *sess, void *_pa, PtrSize_t *pkt);
 };
 

include/MySQL_Thread.h

@@ -581,6 +581,7 @@ class MySQL_Threads_Handler
 		char * ssl_p2s_crlpath;
 		int query_cache_size_MB;
 		int query_cache_soft_ttl_pct;
+		int query_cache_with_warnings_support;
 		int min_num_servers_lantency_awareness;
 		int aurora_max_lag_ms_only_read_from_replicas;
 		bool stats_time_backend_query;

include/mysql_connection.h

@@ -19,6 +19,7 @@ using json = nlohmann::json;
 #define STATUS_MYSQL_CONNECTION_FOUND_ROWS           0x00000200
 #define STATUS_MYSQL_CONNECTION_NO_MULTIPLEX_HG      0x00000400
 #define STATUS_MYSQL_CONNECTION_HAS_SAVEPOINT        0x00000800
+#define STATUS_MYSQL_CONNECTION_HAS_WARNINGS         0x00001000
 
 class Variable {
 public:

include/proxysql_structs.h

@@ -869,6 +869,7 @@ __thread int mysql_thread___client_host_error_counts;
 /* variables used for Query Cache */
 __thread int mysql_thread___query_cache_size_MB;
 __thread int mysql_thread___query_cache_soft_ttl_pct;
+__thread int mysql_thread___query_cache_with_warnings_support;
 
 /* variables used for SSL , from proxy to server (p2s) */
 __thread char * mysql_thread___ssl_p2s_ca;
@@ -1036,6 +1037,7 @@ extern __thread int mysql_thread___client_host_error_counts;
 /* variables used for Query Cache */
 extern __thread int mysql_thread___query_cache_size_MB;
 extern __thread int mysql_thread___query_cache_soft_ttl_pct;
+extern __thread int mysql_thread___query_cache_with_warnings_support;
 
 /* variables used for SSL , from proxy to server (p2s) */
 extern __thread char * mysql_thread___ssl_p2s_ca;

lib/MySQL_Protocol.cpp

@@ -3001,11 +3001,13 @@ void MySQL_ResultSet::add_eof() {
 		//PSarrayOUT->add(pkt.ptr,pkt.size);
 		//sid++;
 		//resultset_size+=pkt.size;
-
+
+		// warnings count will only be sent to the client if mysql-query_digests is enabled
+		const bool warnings_enabled = myds->sess->CurrentQuery.QueryParserArgs.digest_text != nullptr;
 		if (deprecate_eof_active) {
 			PtrSize_t pkt;
 			buffer_to_PSarrayOut();
-			myprot->generate_pkt_OK(false, &pkt.ptr, &pkt.size, sid, 0, 0, setStatus, 0, NULL, true);
+			myprot->generate_pkt_OK(false, &pkt.ptr, &pkt.size, sid, 0, 0, setStatus, (warnings_enabled ? mysql->warning_count : 0), NULL, true);
 			PSarrayOUT.add(pkt.ptr, pkt.size);
 			resultset_size += pkt.size;
 		}
@@ -3015,7 +3017,7 @@ void MySQL_ResultSet::add_eof() {
 			// note that EOF is added on a packet on its own, instead of using a buffer,
 			// so that can be removed using remove_last_eof()
 			buffer_to_PSarrayOut();
-			myprot->generate_pkt_EOF(false, NULL, NULL, sid, 0, setStatus, this);
+			myprot->generate_pkt_EOF(false, NULL, NULL, sid, (warnings_enabled ? mysql->warning_count : 0), setStatus, this);
 			resultset_size += 9;
 			buffer_to_PSarrayOut();
 		}

lib/MySQL_Session.cpp

@@ -588,6 +588,7 @@ MySQL_Session::MySQL_Session() {
 	mirrorPkt.ptr=NULL;
 	mirrorPkt.size=0;
 	set_status(session_status___NONE);
+	warning_in_hg = -1;
 
 	idle_since = 0;
 	transaction_started_at = 0;
@@ -640,6 +641,7 @@ void MySQL_Session::reset() {
 	autocommit_handled=false;
 	sending_set_autocommit=false;
 	autocommit_on_hostgroup=-1;
+	warning_in_hg = -1;
 	current_hostgroup=-1;
 	default_hostgroup=-1;
 	locked_on_hostgroup=-1;
@@ -923,7 +925,7 @@ bool MySQL_Session::handler_CommitRollback(PtrSize_t *pkt) {
 	// specific connection, for that, we update 'current_hostgroup' with the first active transaction we are
 	// able to find. If more transactions are simultaneously open for the session, more 'COMMIT|ROLLBACK'
 	// commands are required to be issued by the client to continue ending transactions.
-	unsigned int hg = FindOneActiveTransaction(true);
+	int hg = FindOneActiveTransaction(true);
 	if (hg != -1) {
 		// there is an active transaction, we must forward the request
 		current_hostgroup = hg;
@@ -1124,6 +1126,7 @@ void MySQL_Session::generate_proxysql_internal_session_json(json &j) {
 	j["last_HG_affected_rows"] = last_HG_affected_rows;
 	j["active_transactions"] = active_transactions;
 	j["transaction_time_ms"] = thread->curtime - transaction_started_at;
+	j["warning_in_hg"] = warning_in_hg;
 	j["gtid"]["hid"] = gtid_hid;
 	j["gtid"]["last"] = ( strlen(gtid_buf) ? gtid_buf : "" );
 	j["qpo"]["create_new_connection"] = qpo->create_new_conn;
@@ -1246,6 +1249,7 @@ void MySQL_Session::generate_proxysql_internal_session_json(json &j) {
 				j["backends"][i]["conn"]["status"]["no_multiplex_HG"] = _myconn->get_status(STATUS_MYSQL_CONNECTION_NO_MULTIPLEX_HG);
 				j["backends"][i]["conn"]["status"]["compression"] = _myconn->get_status(STATUS_MYSQL_CONNECTION_COMPRESSION);
 				j["backends"][i]["conn"]["status"]["prepared_statement"] = _myconn->get_status(STATUS_MYSQL_CONNECTION_PREPARED_STATEMENT);
+				j["backends"][i]["conn"]["status"]["has_warnings"] = _myconn->get_status(STATUS_MYSQL_CONNECTION_HAS_WARNINGS);
 				j["backends"][i]["conn"]["MultiplexDisabled"] = _myconn->MultiplexDisabled();
 				j["backends"][i]["conn"]["ps"]["backend_stmt_to_global_ids"] = _myconn->local_stmts->backend_stmt_to_global_ids;
 				j["backends"][i]["conn"]["ps"]["global_stmt_to_backend_ids"] = _myconn->local_stmts->global_stmt_to_backend_ids;
@@ -1538,19 +1542,41 @@ bool MySQL_Session::handler_special_queries(PtrSize_t *pkt) {
 			return true;
 		}
 	}
-	if ( (pkt->size == 18) && (strncasecmp((char *)"SHOW WARNINGS",(char *)pkt->ptr+5,13)==0) ) {
-		SQLite3_result * resultset=new SQLite3_result(3);
-		resultset->add_column_definition(SQLITE_TEXT,"Level");
-		resultset->add_column_definition(SQLITE_TEXT,"Code");
-		resultset->add_column_definition(SQLITE_TEXT,"Message");
+	// if query digest is disabled, warnings in ProxySQL are also deactivated, 
+	// resulting in an empty response being sent to the client.
+	if ((pkt->size == 18) && (strncasecmp((char*)"SHOW WARNINGS", (char*)pkt->ptr + 5, 13) == 0) &&
+		CurrentQuery.QueryParserArgs.digest_text == nullptr) {
+		SQLite3_result* resultset = new SQLite3_result(3);
+		resultset->add_column_definition(SQLITE_TEXT, "Level");
+		resultset->add_column_definition(SQLITE_TEXT, "Code");
+		resultset->add_column_definition(SQLITE_TEXT, "Message");
 		SQLite3_to_MySQL(resultset, NULL, 0, &client_myds->myprot, false, deprecate_eof_active);
 		delete resultset;
-		client_myds->DSS=STATE_SLEEP;
-		status=WAITING_CLIENT_DATA;
-		if (mirror==false) {
+		client_myds->DSS = STATE_SLEEP;
+		status = WAITING_CLIENT_DATA;
+		if (mirror == false) {
 			RequestEnd(NULL);
 		}
-		l_free(pkt->size,pkt->ptr);
+		l_free(pkt->size, pkt->ptr);
+		return true;
+	}
+	// if query digest is disabled, warnings in ProxySQL are also deactivated, 
+	// resulting in zero warning count sent to the client.
+	if ((pkt->size == 27) && (strncasecmp((char*)"SHOW COUNT(*) WARNINGS", (char*)pkt->ptr + 5, 22) == 0) &&
+		CurrentQuery.QueryParserArgs.digest_text == nullptr) {
+		SQLite3_result* resultset = new SQLite3_result(1);
+		resultset->add_column_definition(SQLITE_TEXT, "@@session.warning_count");
+		char* pta[1];
+		pta[0] = (char*)"0";
+		resultset->add_row(pta);
+		SQLite3_to_MySQL(resultset, NULL, 0, &client_myds->myprot, false, deprecate_eof_active);
+		delete resultset;
+		client_myds->DSS = STATE_SLEEP;
+		status = WAITING_CLIENT_DATA;
+		if (mirror == false) {
+			RequestEnd(NULL);
+		}
+		l_free(pkt->size, pkt->ptr);
 		return true;
 	}
 	// 'LOAD DATA LOCAL INFILE' is unsupported. We report an specific error to inform clients about this fact. For more context see #833.
@@ -5985,26 +6011,82 @@ bool MySQL_Session::handler___status_WAITING_CLIENT_DATA___STATE_SLEEP___MYSQL_C
 
 	if (pkt->size > (unsigned int) mysql_thread___max_allowed_packet) {
 		handler_WCD_SS_MCQ_qpo_LargePacket(pkt);
+		reset_warning_hostgroup_flag_and_release_connection();
 		return true;
 	}
 
 	if (qpo->OK_msg) {
 		handler_WCD_SS_MCQ_qpo_OK_msg(pkt);
+		reset_warning_hostgroup_flag_and_release_connection();
 		return true;
 	}
 
 	if (qpo->error_msg) {
 		handler_WCD_SS_MCQ_qpo_error_msg(pkt);
+		reset_warning_hostgroup_flag_and_release_connection();
 		return true;
 	}
 
 	if (prepared) {	// for prepared statement we exit here
+		reset_warning_hostgroup_flag_and_release_connection();
 		goto __exit_set_destination_hostgroup;
 	}
 
 	// handle here #509, #815 and #816
 	if (CurrentQuery.QueryParserArgs.digest_text) {
 		char *dig=CurrentQuery.QueryParserArgs.digest_text;
+		const size_t dig_len=strlen(dig);
+		if ((dig_len == 13) && (strncasecmp(dig, "SHOW WARNINGS", 13) == 0)) {
+			proxy_debug(PROXY_DEBUG_MYSQL_COM, 5, "Intercepted '%s'\n", dig);
+			if (warning_in_hg > -1) {
+				proxy_debug(PROXY_DEBUG_MYSQL_COM, 5, "Changing current_hostgroup to '%d'\n", warning_in_hg);
+				current_hostgroup = warning_in_hg;
+				//warning_in_hg = -1;
+				return false;
+			} else {
+				proxy_debug(PROXY_DEBUG_MYSQL_COM, 5, "No warnings were detected in the previous query. Sending an empty response.\n");
+				std::unique_ptr<SQLite3_result> resultset(new SQLite3_result(3));
+				resultset->add_column_definition(SQLITE_TEXT, "Level");
+				resultset->add_column_definition(SQLITE_TEXT, "Code");
+				resultset->add_column_definition(SQLITE_TEXT, "Message");
+				SQLite3_to_MySQL(resultset.get(), NULL, 0, &client_myds->myprot, false, (client_myds->myconn->options.client_flag & CLIENT_DEPRECATE_EOF));
+				client_myds->DSS = STATE_SLEEP;
+				status = WAITING_CLIENT_DATA;
+				if (mirror == false) {
+					RequestEnd(NULL);
+				}
+				l_free(pkt->size, pkt->ptr);
+				return true;
+			}
+		} 
+
+		if ((dig_len == 22) && (strncasecmp(dig, "SHOW COUNT(*) WARNINGS", 22) == 0)) {
+			proxy_debug(PROXY_DEBUG_MYSQL_COM, 5, "Intercepted '%s'\n", dig);
+			if (warning_in_hg > -1) {
+				proxy_debug(PROXY_DEBUG_MYSQL_COM, 5, "Changing current_hostgroup to '%d'\n", warning_in_hg);
+				current_hostgroup = warning_in_hg;
+				//warning_in_hg = -1;
+				return false;
+			} else {
+				proxy_debug(PROXY_DEBUG_MYSQL_COM, 5, "No warnings were detected in the previous query. Sending an empty response.\n");
+				std::unique_ptr<SQLite3_result> resultset(new SQLite3_result(1));
+				resultset->add_column_definition(SQLITE_TEXT, "@@session.warning_count");
+				char* pta[1];
+				pta[0] = (char*)"0";
+				resultset->add_row(pta);
+				SQLite3_to_MySQL(resultset.get(), NULL, 0, &client_myds->myprot, false, (client_myds->myconn->options.client_flag & CLIENT_DEPRECATE_EOF));
+				client_myds->DSS = STATE_SLEEP;
+				status = WAITING_CLIENT_DATA;
+				if (mirror == false) {
+					RequestEnd(NULL);
+				}
+				l_free(pkt->size, pkt->ptr);
+				return true;
+			}
+		}
+
+		reset_warning_hostgroup_flag_and_release_connection();
+
 		unsigned int nTrx=NumActiveTransactions();
 		if ((locked_on_hostgroup == -1) && (strncasecmp(dig,(char *)"SET ",4)==0)) {
 			// this code is executed only if locked_on_hostgroup is not set yet
@@ -7288,7 +7370,9 @@ void MySQL_Session::MySQL_Result_to_MySQL_wire(MYSQL *mysql, MySQL_ResultSet *My
 		assert(resultset_completed); // the resultset should always be completed if MySQL_Result_to_MySQL_wire is called
 		if (transfer_started==false) { // we have all the resultset when MySQL_Result_to_MySQL_wire was called
 			if (qpo && qpo->cache_ttl>0 && com_field_list==false) { // the resultset should be cached
-				if (mysql_errno(mysql)==0) { // no errors
+				if (mysql_errno(mysql)==0 &&
+					(mysql_warning_count(mysql)==0 || 
+					 mysql_thread___query_cache_with_warnings_support==1)) { // no errors
 					if (
 						(qpo->cache_empty_result==1)
 						|| (
@@ -8069,3 +8153,20 @@ void MySQL_Session::generate_status_one_hostgroup(int hid, std::string& s) {
 	s = j_res.dump();
 	delete resultset;
 }
+
+void MySQL_Session::reset_warning_hostgroup_flag_and_release_connection()
+{
+	if (warning_in_hg > -1) {
+		// if we've reached this point, it means that warning was found in the previous query, but the
+		// current executed query is not 'SHOW WARNINGS' or 'SHOW COUNT(*) FROM WARNINGS', so we can safely reset warning_in_hg and 
+		// return connection back to the connection pool.
+		MySQL_Backend* _mybe = NULL;
+		_mybe = find_backend(warning_in_hg);
+		MySQL_Data_Stream* myds = _mybe->server_myds;
+		myds->myconn->set_status(false, STATUS_MYSQL_CONNECTION_HAS_WARNINGS);
+		if ((myds->myconn->reusable == true) && myds->myconn->IsActiveTransaction() == false && myds->myconn->MultiplexDisabled() == false) {
+			myds->return_MySQL_Connection_To_Pool();
+		}
+		warning_in_hg = -1;
+	}
+}

lib/MySQL_Thread.cpp

@@ -553,6 +553,7 @@ static char * mysql_thread_variables_names[]= {
 	(char *)"long_query_time",
 	(char *)"query_cache_size_MB",
 	(char *)"query_cache_soft_ttl_pct",
+	(char *)"query_cache_with_warnings_support",
 	(char *)"ping_interval_server_msec",
 	(char *)"ping_timeout_server",
 	(char *)"default_schema",
@@ -1149,6 +1150,7 @@ MySQL_Threads_Handler::MySQL_Threads_Handler() {
 	variables.long_query_time=1000;
 	variables.query_cache_size_MB=256;
 	variables.query_cache_soft_ttl_pct=0;
+	variables.query_cache_with_warnings_support=0;
 	variables.init_connect=NULL;
 	variables.ldap_user_variable=NULL;
 	variables.add_ldap_user_comment=NULL;
@@ -2285,6 +2287,8 @@ char ** MySQL_Threads_Handler::get_variables_list() {
 		VariablesPointers_int["max_transaction_time"]      = make_tuple(&variables.max_transaction_time,      1000,  20*24*3600*1000, false);
 		VariablesPointers_int["query_cache_size_mb"]       = make_tuple(&variables.query_cache_size_MB,          0,       1024*10240, false);
 		VariablesPointers_int["query_cache_soft_ttl_pct"]  = make_tuple(&variables.query_cache_soft_ttl_pct,     0,              100, false);
+		VariablesPointers_int["query_cache_with_warnings_support"] = make_tuple(&variables.query_cache_with_warnings_support,	 0,				   1, false);
+
 #ifdef IDLE_THREADS
 		VariablesPointers_int["session_idle_ms"]           = make_tuple(&variables.session_idle_ms,              1,        3600*1000, false);
 #endif // IDLE_THREADS
@@ -4033,6 +4037,7 @@ void MySQL_Thread::refresh_variables() {
 	mysql_thread___long_query_time=GloMTH->get_variable_int((char *)"long_query_time");
 	mysql_thread___query_cache_size_MB=GloMTH->get_variable_int((char *)"query_cache_size_MB");
 	mysql_thread___query_cache_soft_ttl_pct=GloMTH->get_variable_int((char *)"query_cache_soft_ttl_pct");
+	mysql_thread___query_cache_with_warnings_support =GloMTH->get_variable_int((char*)"query_cache_with_warnings_support");
 	mysql_thread___ping_interval_server_msec=GloMTH->get_variable_int((char *)"ping_interval_server_msec");
 	mysql_thread___ping_timeout_server=GloMTH->get_variable_int((char *)"ping_timeout_server");
 	mysql_thread___shun_on_failures=GloMTH->get_variable_int((char *)"shun_on_failures");

lib/Query_Cache.cpp

@@ -728,8 +728,34 @@ bool Query_Cache::set(uint64_t user_hash, const unsigned char *kp, uint32_t kl,
 		if (hdr.pkt_length < 9 && *payload == 0xfe) {
 			if (deprecate_eof_active) {
 				entry->ok_pkt_offset = it - vp;
+
+				// Reset the warning flags to zero before storing resultset in the cache
+				// Reason: When a warning flag is set, it may prompt the client to invoke "SHOW WARNINGS" or "SHOW COUNT(*) FROM WARNINGS". 
+				// However, when retrieving data from the cache, it's possible that there are no warnings present
+				// that might be associated with previous interactions.
+				unsigned char* payload_temp = payload+1;
+
+				// skip affected_rows
+				payload_temp += mysql_decode_length(payload_temp, nullptr);
+
+				// skip last_insert_id
+				payload_temp += mysql_decode_length(payload_temp, nullptr);
+
+				// skip stats_flags
+				payload_temp += sizeof(uint16_t);
+
+				uint16_t warnings = 0;
+				memcpy(payload_temp, &warnings, sizeof(uint16_t));
+
 			} else {
 				entry->row_eof_pkt_offset = it - vp;
+
+				// Reset the warning flags to zero before storing resultset in the cache
+				// Reason: When a warning flag is set, it may prompt the client to invoke "SHOW WARNINGS" or "SHOW COUNT(*) FROM WARNINGS".  
+				// However, when retrieving data from the cache, it's possible that there are no warnings present
+				// that might be associated with previous interactions.
+				uint16_t warnings = 0;
+				memcpy((payload + 1), &warnings, sizeof(uint16_t));
 			}
 			break;
 		} else {

lib/mysql_connection.cpp

@@ -2448,7 +2448,10 @@ bool MySQL_Connection::MultiplexDisabled(bool check_delay_token) {
 // status_flags stores information about the status of the connection
 // can be used to determine if multiplexing can be enabled or not
 	bool ret=false;
-	if (status_flags & (STATUS_MYSQL_CONNECTION_TRANSACTION|STATUS_MYSQL_CONNECTION_USER_VARIABLE|STATUS_MYSQL_CONNECTION_PREPARED_STATEMENT|STATUS_MYSQL_CONNECTION_LOCK_TABLES|STATUS_MYSQL_CONNECTION_TEMPORARY_TABLE|STATUS_MYSQL_CONNECTION_GET_LOCK|STATUS_MYSQL_CONNECTION_NO_MULTIPLEX|STATUS_MYSQL_CONNECTION_SQL_LOG_BIN0|STATUS_MYSQL_CONNECTION_FOUND_ROWS|STATUS_MYSQL_CONNECTION_NO_MULTIPLEX_HG|STATUS_MYSQL_CONNECTION_HAS_SAVEPOINT) ) {
+	if (status_flags & (STATUS_MYSQL_CONNECTION_TRANSACTION | STATUS_MYSQL_CONNECTION_USER_VARIABLE | STATUS_MYSQL_CONNECTION_PREPARED_STATEMENT |
+		STATUS_MYSQL_CONNECTION_LOCK_TABLES | STATUS_MYSQL_CONNECTION_TEMPORARY_TABLE | STATUS_MYSQL_CONNECTION_GET_LOCK | STATUS_MYSQL_CONNECTION_NO_MULTIPLEX |
+		STATUS_MYSQL_CONNECTION_SQL_LOG_BIN0 | STATUS_MYSQL_CONNECTION_FOUND_ROWS | STATUS_MYSQL_CONNECTION_NO_MULTIPLEX_HG |
+		STATUS_MYSQL_CONNECTION_HAS_SAVEPOINT | STATUS_MYSQL_CONNECTION_HAS_WARNINGS) ) {
 		ret=true;
 	}
 	if (check_delay_token && auto_increment_delay_token) return true;
@@ -2569,6 +2572,34 @@ void MySQL_Connection::ProcessQueryAndSetStatusFlags(char *query_digest_text) {
 			}
 		}
 	}
+	// checking warnings and disabling multiplexing will be effective only when the mysql-query_digests is enabled
+	if (myds->sess->CurrentQuery.QueryParserArgs.digest_text) {
+		if (get_status(STATUS_MYSQL_CONNECTION_HAS_WARNINGS) == false) {
+			if (mysql_warning_count(this->mysql) > 0) {
+				if (myds && myds->sess) {
+					// 'warning_in_hg' will be used if the next query is 'SHOW WARNINGS' or
+					// 'SHOW COUNT(*) WARNINGS'
+					myds->sess->warning_in_hg = myds->sess->current_hostgroup;
+					// enabling multiplexing
+					set_status(true, STATUS_MYSQL_CONNECTION_HAS_WARNINGS);
+				}
+			}
+		} else { // reset warning_in_hg 
+			const char* dig = myds->sess->CurrentQuery.QueryParserArgs.digest_text;
+			const size_t dig_len = strlen(dig);
+			// disable multiplexing and reset the 'warning_in_hg' flag only when the current executed query is not 
+			// 'SHOW WARNINGS' or 'SHOW COUNT(*) WARNINGS', as these queries do not clear the warning message list
+			// on backend.
+			if (!((dig_len == 22 && strncasecmp(dig, "SHOW COUNT(*) WARNINGS", 22) == 0) ||
+				(dig_len == 13 && strncasecmp(dig, "SHOW WARNINGS", 13) == 0))) {
+				if (myds && myds->sess) {
+					myds->sess->warning_in_hg = -1;
+				}
+				// disabling multiplexing
+				set_status(false, STATUS_MYSQL_CONNECTION_HAS_WARNINGS);
+			}
+		}
+	}
 	if (get_status(STATUS_MYSQL_CONNECTION_USER_VARIABLE)==false) { // we search for variables only if not already set
 //			if (
 //				strncasecmp(query_digest_text,"SELECT @@tx_isolation", strlen("SELECT @@tx_isolation"))