MDEV-6594: Use separate domain_id for Galera transactions

6 files changed, 98 insertions, 14 deletions

diff --git a/sql/log.cc b/sql/log.cc
index 0eb475433bd..f94fe12a6f3 100644
--- a/sql/log.cc
+++ b/sql/log.cc
@@ -5616,13 +5616,26 @@ MYSQL_BIN_LOG::write_gtid_event(THD *thd, bool standalone,
                                 bool is_transactional, uint64 commit_id)
 {
   rpl_gtid gtid;
-  uint32 domain_id= thd->variables.gtid_domain_id;
-  uint32 server_id= thd->variables.server_id;
-  uint64 seq_no= thd->variables.gtid_seq_no;
+  uint32 domain_id;
+  uint32 server_id;
+  uint64 seq_no;
   int err;
   DBUG_ENTER("write_gtid_event");
   DBUG_PRINT("enter", ("standalone: %d", standalone));
-  
+
+#ifdef WITH_WSREP
+  if (WSREP(thd) && thd->wsrep_trx_meta.gtid.seqno != -1 && wsrep_gtid_mode)
+  {
+    domain_id= wsrep_gtid_domain_id;
+  } else {
+#endif /* WITH_WSREP */
+  domain_id= thd->variables.gtid_domain_id;
+#ifdef WITH_WSREP
+  }
+#endif /* WITH_WSREP */
+  server_id= thd->variables.server_id;
+  seq_no= thd->variables.gtid_seq_no;
+
   if (thd->variables.option_bits & OPTION_GTID_BEGIN)
   {
     DBUG_PRINT("error", ("OPTION_GTID_BEGIN is set. "
diff --git a/sql/sys_vars.cc b/sql/sys_vars.cc
index 0dea5a134e9..8371df0f0a6 100644
--- a/sql/sys_vars.cc
+++ b/sql/sys_vars.cc
@@ -4859,6 +4859,22 @@ static Sys_var_mybool Sys_wsrep_dirty_reads(
        "is not ready.", SESSION_ONLY(wsrep_dirty_reads), NO_CMD_LINE,
        DEFAULT(FALSE), NO_MUTEX_GUARD, NOT_IN_BINLOG);
 
+static Sys_var_uint Sys_wsrep_gtid_domain_id(
+       "wsrep_gtid_domain_id", "When wsrep_gtid_mode is set, this value is "
+       "used as gtid_domain_id for galera transactions and also copied to the "
+       "joiner nodes during state transfer. It is ignored, otherwise.",
+       GLOBAL_VAR(wsrep_gtid_domain_id), CMD_LINE(REQUIRED_ARG),
+       VALID_RANGE(0, UINT_MAX32), DEFAULT(0), BLOCK_SIZE(1), NO_MUTEX_GUARD,
+       NOT_IN_BINLOG);
+
+static Sys_var_mybool Sys_wsrep_gtid_mode(
+       "wsrep_gtid_mode", "Automatically update the (joiner) node's "
+       "wsrep_gtid_domain_id value with that of donor's (received during "
+       "state transfer) and use it in place of gtid_domain_id for all galera "
+       "transactions. When OFF (default), wsrep_gtid_domain_id is simply "
+       "ignored (backward compatibility).",
+       GLOBAL_VAR(wsrep_gtid_mode), CMD_LINE(OPT_ARG), DEFAULT(FALSE));
+
 #endif /* WITH_WSREP */
 
 static bool fix_host_cache_size(sys_var *, THD *, enum_var_type)
diff --git a/sql/wsrep_mysqld.cc b/sql/wsrep_mysqld.cc
index 19736cef7bf..a441f86df40 100644
--- a/sql/wsrep_mysqld.cc
+++ b/sql/wsrep_mysqld.cc
@@ -89,6 +89,12 @@ my_bool wsrep_restart_slave_activated  = 0; // node has dropped, and slave
 my_bool wsrep_slave_UK_checks          = 0; // slave thread does UK checks
 my_bool wsrep_slave_FK_checks          = 0; // slave thread does FK checks
 bool wsrep_new_cluster                 = false; // Bootstrap the cluster ?
+
+// Use wsrep_gtid_domain_id for galera transactions?
+bool wsrep_gtid_mode                   = 0;
+// gtid_domain_id for galera transactions.
+uint32 wsrep_gtid_domain_id            = 0;
+
 /*
  * End configuration options
  */
@@ -1711,6 +1717,12 @@ pthread_handler_t start_wsrep_THD(void *arg)
   mysql_mutex_lock(&LOCK_thread_count);
   thd->thread_id=thread_id++;
 
+  if (wsrep_gtid_mode)
+  {
+    /* Adjust domain_id. */
+    thd->variables.gtid_domain_id= wsrep_gtid_domain_id;
+  }
+
   thd->real_id=pthread_self(); // Keep purify happy
   thread_count++;
   thread_created++;
diff --git a/sql/wsrep_mysqld.h b/sql/wsrep_mysqld.h
index 085dc5de453..ce40cec5835 100644
--- a/sql/wsrep_mysqld.h
+++ b/sql/wsrep_mysqld.h
@@ -88,6 +88,8 @@ extern my_bool     wsrep_slave_FK_checks;
 extern my_bool     wsrep_slave_UK_checks;
 extern ulong       wsrep_running_threads;
 extern bool        wsrep_new_cluster;
+extern bool        wsrep_gtid_mode;
+extern uint32      wsrep_gtid_domain_id;
 
 enum enum_wsrep_OSU_method { WSREP_OSU_TOI, WSREP_OSU_RSU };
 enum enum_wsrep_sync_wait {
diff --git a/sql/wsrep_sst.cc b/sql/wsrep_sst.cc
index d170ff91bbc..512e8f23dd8 100644
--- a/sql/wsrep_sst.cc
+++ b/sql/wsrep_sst.cc
@@ -435,17 +435,53 @@ static void* sst_joiner_thread (void* a)
 
     if (!tmp)
     {
-      WSREP_ERROR("Failed to read uuid:seqno from joiner script.");
+      WSREP_ERROR("Failed to read uuid:seqno and wsrep_gtid_domain_id from "
+                  "joiner script.");
       if (proc.error()) err = proc.error();
     }
     else
     {
-      err= sst_scan_uuid_seqno (out, &ret_uuid, &ret_seqno);
+      // Read state ID (UUID:SEQNO) followed by wsrep_gtid_domain_id (if any).
+      const char *pos= strchr(out, ' ');
+
+      if (!pos) {
+        // There is no wsrep_gtid_domain_id (some older version SST script?).
+        err= sst_scan_uuid_seqno (out, &ret_uuid, &ret_seqno);
+
+      } else {
+        // Scan state ID first followed by wsrep_gtid_domain_id.
+        char uuid[512];
+        uint32 domain_id;
+        size_t len= pos - out + 1;
+
+        if (len > sizeof(uuid)) goto err;       // safety check
+        memcpy(uuid, out, len);                 // including '\0'
+        err= sst_scan_uuid_seqno (uuid, &ret_uuid, &ret_seqno);
+
+        if (err)
+        {
+          goto err;
+        }
+        else if (wsrep_gtid_mode)
+        {
+          domain_id= strtol(pos + 1, NULL, 10);
+          if (domain_id < 1000 || domain_id > 0xFFFF)
+          {
+            WSREP_ERROR("Failed to get donor wsrep_gtid_domain_id.");
+            err= EINVAL;
+            goto err;
+          } else {
+            wsrep_gtid_domain_id= domain_id;
+          }
+        }
+      }
     }
 
+err:
+
     if (err)
     {
-      ret_uuid=  WSREP_UUID_UNDEFINED;
+      ret_uuid= WSREP_UUID_UNDEFINED;
       ret_seqno= -err;
     }
 
@@ -784,11 +820,12 @@ static int sst_donate_mysqldump (const char*         addr,
               WSREP_SST_OPT_LPORT" '%u' "
               WSREP_SST_OPT_SOCKET" '%s' "
               " %s "
-              WSREP_SST_OPT_GTID" '%s:%lld'"
+              WSREP_SST_OPT_GTID" '%s:%lld' "
+              WSREP_SST_OPT_GTID_DOMAIN_ID" '%d'"
               "%s",
               user, pswd, host, port, mysqld_port, mysqld_unix_port,
-              wsrep_defaults_file, uuid_str,
-              (long long)seqno, bypass ? " "WSREP_SST_OPT_BYPASS : "");
+              wsrep_defaults_file, uuid_str, (long long)seqno,
+              wsrep_gtid_domain_id, bypass ? " "WSREP_SST_OPT_BYPASS : "");
 
     WSREP_DEBUG("Running: '%s'", cmd_str);
 
@@ -885,8 +922,10 @@ static int sst_flush_tables(THD* thd)
     }
     else
     {
-      fprintf(file, "%s:%lld\n",
-              wsrep_cluster_state_uuid, (long long)wsrep_locked_seqno);
+      // Write cluster state ID and wsrep_gtid_domain_id.
+      fprintf(file, "%s:%lld %d\n",
+              wsrep_cluster_state_uuid, (long long)wsrep_locked_seqno,
+              wsrep_gtid_domain_id);
       fsync(fileno(file));
       fclose(file);
       if (rename(tmp_name, real_name) == -1)
@@ -1058,12 +1097,13 @@ static int sst_donate_other (const char*   method,
                  WSREP_SST_OPT_DATA" '%s' "
                  " %s "
                  " %s '%s' "
-                 WSREP_SST_OPT_GTID" '%s:%lld'"
+                 WSREP_SST_OPT_GTID" '%s:%lld' "
+                 WSREP_SST_OPT_GTID_DOMAIN_ID" '%d'"
                  "%s",
                  method, addr, sst_auth_real, mysqld_unix_port,
                  mysql_real_data_home, wsrep_defaults_file,
                  binlog_opt, binlog_opt_val,
-                 uuid, (long long) seqno,
+                 uuid, (long long) seqno, wsrep_gtid_domain_id,
                  bypass ? " "WSREP_SST_OPT_BYPASS : "");
   my_free(binlog_opt_val);
 
diff --git a/sql/wsrep_sst.h b/sql/wsrep_sst.h
index 2a6ab406297..49dd5b39fad 100644
--- a/sql/wsrep_sst.h
+++ b/sql/wsrep_sst.h
@@ -42,6 +42,7 @@
 #define WSREP_SST_OPT_SOCKET   "--socket"
 #define WSREP_SST_OPT_GTID     "--gtid"
 #define WSREP_SST_OPT_BYPASS   "--bypass"
+#define WSREP_SST_OPT_GTID_DOMAIN_ID "--gtid-domain-id"
 
 #define WSREP_SST_MYSQLDUMP    "mysqldump"
 #define WSREP_SST_RSYNC        "rsync"