Implementation of Multi-source replication (MDEV:253)

Documentation of the feature can be found at: http://kb.askmonty.org/en/multi-source-replication/
This code is based on code from Taobao, developed by Plinux

BUILD/SETUP.sh:
  Added -Wno-invalid-offsetof to get rid of warning of offsetof() on C++ class (safe in the contex we use it)
client/mysqltest.cc:
  Added support for error names starting with 'W'
  Added connection_name support to --sync_with_master
cmake/maintainer.cmake:
  Added -Wno-invalid-offsetof to get rid of warning of offsetof() on C++ class (safe in the contex we use it)
mysql-test/r/mysqltest.result:
  Updated results
mysql-test/r/parser.result:
  Updated results
mysql-test/suite/multi_source/my.cnf:
  Setup of multi-master tests
mysql-test/suite/multi_source/simple.result:
  Simple basic test of multi-source functionality
mysql-test/suite/multi_source/simple.test:
  Simple basic test of multi-source functionality
mysql-test/suite/multi_source/syntax.result:
  Test of multi-source syntax
mysql-test/suite/multi_source/syntax.test:
  Test of multi-source syntax
mysql-test/suite/rpl/r/rpl_rotate_logs.result:
  Updated results because of new error messages
mysql-test/t/parser.test:
  Updated test as master_pos_wait() now takes more arguments than before
sql/event_scheduler.cc:
  No reason to initialize slave_thread (it's guaranteed to be zero here)
sql/item_create.cc:
  Added connection_name argument to master_pos_wait()
  Simplified code
sql/item_func.cc:
  Added connection_name argument to master_pos_wait()
sql/item_func.h:
  Added connection_name argument to master_pos_wait()
sql/log.cc:
  Added tag "Master 'connection_name'" to slave errors that has a connection name.
sql/mysqld.cc:
  Added variable mysqld_server_initialized so that other functions can test if server is fully initialized.
  Free all slave data in one place (fewer ifdef's)
  Removed not needed call to close_active_mi()
  Initialize slaves() later in startup to ensure that everthing is really initialized when slaves start.
  Made status variable slave_running multi-source safe
sql/mysqld.h:
  Added mysqld_server_initialized
sql/rpl_mi.cc:
  Store connection name and cmp_connection_name (only used for show full slave status) in Master_info
  Added code for Master_info_index, which handles storage of multi-master information
  Don't write the empty "" connection_name to multi-master.info file. This is handled by the original code.
sql/rpl_mi.h:
  Added connection_name and Master_info_index
sql/rpl_rli.cc:
  Added connection_name to relay log files.
sql/rpl_rli.h:
  Fixed type of slave_skip_counter as we now access it directly in sys_vars.cc, so it must be uint
sql/share/errmsg-utf8.txt:
  Added new error messages needed for multi-source
  Added multi-source name to error ER_MASTER_INFO and WARN_NO_MASTER_INFO
sql/slave.cc:
  Moved things a bit around to make it easier to handle error conditions.
  Create a global master_info_index and add the "" connection to it
  Ensure that new Master_info doesn't fail.
  Don't call terminate_slave_threads(active_mi..) on end_slave() as this is now done automaticly when deleting master_info_index.
  Delete not needed function close_active_mi(). One can achive same thing by calling end_slave().
  Added support for SHOW FULL SLAVE STATUS (show status for all master connections with connection_name as first column)
sql/slave.h:
  Added new prototypes
sql/sql_base.cc:
  More DBUG_PRINT
sql/sql_class.cc:
  Reset thd->connection_name and thd-->default_master_connection
sql/sql_class.h:
  Added thd->connection_name and thd-->default_master_connection
  Added slave_skip_count to variables to make changing the @@sql_slave_skip_count variable thread safe
sql/sql_const.h:
  Added MAX_CONNECTION_NAME
sql/sql_lex.cc:
  Reset 'lex->verbose' (to simplify some sql_yacc.yy code)
sql/sql_lex.h:
  Added connection_name
sql/sql_parse.cc:
  Added support for connection_name to all SLAVE commands.
  - Instead of using active_mi, we now get the current Master_info from master_info_index.
  - Create new replication threads with CHANGE MASTER
  - Added support for show_all_master_info()
sql/sql_reload.cc:
  Made reset/full slave use master_info_index->get_master_info() instead of active_mi.
  If one uses 'RESET SLAVE "connection_name" all' the connection is removed from master_info_index.
sql/sql_repl.cc:
  sql_slave_skip_counter is moved to thd->variables to make it thread safe and fix some bugs with it
  Add connection name to relay log files.
  Added connection name to errors.
  Added some logging for multi-master if log_warnings > 1
  stop_slave():
  - Don't check if thd is set. It's guaranteed to always be set.
  change_master():
  - Check for duplicate connection names in change_master()
  - Check for wrong arguments first in file (to simplify error handling)
  - Register new connections in master_info_index
sql/sql_yacc.yy:
  Added optional connection_name to a all relevant master/slave commands
sql/strfunc.cc:
  my_global.h shoud always be included first.
sql/sys_vars.cc:
  Added variable default_master_connection
  Made variable sql_slave_skip_counter multi-source safe
sql/sys_vars.h:
  Added Sys_var_session_lexstring (needed for default_master_connection)
  Added Sys_var_multi_source_uint (needed for sql_slave_skip_counter).

1 files changed, 31 insertions, 11 deletions

diff --git a/client/mysqltest.cc b/client/mysqltest.cc
index bb89db4b231..f824d53ed1b 100644
--- a/client/mysqltest.cc
+++ b/client/mysqltest.cc
@@ -253,6 +253,8 @@ static void init_re(void);
 static int match_re(my_regex_t *, char *);
 static void free_re(void);
 
+static char *get_string(char **to_ptr, char **from_ptr,
+                        struct st_command *command);
 static int replace(DYNAMIC_STRING *ds_str,
                    const char *search_str, ulong search_len,
                    const char *replace_str, ulong replace_len);
@@ -4580,7 +4582,8 @@ void do_wait_for_slave_to_stop(struct st_command *c __attribute__((unused)))
 }
 
 
-void do_sync_with_master2(struct st_command *command, long offset)
+void do_sync_with_master2(struct st_command *command, long offset,
+                          const char *connection_name)
 {
   MYSQL_RES *res;
   MYSQL_ROW row;
@@ -4591,8 +4594,9 @@ void do_sync_with_master2(struct st_command *command, long offset)
   if (!master_pos.file[0])
     die("Calling 'sync_with_master' without calling 'save_master_pos'");
 
-  sprintf(query_buf, "select master_pos_wait('%s', %ld, %d)",
-          master_pos.file, master_pos.pos + offset, timeout);
+  sprintf(query_buf, "select master_pos_wait('%s', %ld, %d, '%s')",
+          master_pos.file, master_pos.pos + offset, timeout,
+          connection_name);
 
   if (mysql_query(mysql, query_buf))
     die("failed in '%s': %d: %s", query_buf, mysql_errno(mysql),
@@ -4652,16 +4656,32 @@ void do_sync_with_master(struct st_command *command)
   long offset= 0;
   char *p= command->first_argument;
   const char *offset_start= p;
+  char *start, *buff= 0;
+  start= (char*) "";
+
   if (*offset_start)
   {
     for (; my_isdigit(charset_info, *p); p++)
       offset = offset * 10 + *p - '0';
 
-    if(*p && !my_isspace(charset_info, *p))
+    if (*p && !my_isspace(charset_info, *p) && *p != ',')
       die("Invalid integer argument \"%s\"", offset_start);
+
+    while (*p && my_isspace(charset_info, *p))
+      p++;
+    if (*p == ',')
+    {
+      p++;
+      while (*p && my_isspace(charset_info, *p))
+        p++;
+      start= buff= (char*)my_malloc(strlen(p)+1,MYF(MY_WME | MY_FAE));
+      get_string(&buff, &p, command);
+    }
     command->last_argument= p;
   }
-  do_sync_with_master2(command, offset);
+  do_sync_with_master2(command, offset, start);
+  if (buff)
+    my_free(start);
   return;
 }
 
@@ -5286,7 +5306,7 @@ void do_get_errcodes(struct st_command *command)
     {
       die("The sqlstate definition must start with an uppercase S");
     }
-    else if (*p == 'E')
+    else if (*p == 'E' || *p == 'W')
     {
       /* Error name string */
 
@@ -5295,9 +5315,9 @@ void do_get_errcodes(struct st_command *command)
       to->type= ERR_ERRNO;
       DBUG_PRINT("info", ("ERR_ERRNO: %d", to->code.errnum));
     }
-    else if (*p == 'e')
+    else if (*p == 'e' || *p == 'w')
     {
-      die("The error name definition must start with an uppercase E");
+      die("The error name definition must start with an uppercase E or W");
     }
     else
     {
@@ -5360,8 +5380,8 @@ void do_get_errcodes(struct st_command *command)
   If string is a '$variable', return the value of the variable.
 */
 
-char *get_string(char **to_ptr, char **from_ptr,
-                 struct st_command *command)
+static char *get_string(char **to_ptr, char **from_ptr,
+                        struct st_command *command)
 {
   char c, sep;
   char *to= *to_ptr, *from= *from_ptr, *start=to;
@@ -9288,7 +9308,7 @@ int main(int argc, char **argv)
 	  select_connection(command);
 	else
 	  select_connection_name("slave");
-	do_sync_with_master2(command, 0);
+	do_sync_with_master2(command, 0, "");
 	break;
       }
       case Q_COMMENT: