Patch version 3

9 files changed, 195 insertions, 9 deletions

diff --git a/mysql-test/suite/rpl/t/r_ttt.test b/mysql-test/suite/rpl/t/r_ttt.test
new file mode 100644
index 00000000000..73c60463776
--- /dev/null
+++ b/mysql-test/suite/rpl/t/r_ttt.test
@@ -0,0 +1,119 @@
+--source include/have_binlog_format_row.inc
+--source include/have_innodb.inc
+--source include/master-slave.inc
+
+--connection slave
+set global debug_dbug="+d,start_alter_delay_slave";
+stop slave;
+SET GLOBAL slave_parallel_threads=10;
+set global slave_parallel_mode=optimistic;
+start slave;
+
+
+--connection master
+set global debug_dbug="+d,start_alter_delay_master";
+set global binlog_split_alter=true;
+--connection master
+create table t1( a int, b int) engine=myisam;
+insert into t1 values(1,1);
+insert into t1 values(2,2);
+--echo #Normal Alter
+alter table t1 add column c int;
+--echo #Failed Alter
+insert into t1 values(1,1, NULL);
+--error ER_DUP_ENTRY
+alter table t1 change a a int unique ;
+source include/show_binlog_events.inc;
+--sync_slave_with_master
+source include/show_binlog_events.inc;
+show create table t1;
+--connection master
+drop table t1;
+--sync_slave_with_master
+--connection master
+create table t1( a int primary key, b int) engine=innodb;
+insert into t1 values(1,1),(2,2);
+create table t2( a int primary key, b int) engine=innodb;
+insert into t2 values(1,1),(2,2);
+create table t3( a int primary key, b int) engine=innodb;
+insert into t3 values(1,1),(2,2);
+create table t4( a int primary key, b int) engine=innodb;
+insert into t4 values(1,1),(2,2);
+create table t5( a int primary key, b int) engine=innodb;
+insert into t5 values(1,1),(2,2);
+connect (con1,localhost,root,,);
+connect (con2,localhost,root,,);
+connect (con3,localhost,root,,);
+connect (con4,localhost,root,,);
+connect (con5,localhost,root,,);
+--connection con1
+--send alter table t1 add column c int, force, algorithm=inplace;
+--connection con2
+--send alter table t2 add column c int, force, algorithm=inplace;
+--connection con3
+--send alter table t3 add column c int, force, algorithm=inplace;
+--connection con4
+--send alter table t4 add column c int, force, algorithm=inplace;
+--connection con5
+--send alter table t5 add column c int, force, algorithm=inplace;
+
+--connection con1
+--reap
+--connection con2
+--reap
+--connection con3
+--reap
+--connection con4
+--reap
+--connection con5
+--reap
+--connection con1
+--send alter table t1 add column d int, force, algorithm=inplace;
+--connection con2
+--send alter table t2 add column d int, force, algorithm=inplace;
+--connection con3
+--send alter table t3 add column d int, force, algorithm=inplace;
+--connection con4
+--send alter table t4 add column d int, force, algorithm=inplace;
+--connection con5
+--send alter table t5 add column d int, force, algorithm=inplace;
+
+--connection con1
+--reap
+--connection con2
+--reap
+--connection con3
+--reap
+--connection con4
+--reap
+--connection con5
+--reap
+
+show binlog events;
+show create table t5;
+select @@gtid_slave_pos;
+select @@gtid_binlog_state;
+
+--sleep 20
+--connection slave
+select @@gtid_slave_pos;
+select @@gtid_binlog_state;
+select master_gtid_wait('0-1-10');
+show binlog events;
+
+--connection master
+--sync_slave_with_master
+#show create table t5;
+show binlog events;
+
+
+--connection master
+drop table t1,t2,t3,t4,t5;
+set global binlog_split_alter=true;
+
+--connection slave
+stop slave;
+SET GLOBAL slave_parallel_threads=0;
+set global slave_parallel_mode=conservative;
+start slave;
+--source include/rpl_end.inc
diff --git a/mysql-test/suite/rpl/t/r_ttt2.test b/mysql-test/suite/rpl/t/r_ttt2.test
new file mode 100644
index 00000000000..43d0dc6c67d
--- /dev/null
+++ b/mysql-test/suite/rpl/t/r_ttt2.test
@@ -0,0 +1,48 @@
+--source include/have_binlog_format_row.inc
+--source include/have_innodb.inc
+--source include/master-slave.inc
+
+--connection slave
+stop slave;
+SET GLOBAL slave_parallel_threads=10;
+set global slave_parallel_mode=optimistic;
+start slave;
+
+
+--connection master
+set global debug_dbug="+d,start_alter_delay_master";
+set global binlog_split_alter=true;
+create table t1( a int primary key, b int) engine=innodb;
+create table t2( a int primary key, b int) engine=innodb;
+connect (con1,localhost,root,,);
+connect (con2,localhost,root,,);
+--connection con1
+--send alter table t1 add column c int, force, algorithm=inplace;
+--connection con2
+--send alter table t2 add column c int, force, algorithm=inplace;
+
+#--source include/kill_mysqld.inc
+#--let $rpl_server_number= 1
+#--source include/rpl_stop_server.inc
+--connection con1
+--reap
+--connection con2
+--reap
+create table t3( a int primary key, b int) engine=innodb;
+
+show binlog events;
+
+--sync_slave_with_master
+show binlog events;
+show create table t1;
+
+--connection master
+drop table t1,t2,t3;
+set global binlog_split_alter= false;
+
+--connection slave
+stop slave;
+SET GLOBAL slave_parallel_threads=0;
+set global slave_parallel_mode=conservative;
+start slave;
+--source include/rpl_end.inc
diff --git a/sql/log_event_server.cc b/sql/log_event_server.cc
index fa1a0edeab3..de78b01ebd1 100644
--- a/sql/log_event_server.cc
+++ b/sql/log_event_server.cc
@@ -1604,6 +1604,7 @@ int Query_log_event::do_apply_event(rpl_group_info *rgi,
     you.
   */
   thd->catalog= catalog_len ? (char *) catalog : (char *)"";
+  thd->start_alter_ev= this;
 
   size_t valid_len= Well_formed_prefix(system_charset_info,
                                        db, db_len, NAME_LEN).length();
diff --git a/sql/rpl_gtid.cc b/sql/rpl_gtid.cc
index 599129b98f4..0d07236e578 100644
--- a/sql/rpl_gtid.cc
+++ b/sql/rpl_gtid.cc
@@ -620,7 +620,6 @@ rpl_slave_state::record_gtid(THD *thd, const rpl_gtid *gtid, uint64 sub_id,
     DBUG_RETURN(0);
   }
 
-  if (!in_statement)
     thd->reset_for_next_command();
 
   /*
diff --git a/sql/rpl_parallel.cc b/sql/rpl_parallel.cc
index 669a8455a3d..7e271dd57c4 100644
--- a/sql/rpl_parallel.cc
+++ b/sql/rpl_parallel.cc
@@ -1060,6 +1060,7 @@ handle_rpl_parallel_thread(void *arg)
   /* Ensure that slave can exeute any alter table it gets from master */
   thd->variables.alter_algorithm= (ulong) Alter_info::ALTER_TABLE_ALGORITHM_DEFAULT;
   thd->slave_thread= 1;
+  thd->rpt= rpt;
 
   set_slave_thread_options(thd);
   thd->client_capabilities = CLIENT_LOCAL_FILES;
@@ -1977,6 +1978,11 @@ rpl_parallel_thread::loc_free_gco(group_commit_orderer *gco)
     gco->next_gco= loc_gco_list;
   loc_gco_list= gco;
 }
+void rpl_parallel_thread::__finish_event_group(rpl_group_info *group_rgi)
+{
+  finish_event_group(this, group_rgi->gtid_sub_id,
+                                 group_rgi->parallel_entry, group_rgi);
+}
 
 rpl_parallel_thread_pool::rpl_parallel_thread_pool()
   : threads(0), free_list(0), count(0), inited(false), busy(false)
diff --git a/sql/rpl_parallel.h b/sql/rpl_parallel.h
index 6d95c303e3b..2592f22290c 100644
--- a/sql/rpl_parallel.h
+++ b/sql/rpl_parallel.h
@@ -224,6 +224,9 @@ struct rpl_parallel_thread {
   void batch_free();
   /* Update inuse_relaylog refcounts with what we have accumulated so far. */
   void inuse_relaylog_refcount_update();
+  /*
+   internal call to finish_event_group
+   */
   void __finish_event_group(rpl_group_info *);
 };
 
diff --git a/sql/sql_alter.cc b/sql/sql_alter.cc
index c7d18d7d242..b93ff30a585 100644
--- a/sql/sql_alter.cc
+++ b/sql/sql_alter.cc
@@ -696,10 +696,9 @@ bool Sql_cmd_alter_table::execute(THD *thd)
     We will follow a different executation path if it is START ALTER
     or commit/rollback alter
    */
+  if (0)
+      process_start_alter(thd, alter_info.alter_identifier);
   switch (alter_info.alter_state)  {
-  case Alter_info::ALTER_TABLE_START:
-    alter_res= process_start_alter(thd, alter_info.alter_identifier);
-    break;
   case Alter_info::ALTER_TABLE_COMMIT:
     alter_res= process_commit_alter(thd, alter_info.alter_identifier);
     break;
@@ -707,6 +706,7 @@ bool Sql_cmd_alter_table::execute(THD *thd)
     alter_res= process_rollback_alter(thd, alter_info.alter_identifier);
     break;
   case Alter_info::ALTER_TABLE_NORMAL:
+  default:
     alter_res= START_ALTER_PARSE;
     break;
   }
diff --git a/sql/sql_class.h b/sql/sql_class.h
index 4ccc4c3b1cc..1f637945e14 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -80,6 +80,7 @@ class Wsrep_applier_service;
 class Reprepare_observer;
 class Relay_log_info;
 struct rpl_group_info;
+struct rpl_parallel_thread;
 class Rpl_filter;
 class Query_log_event;
 class Load_log_event;
@@ -2224,6 +2225,8 @@ public:
   rpl_group_info* rgi_fake;
   /* Slave applier execution context */
   rpl_group_info* rgi_slave;
+  rpl_parallel_thread *rpt;
+  Query_log_event *start_alter_ev;
   bool slave_shutdown;
   bool direct_commit_alter;
   //Will be reseted when gtid is written into binlog
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 8b243693146..cd0ceb31688 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -9402,6 +9402,8 @@ static bool write_start_alter(THD *thd, bool* partial_alter, start_alter_info *i
   ulong start_alter_id= thd->lex->alter_info.alter_identifier;
   if (start_alter_id)
   {
+    if (write_bin_log(thd, true, thd->query(), thd->query_length()))
+      return 0;
     Master_info *mi= thd->rgi_slave->rli->mi;
     info->error= 0;
     info->thread_id= start_alter_id;
@@ -9413,6 +9415,11 @@ static bool write_start_alter(THD *thd, bool* partial_alter, start_alter_info *i
     DBUG_EXECUTE_IF("start_alter_delay_slave", {
       my_sleep(5000000);
       });
+    thd->start_alter_ev->update_pos(thd->rgi_slave);
+    thd->rgi_slave->mark_start_commit();
+    thd->wakeup_subsequent_commits(0);
+    //Finish event group
+    thd->rpt->__finish_event_group(thd->rgi_slave);
     if (thd->slave_shutdown)
       return true;
     return false;
@@ -9421,7 +9428,7 @@ static bool write_start_alter(THD *thd, bool* partial_alter, start_alter_info *i
   {
     char *send_query= (char *)thd->alloc(thd->query_length() + 20);
     thd->gtid_flags3|= Gtid_log_event::FL_START_ALTER_E1;
-    sprintf(send_query, "/*!105001  %s EXECUTE = UNTIL COMMIT %ld */",
+    sprintf(send_query, "/*!100001  %s EXECUTE = UNTIL COMMIT %ld */",
                                       thd->query(), (long)thd->thread_id);
     if (write_bin_log(thd, FALSE, send_query, strlen(send_query), true))
       return true;
@@ -10238,7 +10245,7 @@ do_continue:;
         if (opt_binlog_split_alter)
         {
           thd->gtid_flags3|= Gtid_log_event::FL_ROLLBACK_ALTER_E1;
-          sprintf(send_query, "/*!105001  %s EXECUTE = ROLLBACK %ld */", thd->query(),
+          sprintf(send_query, "/*!100001  %s EXECUTE = ROLLBACK %ld */", thd->query(),
                                                         (long)thd->thread_id);
           if(write_bin_log(thd, false, send_query, strlen(send_query), true, true))
             DBUG_RETURN(true);
@@ -10403,7 +10410,7 @@ do_continue:;
     if (partial_alter)
     {
       thd->gtid_flags3|= Gtid_log_event::FL_COMMIT_ALTER_E1;
-      sprintf(send_query, "%s /*!105001 EXECUTE = COMMIT %ld */", thd->query(),
+      sprintf(send_query, "%s /*!100001 EXECUTE = COMMIT %ld */", thd->query(),
                                                         (long)thd->thread_id);
       if(write_bin_log(thd, false, send_query, strlen(send_query)))
         DBUG_RETURN(true);
@@ -10612,7 +10619,7 @@ end_inplace:
   if (partial_alter)
   {
     thd->gtid_flags3|= Gtid_log_event::FL_COMMIT_ALTER_E1;
-    sprintf(send_query, "%s /*!105001 EXECUTE = COMMIT %ld */", thd->query(),
+    sprintf(send_query, "%s /*!100001 EXECUTE = COMMIT %ld */", thd->query(),
                                                         (long)thd->thread_id);
     if(write_bin_log(thd, false, send_query, strlen(send_query)))
       DBUG_RETURN(true);
@@ -10681,7 +10688,7 @@ err_new_table_cleanup:
   else if (opt_binlog_split_alter)
   {
     thd->gtid_flags3|= Gtid_log_event::FL_ROLLBACK_ALTER_E1;
-    sprintf(send_query, "/*!105001  %s EXECUTE = ROLLBACK %ld */", thd->query(),
+    sprintf(send_query, "/*!100001  %s EXECUTE = ROLLBACK %ld */", thd->query(),
                                                         (long)thd->thread_id);
     if(write_bin_log(thd, false, send_query, strlen(send_query), true, true))
       DBUG_RETURN(true);