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, 78 insertions, 29 deletions

diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index 175c2c1d672..114035bc16d 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -82,6 +82,7 @@
 #include <myisam.h>
 #include <my_dir.h>
 #include "rpl_handler.h"
+#include "rpl_mi.h"
 
 #include "sp_head.h"
 #include "sp.h"
@@ -2343,10 +2344,41 @@ case SQLCOM_PREPARE:
 #ifdef HAVE_REPLICATION
   case SQLCOM_CHANGE_MASTER:
   {
+    LEX_MASTER_INFO *lex_mi= &thd->lex->mi;
+    Master_info *mi;
+    bool new_master= 0;
+
     if (check_global_access(thd, SUPER_ACL))
       goto error;
     mysql_mutex_lock(&LOCK_active_mi);
-    res = change_master(thd,active_mi);
+
+    mi= master_info_index->get_master_info(&lex_mi->connection_name,
+                                           MYSQL_ERROR::WARN_LEVEL_NOTE);
+
+    if (mi == NULL)
+    {
+      /* New replication created */
+      mi= new Master_info(&lex_mi->connection_name, relay_log_recovery); 
+      if (!mi || mi->error())
+      {
+        delete mi;
+        res= 1;
+        mysql_mutex_unlock(&LOCK_active_mi);
+        break;
+      }
+      new_master= 1;
+    }
+
+    res= change_master(thd, mi);
+    if (res && new_master)
+    {
+      /*
+        The new master was added by change_master(). Remove it as it didn't
+        work.
+      */
+      master_info_index->remove_master_info(&lex_mi->connection_name);
+    }
+
     mysql_mutex_unlock(&LOCK_active_mi);
     break;
   }
@@ -2356,15 +2388,19 @@ case SQLCOM_PREPARE:
     if (check_global_access(thd, SUPER_ACL | REPL_CLIENT_ACL))
       goto error;
     mysql_mutex_lock(&LOCK_active_mi);
-    if (active_mi != NULL)
-    {
-      res = show_master_info(thd, active_mi);
-    }
+
+    if (lex->verbose)
+      res= show_all_master_info(thd);
     else
     {
-      push_warning(thd, MYSQL_ERROR::WARN_LEVEL_WARN,
-                   WARN_NO_MASTER_INFO, ER(WARN_NO_MASTER_INFO));
-      my_ok(thd);
+      LEX_MASTER_INFO *lex_mi= &thd->lex->mi;
+      Master_info *mi;
+      mi= master_info_index->get_master_info(&lex_mi->connection_name,
+                                             MYSQL_ERROR::WARN_LEVEL_ERROR);
+      if (mi != NULL)
+      {
+        res= show_master_info(thd, mi, 0);
+      }
     }
     mysql_mutex_unlock(&LOCK_active_mi);
     break;
@@ -2664,40 +2700,53 @@ end_with_restore_list:
 #ifdef HAVE_REPLICATION
   case SQLCOM_SLAVE_START:
   {
+    LEX_MASTER_INFO* lex_mi= &thd->lex->mi;
+    Master_info *mi;
     mysql_mutex_lock(&LOCK_active_mi);
-    start_slave(thd,active_mi,1 /* net report*/);
+
+    if ((mi= (master_info_index->
+              get_master_info(&lex_mi->connection_name,
+                              MYSQL_ERROR::WARN_LEVEL_ERROR))))
+      start_slave(thd, mi, 1 /* net report*/);
     mysql_mutex_unlock(&LOCK_active_mi);
     break;
   }
   case SQLCOM_SLAVE_STOP:
-  /*
-    If the client thread has locked tables, a deadlock is possible.
-    Assume that
-    - the client thread does LOCK TABLE t READ.
-    - then the master updates t.
-    - then the SQL slave thread wants to update t,
-      so it waits for the client thread because t is locked by it.
+  {
+    LEX_MASTER_INFO *lex_mi;
+    Master_info *mi;
+    /*
+      If the client thread has locked tables, a deadlock is possible.
+      Assume that
+      - the client thread does LOCK TABLE t READ.
+      - then the master updates t.
+      - then the SQL slave thread wants to update t,
+        so it waits for the client thread because t is locked by it.
     - then the client thread does SLAVE STOP.
       SLAVE STOP waits for the SQL slave thread to terminate its
       update t, which waits for the client thread because t is locked by it.
-    To prevent that, refuse SLAVE STOP if the
-    client thread has locked tables
-  */
-  if (thd->locked_tables_mode ||
-      thd->in_active_multi_stmt_transaction() || thd->global_read_lock.is_acquired())
-  {
-    my_message(ER_LOCK_OR_ACTIVE_TRANSACTION,
-               ER(ER_LOCK_OR_ACTIVE_TRANSACTION), MYF(0));
-    goto error;
-  }
-  {
+      To prevent that, refuse SLAVE STOP if the
+      client thread has locked tables
+    */
+    if (thd->locked_tables_mode ||
+        thd->in_active_multi_stmt_transaction() ||
+        thd->global_read_lock.is_acquired())
+    {
+      my_message(ER_LOCK_OR_ACTIVE_TRANSACTION,
+                 ER(ER_LOCK_OR_ACTIVE_TRANSACTION), MYF(0));
+      goto error;
+    }
+
+    lex_mi= &thd->lex->mi;
     mysql_mutex_lock(&LOCK_active_mi);
-    stop_slave(thd,active_mi,1/* net report*/);
+    if ((mi= (master_info_index->
+              get_master_info(&lex_mi->connection_name,
+                              MYSQL_ERROR::WARN_LEVEL_ERROR))))
+      stop_slave(thd, mi, 1/* net report*/);
     mysql_mutex_unlock(&LOCK_active_mi);
     break;
   }
 #endif /* HAVE_REPLICATION */
-
   case SQLCOM_RENAME_TABLE:
   {
     if (execute_rename_table(thd, first_table, all_tables))