From c76e1476ff9b77b247aa981f7743ac9ceae2e4bc Mon Sep 17 00:00:00 2001 From: peterlyoo Date: Thu, 9 Jul 2026 18:22:25 +0900 Subject: [PATCH 1/4] Add destination_schema support to mysql_query_rules attributes (#4880) A query rule whose attributes JSON contains {"destination_schema": "name"} switches the session schema before backend connection selection. The connection pool matches on (username, schemaname) and issues COM_INIT_DB on schema mismatch, so matched queries transparently execute against the remapped schema. This enables per-user / per-rule logical database routing on the same cluster (multi-tenant style redirection). MySQL only; PgSQL rule creation initializes the field to NULL. --- include/query_processor.h | 7 +++++++ lib/MySQL_Query_Processor.cpp | 24 ++++++++++++++++++++++++ lib/MySQL_Session.cpp | 6 ++++++ lib/PgSQL_Query_Processor.cpp | 2 ++ lib/Query_Processor.cpp | 11 ++++++++++- 5 files changed, 49 insertions(+), 1 deletion(-) diff --git a/include/query_processor.h b/include/query_processor.h index 7c8f033eeb..9b2b2837bb 100644 --- a/include/query_processor.h +++ b/include/query_processor.h @@ -132,6 +132,7 @@ typedef struct _Query_Processor_rule_t { int log; bool apply; char* attributes; + char *destination_schema; // parsed from attributes JSON key "destination_schema" char *comment; // #643 void *regex_engine1; void *regex_engine2; @@ -175,6 +176,7 @@ class Query_Processor_Output { int log; int firewall_whitelist_mode; char *attributes; + char *destination_schema; // when set, session schema is switched before routing char *comment; // #643 bool create_new_conn; @@ -213,6 +215,7 @@ class Query_Processor_Output { error_msg=NULL; OK_msg=NULL; attributes=NULL; + destination_schema=NULL; comment=NULL; // #643 firewall_whitelist_mode = WUS_NOT_FOUND; create_new_conn=0; @@ -229,6 +232,10 @@ class Query_Processor_Output { if (attributes) { free(attributes); } + if (destination_schema) { + free(destination_schema); + destination_schema=NULL; + } if (comment) { // #643 free(comment); } diff --git a/lib/MySQL_Query_Processor.cpp b/lib/MySQL_Query_Processor.cpp index 4578f236c8..62377539a0 100644 --- a/lib/MySQL_Query_Processor.cpp +++ b/lib/MySQL_Query_Processor.cpp @@ -752,6 +752,7 @@ MySQL_Query_Processor_Rule_t* MySQL_Query_Processor::new_query_rule(int rule_id, newQR->gtid_from_hostgroup = gtid_from_hostgroup; newQR->apply = apply; newQR->attributes = (attributes ? strdup(attributes) : NULL); + newQR->destination_schema = NULL; newQR->comment = (comment ? strdup(comment) : NULL); // see issue #643 newQR->regex_engine1 = NULL; newQR->regex_engine2 = NULL; @@ -821,6 +822,17 @@ MySQL_Query_Processor_Rule_t* MySQL_Query_Processor::new_query_rule(int rule_id, proxy_error("Failed to parse flagOUTs attributes for rule_id %d : %s\n", newQR->rule_id, flagOUTs.dump().c_str()); } } + if (j_attributes.find("destination_schema") != j_attributes.end()) { + const nlohmann::json& dest_schema = j_attributes["destination_schema"]; + if (dest_schema.type() == nlohmann::json::value_t::string) { + std::string s = dest_schema; + if (s.length() > 0) { + newQR->destination_schema = strdup(s.c_str()); + } + } else { + proxy_error("Failed to parse destination_schema in JSON on attributes for rule_id %d : %s\n", newQR->rule_id, dest_schema.dump().c_str()); + } + } } } proxy_debug(PROXY_DEBUG_MYSQL_QUERY_PROCESSOR, 5, "Creating new rule in %p : rule_id:%d, active:%d, username=%s, schemaname=%s, flagIN:%d, %smatch_digest=\"%s\", %smatch_pattern=\"%s\", flagOUT:%d replace_pattern=\"%s\", destination_hostgroup:%d, apply:%d\n", newQR, newQR->rule_id, newQR->active, newQR->username, newQR->schemaname, newQR->flagIN, (newQR->negate_match_pattern ? "(!)" : ""), newQR->match_digest, (newQR->negate_match_pattern ? "(!)" : ""), newQR->match_pattern, newQR->flagOUT, newQR->replace_pattern, newQR->destination_hostgroup, newQR->apply); @@ -888,6 +900,7 @@ MySQL_Query_Processor_Rule_t* MySQL_Query_Processor::new_query_rule(const MySQL_ newQR->gtid_from_hostgroup = mqr->gtid_from_hostgroup; newQR->apply = mqr->apply; newQR->attributes = (mqr->attributes ? strdup(mqr->attributes) : NULL); + newQR->destination_schema = NULL; newQR->comment = (mqr->comment ? strdup(mqr->comment) : NULL); // see issue #643 newQR->regex_engine1 = NULL; newQR->regex_engine2 = NULL; @@ -957,6 +970,17 @@ MySQL_Query_Processor_Rule_t* MySQL_Query_Processor::new_query_rule(const MySQL_ proxy_error("Failed to parse flagOUTs attributes for rule_id %d : %s\n", newQR->rule_id, flagOUTs.dump().c_str()); } } + if (j_attributes.find("destination_schema") != j_attributes.end()) { + const nlohmann::json& dest_schema = j_attributes["destination_schema"]; + if (dest_schema.type() == nlohmann::json::value_t::string) { + std::string s = dest_schema; + if (s.length() > 0) { + newQR->destination_schema = strdup(s.c_str()); + } + } else { + proxy_error("Failed to parse destination_schema in JSON on attributes for rule_id %d : %s\n", newQR->rule_id, dest_schema.dump().c_str()); + } + } } } proxy_debug(PROXY_DEBUG_MYSQL_QUERY_PROCESSOR, 5, "Creating new rule in %p : rule_id:%d, active:%d, username=%s, schemaname=%s, flagIN:%d, %smatch_digest=\"%s\", %smatch_pattern=\"%s\", flagOUT:%d replace_pattern=\"%s\", destination_hostgroup:%d, apply:%d\n", newQR, newQR->rule_id, newQR->active, newQR->username, newQR->schemaname, newQR->flagIN, (newQR->negate_match_pattern ? "(!)" : ""), newQR->match_digest, (newQR->negate_match_pattern ? "(!)" : ""), newQR->match_pattern, newQR->flagOUT, newQR->replace_pattern, newQR->destination_hostgroup, newQR->apply); diff --git a/lib/MySQL_Session.cpp b/lib/MySQL_Session.cpp index bf143b39dd..4102c07e14 100644 --- a/lib/MySQL_Session.cpp +++ b/lib/MySQL_Session.cpp @@ -8226,6 +8226,12 @@ bool MySQL_Session::handler___status_WAITING_CLIENT_DATA___STATE_SLEEP___MYSQL_C current_hostgroup=qpo->destination_hostgroup; } } + if (qpo->destination_schema) { + // switch the session schema before backend connection selection: the + // connection pool matches on (username, schemaname) and issues + // COM_INIT_DB on schema mismatch, so the query lands on this schema + client_myds->myconn->userinfo->set_schemaname(qpo->destination_schema, strlen(qpo->destination_schema)); + } if (mysql_thread___set_query_lock_on_hostgroup == 1) { // algorithm introduced in 2.0.6 if (locked_on_hostgroup >= 0) { diff --git a/lib/PgSQL_Query_Processor.cpp b/lib/PgSQL_Query_Processor.cpp index 43cf827e77..0a98f1a05f 100644 --- a/lib/PgSQL_Query_Processor.cpp +++ b/lib/PgSQL_Query_Processor.cpp @@ -370,6 +370,7 @@ PgSQL_Query_Processor_Rule_t* PgSQL_Query_Processor::new_query_rule(int rule_id, newQR->multiplex = multiplex; newQR->apply = apply; newQR->attributes = (attributes ? strdup(attributes) : NULL); + newQR->destination_schema = NULL; // not supported for PgSQL newQR->comment = (comment ? strdup(comment) : NULL); // see issue #643 newQR->regex_engine1 = NULL; newQR->regex_engine2 = NULL; @@ -505,6 +506,7 @@ PgSQL_Query_Processor_Rule_t* PgSQL_Query_Processor::new_query_rule(const PgSQL_ newQR->multiplex = pqr->multiplex; newQR->apply = pqr->apply; newQR->attributes = (pqr->attributes ? strdup(pqr->attributes) : NULL); + newQR->destination_schema = NULL; // not supported for PgSQL newQR->comment = (pqr->comment ? strdup(pqr->comment) : NULL); // see issue #643 newQR->regex_engine1 = NULL; newQR->regex_engine2 = NULL; diff --git a/lib/Query_Processor.cpp b/lib/Query_Processor.cpp index b14c0b1b95..a174135bc9 100644 --- a/lib/Query_Processor.cpp +++ b/lib/Query_Processor.cpp @@ -390,6 +390,8 @@ static void __delete_query_rule(QP_rule_t *qr) { free(qr->OK_msg); if (qr->attributes) free(qr->attributes); + if (qr->destination_schema) + free(qr->destination_schema); if (qr->comment) free(qr->comment); if (qr->regex_engine1) { @@ -2068,7 +2070,14 @@ Query_Processor_Output* Query_Processor::process_query(TypeSession* // Note: negative hostgroup means this rule doesn't change proxy_debug(PROXY_DEBUG_MYSQL_QUERY_PROCESSOR, 5, "query rule %d has set destination hostgroup: %d\n", qr->rule_id, qr->destination_hostgroup); ret->destination_hostgroup=qr->destination_hostgroup; - } + } + if (qr->destination_schema) { + proxy_debug(PROXY_DEBUG_MYSQL_QUERY_PROCESSOR, 5, "query rule %d has set destination schema: %s\n", qr->rule_id, qr->destination_schema); + if (ret->destination_schema) { + free(ret->destination_schema); + } + ret->destination_schema=strdup(qr->destination_schema); + } if constexpr (has_process_query_extended::value) { (static_cast(this))->process_query_extended(static_cast(ret), static_cast(qr)); } From d6bc2a6b04df7592a9c34e4628488f53c5be4b87 Mon Sep 17 00:00:00 2001 From: peterlyoo Date: Thu, 9 Jul 2026 19:55:31 +0900 Subject: [PATCH 2/4] Add TAP test for destination_schema query rule routing (#4880) Verifies the handshake, COM_INIT_DB and USE schema-selection paths are all remapped by a destination_schema rule, and that behavior reverts after the rule is removed. --- test/tap/groups/groups.json | 1 + .../tap/tests/mysql-dest_schema_routing-t.cpp | 188 ++++++++++++++++++ 2 files changed, 189 insertions(+) create mode 100644 test/tap/tests/mysql-dest_schema_routing-t.cpp diff --git a/test/tap/groups/groups.json b/test/tap/groups/groups.json index 6956607612..ff1d095091 100644 --- a/test/tap/groups/groups.json +++ b/test/tap/groups/groups.json @@ -83,6 +83,7 @@ "monitor_health_unit-t" : [ "unit-tests-g1" ], "multiple_prepared_statements-t" : [ "legacy-g1","mysql-auto_increment_delay_multiplex=0-g1","mysql-multiplexing=false-g1","mysql-query_digests=0-g1","mysql-query_digests_keep_comment=1-g1","mysql84-g1","mysql90-g1","mysql95-g1" ], "mysql-fast_forward-t" : [ "legacy-g1","mysql-auto_increment_delay_multiplex=0-g1","mysql-multiplexing=false-g1","mysql-query_digests=0-g1","mysql-query_digests_keep_comment=1-g1","mysql84-g1","mysql90-g1","mysql95-g1" ], + "mysql-dest_schema_routing-t" : [ "legacy-g1","mysql-auto_increment_delay_multiplex=0-g1","mysql-multiplexing=false-g1","mysql-query_digests=0-g1","mysql-query_digests_keep_comment=1-g1","mysql84-g1","mysql90-g1","mysql95-g1" ], "mysql-init_connect-1-t" : [ "legacy-g1","mariadb10-galera-g1","mysql-auto_increment_delay_multiplex=0-g1","mysql-multiplexing=false-g1","mysql-query_digests=0-g1","mysql-query_digests_keep_comment=1-g1","mysql84-g1","mysql84-gr-g1","mysql90-g1","mysql90-gr-g1","mysql93-g1","mysql93-gr-g1","mysql95-g1","mysql95-gr-g1" ], "mysql-init_connect-2-t" : [ "legacy-g1","mariadb10-galera-g1","mysql-auto_increment_delay_multiplex=0-g1","mysql-multiplexing=false-g1","mysql-query_digests=0-g1","mysql-query_digests_keep_comment=1-g1","mysql84-g1","mysql84-gr-g1","mysql90-g1","mysql90-gr-g1","mysql93-g1","mysql93-gr-g1","mysql95-g1","mysql95-gr-g1" ], "mysql-last_insert_id-t" : [ "legacy-g1","mariadb10-galera-g1","mysql-auto_increment_delay_multiplex=0-g1","mysql-multiplexing=false-g1","mysql-query_digests=0-g1","mysql-query_digests_keep_comment=1-g1","mysql84-g1","mysql84-gr-g1","mysql90-g1","mysql90-gr-g1","mysql93-g1","mysql93-gr-g1","mysql95-g1","mysql95-gr-g1" ], diff --git a/test/tap/tests/mysql-dest_schema_routing-t.cpp b/test/tap/tests/mysql-dest_schema_routing-t.cpp new file mode 100644 index 0000000000..231087d2aa --- /dev/null +++ b/test/tap/tests/mysql-dest_schema_routing-t.cpp @@ -0,0 +1,188 @@ +/** + * @file mysql-dest_schema_routing-t.cpp + * @brief E2E test for logical db routing via mysql_query_rules.attributes + * {"destination_schema": "..."}. A matching rule switches the session schema + * before backend connection selection, so queries land on the remapped + * schema regardless of the db requested by the client. Verifies all three + * schema-selection paths (handshake db, COM_INIT_DB, USE statement) and + * that removing the rule restores the original behavior. + */ + +#include +#include + +#include "mysql.h" +#include "command_line.h" +#include "tap.h" +#include "utils.h" + +CommandLine cl; + +const char* SRC_DB = "dsr_src"; +const char* DST_DB = "dsr_dst"; +const int RULE_ID = 2; // must sort before the infra read/write split rules (3,4) + +#define MYSQL_QUERY_ON_ERR_CLEANUP(mysql, query) \ + do { \ + if (mysql_query(mysql, query)) { \ + fprintf(stderr, "File %s, line %d, Error: %s (%s)\n", __FILE__, __LINE__, mysql_error(mysql), query); \ + goto cleanup; \ + } \ + } while(0) + +/** + * @brief Run a single-value query and return the value ("" on NULL/error). + */ +std::string fetch_single(MYSQL* mysql, const char* query) { + std::string result {}; + if (mysql_query(mysql, query)) { + diag("Query failed: '%s' error: '%s'", query, mysql_error(mysql)); + return result; + } + MYSQL_RES* res = mysql_store_result(mysql); + if (res) { + MYSQL_ROW row = mysql_fetch_row(res); + if (row && row[0]) { + result = row[0]; + } + mysql_free_result(res); + } + return result; +} + +/** + * @brief Open a fresh proxy connection with 'db' as the handshake schema. + */ +MYSQL* connect_proxy(const char* db) { + MYSQL* conn = mysql_init(NULL); + if (!mysql_real_connect(conn, cl.host, cl.username, cl.password, db, cl.port, NULL, 0)) { + diag("Failed to connect to proxy (db=%s): %s", db ? db : "NULL", mysql_error(conn)); + mysql_close(conn); + return NULL; + } + return conn; +} + +int main() { + plan(8); + + if (cl.getEnv()) + return exit_status(); + + MYSQL* admin = mysql_init(NULL); + MYSQL* setup = NULL; + MYSQL* conn = NULL; + std::string val {}; + char query[512]; + + if (!mysql_real_connect(admin, cl.host, cl.admin_username, cl.admin_password, NULL, cl.admin_port, NULL, 0)) { + fprintf(stderr, "File %s, line %d, Error: %s\n", __FILE__, __LINE__, mysql_error(admin)); + return -1; + } + + // setup: create the two schemas with distinct markers, before any rule exists + setup = connect_proxy(NULL); + if (!setup) { + goto cleanup; + } + snprintf(query, sizeof(query), "CREATE DATABASE IF NOT EXISTS %s", SRC_DB); + MYSQL_QUERY_ON_ERR_CLEANUP(setup, query); + snprintf(query, sizeof(query), "CREATE DATABASE IF NOT EXISTS %s", DST_DB); + MYSQL_QUERY_ON_ERR_CLEANUP(setup, query); + snprintf(query, sizeof(query), "CREATE TABLE IF NOT EXISTS %s.marker (v VARCHAR(32))", SRC_DB); + MYSQL_QUERY_ON_ERR_CLEANUP(setup, query); + snprintf(query, sizeof(query), "CREATE TABLE IF NOT EXISTS %s.marker (v VARCHAR(32))", DST_DB); + MYSQL_QUERY_ON_ERR_CLEANUP(setup, query); + snprintf(query, sizeof(query), "DELETE FROM %s.marker", SRC_DB); + MYSQL_QUERY_ON_ERR_CLEANUP(setup, query); + snprintf(query, sizeof(query), "INSERT INTO %s.marker VALUES ('in_src')", SRC_DB); + MYSQL_QUERY_ON_ERR_CLEANUP(setup, query); + snprintf(query, sizeof(query), "DELETE FROM %s.marker", DST_DB); + MYSQL_QUERY_ON_ERR_CLEANUP(setup, query); + snprintf(query, sizeof(query), "INSERT INTO %s.marker VALUES ('in_dst')", DST_DB); + MYSQL_QUERY_ON_ERR_CLEANUP(setup, query); + // let replicas catch up: reads may be routed to a reader hostgroup + sleep(2); + + // baseline: no rule, handshake db is honored + conn = connect_proxy(SRC_DB); + if (!conn) { + goto cleanup; + } + val = fetch_single(conn, "SELECT DATABASE()"); + ok(val == SRC_DB, "baseline: DATABASE() should be '%s', got '%s'", SRC_DB, val.c_str()); + val = fetch_single(conn, "SELECT v FROM marker"); + ok(val == "in_src", "baseline: marker should be 'in_src', got '%s'", val.c_str()); + mysql_close(conn); + conn = NULL; + + // add destination_schema rule for this user (apply=0: compose with later rules) + snprintf(query, sizeof(query), "DELETE FROM mysql_query_rules WHERE rule_id=%d", RULE_ID); + MYSQL_QUERY_ON_ERR_CLEANUP(admin, query); + snprintf(query, sizeof(query), + "INSERT INTO mysql_query_rules (rule_id, active, username, apply, attributes) " + "VALUES (%d, 1, '%s', 0, '{\"destination_schema\": \"%s\"}')", RULE_ID, cl.username, DST_DB); + MYSQL_QUERY_ON_ERR_CLEANUP(admin, query); + MYSQL_QUERY_ON_ERR_CLEANUP(admin, "LOAD MYSQL QUERY RULES TO RUNTIME"); + + // path 1: handshake db + conn = connect_proxy(SRC_DB); + if (!conn) { + goto cleanup; + } + val = fetch_single(conn, "SELECT DATABASE()"); + ok(val == DST_DB, "handshake path: DATABASE() should be remapped to '%s', got '%s'", DST_DB, val.c_str()); + val = fetch_single(conn, "SELECT v FROM marker"); + ok(val == "in_dst", "handshake path: marker should be 'in_dst', got '%s'", val.c_str()); + + // path 2: COM_INIT_DB resets the schema, next query remaps again + if (mysql_select_db(conn, SRC_DB)) { + diag("mysql_select_db failed: %s", mysql_error(conn)); + } + val = fetch_single(conn, "SELECT DATABASE()"); + ok(val == DST_DB, "COM_INIT_DB path: DATABASE() should be remapped to '%s', got '%s'", DST_DB, val.c_str()); + + // path 3: USE statement resets the schema, next query remaps again + snprintf(query, sizeof(query), "USE %s", SRC_DB); + if (mysql_query(conn, query)) { + diag("USE failed: %s", mysql_error(conn)); + } + val = fetch_single(conn, "SELECT DATABASE()"); + ok(val == DST_DB, "USE path: DATABASE() should be remapped to '%s', got '%s'", DST_DB, val.c_str()); + mysql_close(conn); + conn = NULL; + + // remove the rule: behavior must revert + snprintf(query, sizeof(query), "DELETE FROM mysql_query_rules WHERE rule_id=%d", RULE_ID); + MYSQL_QUERY_ON_ERR_CLEANUP(admin, query); + MYSQL_QUERY_ON_ERR_CLEANUP(admin, "LOAD MYSQL QUERY RULES TO RUNTIME"); + + conn = connect_proxy(SRC_DB); + if (!conn) { + goto cleanup; + } + val = fetch_single(conn, "SELECT DATABASE()"); + ok(val == SRC_DB, "after rule removal: DATABASE() should be '%s', got '%s'", SRC_DB, val.c_str()); + val = fetch_single(conn, "SELECT v FROM marker"); + ok(val == "in_src", "after rule removal: marker should be 'in_src', got '%s'", val.c_str()); + mysql_close(conn); + conn = NULL; + +cleanup: + if (conn) { + mysql_close(conn); + } + if (setup) { + snprintf(query, sizeof(query), "DROP DATABASE IF EXISTS %s", SRC_DB); + mysql_query(setup, query); + snprintf(query, sizeof(query), "DROP DATABASE IF EXISTS %s", DST_DB); + mysql_query(setup, query); + mysql_close(setup); + } + snprintf(query, sizeof(query), "DELETE FROM mysql_query_rules WHERE rule_id=%d", RULE_ID); + mysql_query(admin, query); + mysql_query(admin, "LOAD MYSQL QUERY RULES TO RUNTIME"); + mysql_close(admin); + + return exit_status(); +} From a686ee6832e78da280c6ff1a1b44721a4c804107 Mon Sep 17 00:00:00 2001 From: peterlyoo Date: Fri, 10 Jul 2026 11:10:53 +0900 Subject: [PATCH 3/4] Address review feedback on destination_schema (#4880) - Apply the schema switch before the query cache lookup so cache keys use the remapped schema and a cache hit no longer bypasses the switch - Warn when destination_schema is an empty string in rule attributes - Serialize destination_schema in Query_Processor_Output::get_info_json - Set attributes/comment to NULL after free in destroy() for consistency - Extend the TAP test with a query-cache interaction case (12 assertions) --- include/query_processor.h | 2 ++ lib/MySQL_Query_Processor.cpp | 4 +++ lib/MySQL_Session.cpp | 14 ++++---- lib/Query_Processor.cpp | 1 + .../tap/tests/mysql-dest_schema_routing-t.cpp | 33 ++++++++++++++++++- 5 files changed, 47 insertions(+), 7 deletions(-) diff --git a/include/query_processor.h b/include/query_processor.h index 9b2b2837bb..5e14d780bd 100644 --- a/include/query_processor.h +++ b/include/query_processor.h @@ -231,6 +231,7 @@ class Query_Processor_Output { } if (attributes) { free(attributes); + attributes=NULL; } if (destination_schema) { free(destination_schema); @@ -238,6 +239,7 @@ class Query_Processor_Output { } if (comment) { // #643 free(comment); + comment=NULL; } } void get_info_json(nlohmann::json& j); diff --git a/lib/MySQL_Query_Processor.cpp b/lib/MySQL_Query_Processor.cpp index 62377539a0..350ff6d54f 100644 --- a/lib/MySQL_Query_Processor.cpp +++ b/lib/MySQL_Query_Processor.cpp @@ -828,6 +828,8 @@ MySQL_Query_Processor_Rule_t* MySQL_Query_Processor::new_query_rule(int rule_id, std::string s = dest_schema; if (s.length() > 0) { newQR->destination_schema = strdup(s.c_str()); + } else { + proxy_warning("destination_schema is empty in attributes for rule_id %d , ignoring it\n", newQR->rule_id); } } else { proxy_error("Failed to parse destination_schema in JSON on attributes for rule_id %d : %s\n", newQR->rule_id, dest_schema.dump().c_str()); @@ -976,6 +978,8 @@ MySQL_Query_Processor_Rule_t* MySQL_Query_Processor::new_query_rule(const MySQL_ std::string s = dest_schema; if (s.length() > 0) { newQR->destination_schema = strdup(s.c_str()); + } else { + proxy_warning("destination_schema is empty in attributes for rule_id %d , ignoring it\n", newQR->rule_id); } } else { proxy_error("Failed to parse destination_schema in JSON on attributes for rule_id %d : %s\n", newQR->rule_id, dest_schema.dump().c_str()); diff --git a/lib/MySQL_Session.cpp b/lib/MySQL_Session.cpp index 4102c07e14..19b19eb5ab 100644 --- a/lib/MySQL_Session.cpp +++ b/lib/MySQL_Session.cpp @@ -7114,6 +7114,14 @@ bool MySQL_Session::handler___status_WAITING_CLIENT_DATA___STATE_SLEEP___MYSQL_C return true; } + if (qpo->destination_schema) { + // switch the session schema before the query cache lookup and backend + // connection selection: the cache key and the connection pool both use + // (username, schemaname), and the pool issues COM_INIT_DB on schema + // mismatch, so the query (cached or not) lands on this schema + client_myds->myconn->userinfo->set_schemaname(qpo->destination_schema, strlen(qpo->destination_schema)); + } + if (prepare_stmt_type & ps_type_execute_stmt) { // for prepared statement execute we exit here reset_warning_hostgroup_flag_and_release_connection(); goto __exit_set_destination_hostgroup; @@ -8226,12 +8234,6 @@ bool MySQL_Session::handler___status_WAITING_CLIENT_DATA___STATE_SLEEP___MYSQL_C current_hostgroup=qpo->destination_hostgroup; } } - if (qpo->destination_schema) { - // switch the session schema before backend connection selection: the - // connection pool matches on (username, schemaname) and issues - // COM_INIT_DB on schema mismatch, so the query lands on this schema - client_myds->myconn->userinfo->set_schemaname(qpo->destination_schema, strlen(qpo->destination_schema)); - } if (mysql_thread___set_query_lock_on_hostgroup == 1) { // algorithm introduced in 2.0.6 if (locked_on_hostgroup >= 0) { diff --git a/lib/Query_Processor.cpp b/lib/Query_Processor.cpp index a174135bc9..707a3b897d 100644 --- a/lib/Query_Processor.cpp +++ b/lib/Query_Processor.cpp @@ -2949,6 +2949,7 @@ void Query_Processor_Output::get_info_json(json& j) { j["cache_ttl"] = cache_ttl; j["delay"] = delay; j["destination_hostgroup"] = destination_hostgroup; + j["destination_schema"] = ( destination_schema ? destination_schema : "" ); j["firewall_whitelist_mode"] = firewall_whitelist_mode; j["multiplex"] = multiplex; j["timeout"] = timeout; diff --git a/test/tap/tests/mysql-dest_schema_routing-t.cpp b/test/tap/tests/mysql-dest_schema_routing-t.cpp index 231087d2aa..4496051b82 100644 --- a/test/tap/tests/mysql-dest_schema_routing-t.cpp +++ b/test/tap/tests/mysql-dest_schema_routing-t.cpp @@ -64,7 +64,7 @@ MYSQL* connect_proxy(const char* db) { } int main() { - plan(8); + plan(12); if (cl.getEnv()) return exit_status(); @@ -152,6 +152,37 @@ int main() { mysql_close(conn); conn = NULL; + // query cache interaction: the schema switch must happen before the cache + // lookup, so cache keys use the remapped schema and a cache HIT still + // leaves the session on the remapped schema + snprintf(query, sizeof(query), + "UPDATE mysql_query_rules SET cache_ttl=60000 WHERE rule_id=%d", RULE_ID); + MYSQL_QUERY_ON_ERR_CLEANUP(admin, query); + MYSQL_QUERY_ON_ERR_CLEANUP(admin, "LOAD MYSQL QUERY RULES TO RUNTIME"); + + conn = connect_proxy(SRC_DB); + if (!conn) { + goto cleanup; + } + val = fetch_single(conn, "SELECT v FROM marker"); + ok(val == "in_dst", "cache path: first (cache-miss) marker should be 'in_dst', got '%s'", val.c_str()); + { + std::string hits_before = fetch_single(admin, + "SELECT variable_value FROM stats_mysql_global WHERE variable_name='Query_Cache_count_GET_OK'"); + val = fetch_single(conn, "SELECT v FROM marker"); + ok(val == "in_dst", "cache path: second (cache-hit) marker should be 'in_dst', got '%s'", val.c_str()); + std::string hits_after = fetch_single(admin, + "SELECT variable_value FROM stats_mysql_global WHERE variable_name='Query_Cache_count_GET_OK'"); + ok(atoll(hits_after.c_str()) > atoll(hits_before.c_str()), + "cache path: Query_Cache_count_GET_OK should increase (before=%s, after=%s)", + hits_before.c_str(), hits_after.c_str()); + } + // even after a cache hit the session must stay on the remapped schema + val = fetch_single(conn, "SELECT DATABASE()"); + ok(val == DST_DB, "cache path: DATABASE() after cache hit should be '%s', got '%s'", DST_DB, val.c_str()); + mysql_close(conn); + conn = NULL; + // remove the rule: behavior must revert snprintf(query, sizeof(query), "DELETE FROM mysql_query_rules WHERE rule_id=%d", RULE_ID); MYSQL_QUERY_ON_ERR_CLEANUP(admin, query); From 67b6e086c7af62e705d6629674ff52bc385343a4 Mon Sep 17 00:00:00 2001 From: peterlyoo Date: Tue, 4 Aug 2026 08:25:44 +0900 Subject: [PATCH 4/4] Extract destination_schema attribute parsing into a helper (#4880) SonarCloud cpp:S134 flagged >3 levels of nesting at both new_query_rule() parsing sites, where the block was also duplicated verbatim. Behavior is unchanged; mysql-dest_schema_routing-t still passes 12/12. --- lib/MySQL_Query_Processor.cpp | 54 ++++++++++++++++++----------------- 1 file changed, 28 insertions(+), 26 deletions(-) diff --git a/lib/MySQL_Query_Processor.cpp b/lib/MySQL_Query_Processor.cpp index 350ff6d54f..3f4ff53011 100644 --- a/lib/MySQL_Query_Processor.cpp +++ b/lib/MySQL_Query_Processor.cpp @@ -701,6 +701,32 @@ MySQL_Query_Processor_Output* MySQL_Query_Processor::process_query(MySQL_Session return ret; } +/** + * @brief Parses the optional "destination_schema" key out of a rule's attributes JSON. + * @details Shared by both new_query_rule() overloads. On any problem the rule is left + * with a NULL destination_schema (no schema switch) and the reason is logged: a + * non-string value is a configuration error, an empty string is a no-op worth warning + * about. Kept as a free function so the attributes parsing in the callers stays flat. + * @param newQR Rule being built; its destination_schema is set on success. + * @param j_attributes Already parsed attributes JSON of the rule. + */ +static void parse_rule_destination_schema(MySQL_Query_Processor_Rule_t* newQR, const nlohmann::json& j_attributes) { + const auto it = j_attributes.find("destination_schema"); + if (it == j_attributes.end()) { + return; + } + if (it->type() != nlohmann::json::value_t::string) { + proxy_error("Failed to parse destination_schema in JSON on attributes for rule_id %d : %s\n", newQR->rule_id, it->dump().c_str()); + return; + } + const std::string s = *it; + if (s.empty()) { + proxy_warning("destination_schema is empty in attributes for rule_id %d , ignoring it\n", newQR->rule_id); + return; + } + newQR->destination_schema = strdup(s.c_str()); +} + MySQL_Query_Processor_Rule_t* MySQL_Query_Processor::new_query_rule(int rule_id, bool active, const char* username, const char* schemaname, int flagIN, const char* client_addr, const char* proxy_addr, int proxy_port, const char* digest, const char* match_digest, const char* match_pattern, bool negate_match_pattern, const char* re_modifiers, int flagOUT, const char* replace_pattern, int destination_hostgroup, int cache_ttl, int cache_empty_result, @@ -822,19 +848,7 @@ MySQL_Query_Processor_Rule_t* MySQL_Query_Processor::new_query_rule(int rule_id, proxy_error("Failed to parse flagOUTs attributes for rule_id %d : %s\n", newQR->rule_id, flagOUTs.dump().c_str()); } } - if (j_attributes.find("destination_schema") != j_attributes.end()) { - const nlohmann::json& dest_schema = j_attributes["destination_schema"]; - if (dest_schema.type() == nlohmann::json::value_t::string) { - std::string s = dest_schema; - if (s.length() > 0) { - newQR->destination_schema = strdup(s.c_str()); - } else { - proxy_warning("destination_schema is empty in attributes for rule_id %d , ignoring it\n", newQR->rule_id); - } - } else { - proxy_error("Failed to parse destination_schema in JSON on attributes for rule_id %d : %s\n", newQR->rule_id, dest_schema.dump().c_str()); - } - } + parse_rule_destination_schema(newQR, j_attributes); } } proxy_debug(PROXY_DEBUG_MYSQL_QUERY_PROCESSOR, 5, "Creating new rule in %p : rule_id:%d, active:%d, username=%s, schemaname=%s, flagIN:%d, %smatch_digest=\"%s\", %smatch_pattern=\"%s\", flagOUT:%d replace_pattern=\"%s\", destination_hostgroup:%d, apply:%d\n", newQR, newQR->rule_id, newQR->active, newQR->username, newQR->schemaname, newQR->flagIN, (newQR->negate_match_pattern ? "(!)" : ""), newQR->match_digest, (newQR->negate_match_pattern ? "(!)" : ""), newQR->match_pattern, newQR->flagOUT, newQR->replace_pattern, newQR->destination_hostgroup, newQR->apply); @@ -972,19 +986,7 @@ MySQL_Query_Processor_Rule_t* MySQL_Query_Processor::new_query_rule(const MySQL_ proxy_error("Failed to parse flagOUTs attributes for rule_id %d : %s\n", newQR->rule_id, flagOUTs.dump().c_str()); } } - if (j_attributes.find("destination_schema") != j_attributes.end()) { - const nlohmann::json& dest_schema = j_attributes["destination_schema"]; - if (dest_schema.type() == nlohmann::json::value_t::string) { - std::string s = dest_schema; - if (s.length() > 0) { - newQR->destination_schema = strdup(s.c_str()); - } else { - proxy_warning("destination_schema is empty in attributes for rule_id %d , ignoring it\n", newQR->rule_id); - } - } else { - proxy_error("Failed to parse destination_schema in JSON on attributes for rule_id %d : %s\n", newQR->rule_id, dest_schema.dump().c_str()); - } - } + parse_rule_destination_schema(newQR, j_attributes); } } proxy_debug(PROXY_DEBUG_MYSQL_QUERY_PROCESSOR, 5, "Creating new rule in %p : rule_id:%d, active:%d, username=%s, schemaname=%s, flagIN:%d, %smatch_digest=\"%s\", %smatch_pattern=\"%s\", flagOUT:%d replace_pattern=\"%s\", destination_hostgroup:%d, apply:%d\n", newQR, newQR->rule_id, newQR->active, newQR->username, newQR->schemaname, newQR->flagIN, (newQR->negate_match_pattern ? "(!)" : ""), newQR->match_digest, (newQR->negate_match_pattern ? "(!)" : ""), newQR->match_pattern, newQR->flagOUT, newQR->replace_pattern, newQR->destination_hostgroup, newQR->apply);