Bug#36674083 session-variables aren't propagated if read-write splitting is enabled

If read-write splitting is used and the sequence:

  mysql> select @@sql_mode;
  ONLY_FULL_GROUP_BY,...

  mysql> set sql_mode="ansi";

  mysql> select @@sql_mode;
  ONLY_FULL_GROUP_BY,...

then the 2nd SELECT returns the initial sql_mode, not the updated one.

Change
======

After taking a read-connection from the stash, apply all system-variable
changes done on the write-connection (and vice versa).

  mysql> select @@sql_mode;
  ONLY_FULL_GROUP_BY,...

  mysql> set sql_mode="ansi";

  mysql> select @@sql_mode;
  REAL_AS_FLOAT,PIPES_AS_CONCAT,ANSI_QUOTES,IGNORE_SPACE,ONLY_FULL_GROUP_BY,ANSI

Change-Id: I57c0f2d258a4037f75115a7719e1e889272f3b6e

Author: weigon

router/src/routing/include/mysqlrouter/classic_protocol_state.h

@@ -27,7 +27,9 @@
 #define ROUTING_CLASSIC_PROTOCOL_STATE_INCLUDED
 
 #include <chrono>
+#include <map>
 #include <optional>
+#include <string>
 
 #include "classic_prepared_statement.h"
 
@@ -43,6 +45,98 @@ class ClassicProtocolState {
     kFinished,
   };
 
+  /**
+   * system-variables as returned by the server.
+   *
+   * can be queried from the server with:
+   *
+   * - SELECT @@SESSION.{k}
+   * - SELECT @@LOCAL.{k}
+   *
+   * can be set on the server with:
+   *
+   * - SET k = v;
+   * - SET @@SESSION.k = v;
+   * - SET @@LOCAL.k = v;
+   * - SET SESSION k = v;
+   * - SET LOCAL k = v;
+   *
+   * changes to system-vars on the server are returned via
+   * the sesssion-tracker for system-variables.
+   */
+  class SystemVariables {
+   public:
+    using key_type = std::string;
+    using key_view_type = std::string_view;
+    using value_type = std::optional<std::string>;
+
+    /**
+     * set k to v.
+     *
+     * if k doesn't exist in the system-vars yet, it gets inserted.
+     */
+    void set(key_type k, value_type v) {
+      vars_.insert_or_assign(std::move(k), std::move(v));
+    }
+
+    /**
+     * find 'k' in sytem-vars.
+     *
+     * @param k key
+     *
+     * if 'k' does not exist in system-vars, a NULL-like value is returned.
+     * otherwise return the value for the system-var referenced by 'k'
+     *
+     * @returns std::nullopt if key is not found, the found value otherwise.
+     */
+    std::optional<value_type> find(const key_view_type &k) const {
+      const auto it = vars_.find(k);
+      if (it == vars_.end()) return {std::nullopt};
+
+      return it->second;
+    }
+
+    /**
+     * get 'k' from system-vars.
+     *
+     * @param k key
+     *
+     * if 'k' does not exist in system-vars, a NULL-like value is returned.
+     * otherwise return the value for the system-var referenced by 'k' which may
+     * be NULL-like or a string.
+     *
+     * @returns std::nullopt if key is not found or value is NULL-like, the
+     * found value otherwise
+     */
+    value_type get(const key_view_type &k) const {
+      const auto it = vars_.find(k);
+      if (it == vars_.end()) return {std::nullopt};
+
+      return it->second;
+    }
+
+    using iterator = std::map<key_type, value_type>::iterator;
+    using const_iterator = std::map<key_type, value_type>::const_iterator;
+
+    iterator begin() { return vars_.begin(); }
+    const_iterator begin() const { return vars_.begin(); }
+    iterator end() { return vars_.end(); }
+    const_iterator end() const { return vars_.end(); }
+
+    /**
+     * check if there is no system-var.
+     */
+    bool empty() const { return vars_.empty(); }
+
+    /**
+     * clear the system-vars.
+     */
+    void clear() { vars_.clear(); }
+
+   private:
+    std::map<key_type, value_type, std::less<>> vars_;
+  };
+
   ClassicProtocolState() = default;
 
   ClassicProtocolState(
@@ -169,6 +263,10 @@ class ClassicProtocolState {
   }
 #endif
 
+  SystemVariables &system_variables() { return system_variables_; }
+
+  const SystemVariables &system_variables() const { return system_variables_; }
+
  private:
   classic_protocol::capabilities::value_type server_caps_{};
   classic_protocol::capabilities::value_type client_caps_{};
@@ -193,6 +291,8 @@ class ClassicProtocolState {
   classic_protocol::status::value_type status_flags_{};
 
   HandshakeState handshake_state_{HandshakeState::kConnected};
+
+  SystemVariables system_variables_;
 };
 
 class ClientSideClassicProtocolState : public ClassicProtocolState {

router/src/routing/src/classic_change_user_forwarder.cc

@@ -331,8 +331,8 @@ stdx::expected<Processor::Result, std::error_code> ChangeUserForwarder::ok() {
     // if connection sharing is enabled in the config, enable the
     // session-tracker.
     connection()->push_processor(std::make_unique<QuerySender>(connection(), R"(
-SET @@SESSION.session_track_schema           = 'ON',
-    @@SESSION.session_track_system_variables = '*',
+SET @@SESSION.session_track_system_variables = '*',
+    @@SESSION.session_track_schema           = 'ON',
     @@SESSION.session_track_transaction_info = 'CHARACTERISTICS',
     @@SESSION.session_track_gtids            = 'OWN_GTID',
     @@SESSION.session_track_state_change     = 'ON')"));

router/src/routing/src/classic_change_user_sender.cc

@@ -286,6 +286,8 @@ stdx::expected<Processor::Result, std::error_code> ChangeUserSender::command() {
 
   dst_protocol.seq_id(0xff);  // reset seq-id
 
+  dst_protocol.system_variables().clear();
+
   auto send_res = ClassicFrame::send_msg(dst_conn, *change_user_msg_);
   if (!send_res) return send_server_failed(send_res.error());
 

router/src/routing/src/classic_command.cc

@@ -282,7 +282,9 @@ class SelectSessionCollationConnectionHandler : public QuerySender::Handler {
       connection_->some_state_changed(true);
     } else {
       // all rows received,
-      connection_->execution_context().system_variables().set(
+      connection_->client_protocol().system_variables().set(
+          "collation_connection", collation_connection_);
+      connection_->server_protocol().system_variables().set(
           "collation_connection", collation_connection_);
 
       connection_->collation_connection_maybe_dirty(false);
@@ -307,7 +309,7 @@ class SelectSessionCollationConnectionHandler : public QuerySender::Handler {
 
   bool something_failed_{false};
 
-  Value collation_connection_{std::nullopt};
+  std::optional<std::string> collation_connection_{std::nullopt};
 };
 
 stdx::expected<Processor::Result, std::error_code>

router/src/routing/src/classic_connection_base.cc

@@ -532,7 +532,7 @@ MysqlRoutingClassicConnectionBase::track_session_changes(
             set_names_sysvar.set(3);
           }
 
-          auto value_from_kv = [](auto kv) -> Value {
+          auto value_from_kv = [](auto kv) -> std::optional<std::string> {
             // the session tracker can't report NULL. Instead it reports "".
             //
             // In the case of 'character_set_results' setting "" leads to an
@@ -545,8 +545,10 @@ MysqlRoutingClassicConnectionBase::track_session_changes(
             return {std::string(kv.value())};
           };
 
-          exec_ctx_.system_variables().set(std::string(kv.key()),
-                                           value_from_kv(kv));
+          client_protocol().system_variables().set(std::string(kv.key()),
+                                                   value_from_kv(kv));
+          server_protocol().system_variables().set(std::string(kv.key()),
+                                                   value_from_kv(kv));
 
           if (auto &tr = tracer()) {
             std::ostringstream oss;
@@ -836,7 +838,7 @@ void MysqlRoutingClassicConnectionBase::loop() {
 }
 
 bool MysqlRoutingClassicConnectionBase::connection_sharing_possible() const {
-  const auto &sysvars = exec_ctx_.system_variables();
+  const auto &sysvars = client_protocol().system_variables();
 
   return context_.connection_sharing() &&             // config must allow it.
          client_protocol().password().has_value() &&  // a password is required
@@ -955,18 +957,18 @@ bool MysqlRoutingClassicConnectionBase::connection_sharing_allowed() const {
 
 std::string MysqlRoutingClassicConnectionBase::connection_sharing_blocked_by()
     const {
-  const auto &sysvars = exec_ctx_.system_variables();
+  const auto &sysvars = client_protocol().system_variables();
 
   // "possible"
   if (!context_.connection_sharing()) return "config";
   if (!client_protocol().password().has_value()) return "no-password";
-  if (sysvars.get("session_track_gtids") != Value("OWN_GTID"))
+  if (sysvars.get("session_track_gtids") != "OWN_GTID")
     return "session-track-gtids";
-  if (sysvars.get("session_track_state_change") != Value("ON"))
+  if (sysvars.get("session_track_state_change") != "ON")
     return "session-track-state-change";
-  if (sysvars.get("session_track_system_variables") != Value("*"))
+  if (sysvars.get("session_track_system_variables") != "*")
     return "session-track-system-variables";
-  if (sysvars.get("session_track_transaction_info") != Value("CHARACTERISTICS"))
+  if (sysvars.get("session_track_transaction_info") != "CHARACTERISTICS")
     return "session-track-transaction-info";
 
   // "allowed"
@@ -1001,7 +1003,8 @@ void MysqlRoutingClassicConnectionBase::reset_to_initial() {
   execution_context().diagnostics_area().warnings().clear();
 
   // clear the tracked-system-vars like sql_mode, ...
-  execution_context().system_variables().clear();
+  src_protocol.system_variables().clear();
+  server_protocol().system_variables().clear();
 
   // clear the prepared statements.
   src_protocol.prepared_statements().clear();