Backport fix for MDEV-7673, MDEV-7203 and MDEV-7192 from 10.0-galera

mysql-test/suite/galera/r/create.result

@@ -20,4 +20,9 @@ i
 1
 DROP TABLE t1;
 SET @@GLOBAL.wsrep_forced_binlog_format=@wsrep_forced_binlog_format_saved;
+#
+# MDEV-7673: CREATE TABLE SELECT fails on Galera cluster
+#
+CREATE TABLE t1 (i INT) ENGINE=INNODB DEFAULT CHARSET=utf8 SELECT 1 as i;
+DROP TABLE t1;
 # End of tests

mysql-test/suite/galera/r/rpl_row_annotate.result

+# On node_2
+RESET MASTER;
+# On node_1
+RESET MASTER;
+CREATE TABLE t1(i INT)ENGINE=INNODB;
+INSERT INTO t1 VALUES(1);
+DELETE FROM t1 WHERE i = 1;
+# On node_2
+INSERT INTO t1 VALUES(2);
+DELETE FROM t1 WHERE i = 2;
+# On node_1
+SHOW BINLOG EVENTS IN 'mysqld-bin.000001' FROM <start_pos>;
+Log_name	Pos	Event_type	Server_id	End_log_pos	Info
+mysqld-bin.000001	#	Query	1	#	use `test`; CREATE TABLE t1(i INT)ENGINE=INNODB
+mysqld-bin.000001	#	Query	1	#	BEGIN
+mysqld-bin.000001	#	Annotate_rows	1	#	INSERT INTO t1 VALUES(1)
+mysqld-bin.000001	#	Table_map	1	#	table_id: # (test.t1)
+mysqld-bin.000001	#	Write_rows	1	#	table_id: # flags: STMT_END_F
+mysqld-bin.000001	#	Xid	1	#	COMMIT /* xid= */
+mysqld-bin.000001	#	Query	1	#	BEGIN
+mysqld-bin.000001	#	Annotate_rows	1	#	DELETE FROM t1 WHERE i = 1
+mysqld-bin.000001	#	Table_map	1	#	table_id: # (test.t1)
+mysqld-bin.000001	#	Delete_rows	1	#	table_id: # flags: STMT_END_F
+mysqld-bin.000001	#	Xid	1	#	COMMIT /* xid= */
+mysqld-bin.000001	#	Query	2	#	BEGIN
+mysqld-bin.000001	#	Annotate_rows	2	#	INSERT INTO t1 VALUES(2)
+mysqld-bin.000001	#	Table_map	2	#	table_id: # (test.t1)
+mysqld-bin.000001	#	Write_rows	2	#	table_id: # flags: STMT_END_F
+mysqld-bin.000001	#	Xid	2	#	COMMIT /* xid= */
+mysqld-bin.000001	#	Query	2	#	BEGIN
+mysqld-bin.000001	#	Annotate_rows	2	#	DELETE FROM t1 WHERE i = 2
+mysqld-bin.000001	#	Table_map	2	#	table_id: # (test.t1)
+mysqld-bin.000001	#	Delete_rows	2	#	table_id: # flags: STMT_END_F
+mysqld-bin.000001	#	Xid	2	#	COMMIT /* xid= */
+# On node_2
+SHOW BINLOG EVENTS IN 'mysqld-bin.000001' FROM <start_pos>;
+Log_name	Pos	Event_type	Server_id	End_log_pos	Info
+mysqld-bin.000001	#	Query	1	#	use `test`; CREATE TABLE t1(i INT)ENGINE=INNODB
+mysqld-bin.000001	#	Query	1	#	BEGIN
+mysqld-bin.000001	#	Annotate_rows	1	#	INSERT INTO t1 VALUES(1)
+mysqld-bin.000001	#	Table_map	1	#	table_id: # (test.t1)
+mysqld-bin.000001	#	Write_rows	1	#	table_id: # flags: STMT_END_F
+mysqld-bin.000001	#	Xid	1	#	COMMIT /* xid= */
+mysqld-bin.000001	#	Query	1	#	BEGIN
+mysqld-bin.000001	#	Annotate_rows	1	#	DELETE FROM t1 WHERE i = 1
+mysqld-bin.000001	#	Table_map	1	#	table_id: # (test.t1)
+mysqld-bin.000001	#	Delete_rows	1	#	table_id: # flags: STMT_END_F
+mysqld-bin.000001	#	Xid	1	#	COMMIT /* xid= */
+mysqld-bin.000001	#	Query	2	#	BEGIN
+mysqld-bin.000001	#	Annotate_rows	2	#	INSERT INTO t1 VALUES(2)
+mysqld-bin.000001	#	Table_map	2	#	table_id: # (test.t1)
+mysqld-bin.000001	#	Write_rows	2	#	table_id: # flags: STMT_END_F
+mysqld-bin.000001	#	Xid	2	#	COMMIT /* xid= */
+mysqld-bin.000001	#	Query	2	#	BEGIN
+mysqld-bin.000001	#	Annotate_rows	2	#	DELETE FROM t1 WHERE i = 2
+mysqld-bin.000001	#	Table_map	2	#	table_id: # (test.t1)
+mysqld-bin.000001	#	Delete_rows	2	#	table_id: # flags: STMT_END_F
+mysqld-bin.000001	#	Xid	2	#	COMMIT /* xid= */
+DROP TABLE t1;
+# End of test

mysql-test/suite/galera/t/create.test

@@ -22,5 +22,13 @@ SELECT * FROM t1_temp;
 DROP TABLE t1;
 SET @@GLOBAL.wsrep_forced_binlog_format=@wsrep_forced_binlog_format_saved;
 
+--echo #
+--echo # MDEV-7673: CREATE TABLE SELECT fails on Galera cluster
+--echo #
+CREATE TABLE t1 (i INT) ENGINE=INNODB DEFAULT CHARSET=utf8 SELECT 1 as i;
+
+# Cleanup
+DROP TABLE t1;
+
 --echo # End of tests
 

mysql-test/suite/galera/t/rpl_row_annotate.cnf

+!include ../galera_2nodes.cnf
+
+[mysqld]
+log-bin
+log-slave-updates
+binlog-annotate-row-events=ON

mysql-test/suite/galera/t/rpl_row_annotate.test

+--source include/galera_cluster.inc
+--source include/have_innodb.inc
+
+--echo # On node_2
+--connection node_2
+RESET MASTER;
+
+--echo # On node_1
+--connection node_1
+RESET MASTER;
+CREATE TABLE t1(i INT)ENGINE=INNODB;
+INSERT INTO t1 VALUES(1);
+DELETE FROM t1 WHERE i = 1;
+
+--echo # On node_2
+--connection node_2
+INSERT INTO t1 VALUES(2);
+DELETE FROM t1 WHERE i = 2;
+
+--echo # On node_1
+--connection node_1
+--source include/binlog_start_pos.inc
+let $start_pos= `select @binlog_start_pos`;
+--replace_column 2 # 5 #
+--replace_result $start_pos <start_pos>
+--replace_regex /table_id: [0-9]+/table_id: #/ /\/\* xid=.* \*\//\/* xid= *\//
+--eval SHOW BINLOG EVENTS IN 'mysqld-bin.000001' FROM $start_pos
+
+--echo # On node_2
+--connection node_2
+--source include/binlog_start_pos.inc
+let $start_pos= `select @binlog_start_pos`;
+--replace_column 2 # 5 #
+--replace_result $start_pos <start_pos>
+--replace_regex /table_id: [0-9]+/table_id: #/ /\/\* xid=.* \*\//\/* xid= *\//
+--eval SHOW BINLOG EVENTS IN 'mysqld-bin.000001' FROM $start_pos
+
+# Cleanup
+DROP TABLE t1;
+
+--source include/galera_end.inc
+--echo # End of test

sql/rpl_rli.cc

@@ -1218,6 +1218,47 @@ void Relay_log_info::stmt_done(my_off_t event_master_log_pos,
 }
 
 #if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
+
+void
+delete_or_keep_event_post_apply(Relay_log_info *rli,
+                                Log_event_type typ, Log_event *ev)
+{
+  switch (typ) {
+  case FORMAT_DESCRIPTION_EVENT:
+    /*
+      Format_description_log_event should not be deleted because it
+      will be used to read info about the relay log's format;
+      it will be deleted when the SQL thread does not need it,
+      i.e. when this thread terminates.
+    */
+    break;
+  case ANNOTATE_ROWS_EVENT:
+    /*
+      Annotate_rows event should not be deleted because after it has
+      been applied, thd->query points to the string inside this event.
+      The thd->query will be used to generate new Annotate_rows event
+      during applying the subsequent Rows events.
+    */
+    rli->set_annotate_event((Annotate_rows_log_event*) ev);
+    break;
+  case DELETE_ROWS_EVENT:
+  case UPDATE_ROWS_EVENT:
+  case WRITE_ROWS_EVENT:
+    /*
+      After the last Rows event has been applied, the saved Annotate_rows
+      event (if any) is not needed anymore and can be deleted.
+    */
+    if (((Rows_log_event*)ev)->get_flags(Rows_log_event::STMT_END_F))
+      rli->free_annotate_event();
+    /* fall through */
+  default:
+    DBUG_PRINT("info", ("Deleting the event after it has been executed"));
+    if (!rli->is_deferred_event(ev))
+      delete ev;
+    break;
+  }
+}
+
 void Relay_log_info::cleanup_context(THD *thd, bool error)
 {
   DBUG_ENTER("Relay_log_info::cleanup_context");

sql/rpl_rli.h

@@ -580,6 +580,7 @@ class Relay_log_info : public Slave_reporting_capability
 
 // Defined in rpl_rli.cc
 int init_relay_log_info(Relay_log_info* rli, const char* info_fname);
-
+void delete_or_keep_event_post_apply(Relay_log_info *rli,
+                                     Log_event_type typ, Log_event *ev);
 
 #endif /* RPL_RLI_H */

sql/slave.cc

@@ -2820,41 +2820,7 @@ static int exec_relay_log_event(THD* thd, Relay_log_info* rli)
 
     exec_res= apply_event_and_update_pos(ev, thd, rli);
 
-    switch (ev->get_type_code()) {
-      case FORMAT_DESCRIPTION_EVENT:
-        /*
-          Format_description_log_event should not be deleted because it
-          will be used to read info about the relay log's format;
-          it will be deleted when the SQL thread does not need it,
-          i.e. when this thread terminates.
-        */
-        break;
-      case ANNOTATE_ROWS_EVENT:
-        /*
-          Annotate_rows event should not be deleted because after it has
-          been applied, thd->query points to the string inside this event.
-          The thd->query will be used to generate new Annotate_rows event
-          during applying the subsequent Rows events.
-        */
-        rli->set_annotate_event((Annotate_rows_log_event*) ev);
-        break;
-      case DELETE_ROWS_EVENT:
-      case UPDATE_ROWS_EVENT:
-      case WRITE_ROWS_EVENT:
-        /*
-          After the last Rows event has been applied, the saved Annotate_rows
-          event (if any) is not needed anymore and can be deleted.
-        */
-        if (((Rows_log_event*)ev)->get_flags(Rows_log_event::STMT_END_F))
-          rli->free_annotate_event();
-        /* fall through */
-      default:
-        DBUG_PRINT("info", ("Deleting the event after it has been executed"));
-        if (!rli->is_deferred_event(ev))
-          delete ev;
-        break;
-    }
-
+    delete_or_keep_event_post_apply(rli, ev->get_type_code(), ev);
 
     /*
       update_log_pos failed: this should not happen, so we don't

sql/sql_parse.cc

@@ -2913,6 +2913,14 @@ case SQLCOM_PREPARE:
         if (create_info.options & HA_LEX_CREATE_TMP_TABLE)
           thd->variables.option_bits|= OPTION_KEEP_LOG;
 
+#ifdef WITH_WSREP
+        if (WSREP(thd) &&
+            (!thd->is_current_stmt_binlog_format_row() ||
+             !(create_info.options & HA_LEX_CREATE_TMP_TABLE)))
+          WSREP_TO_ISOLATION_BEGIN(create_table->db, create_table->table_name,
+                                   NULL)
+#endif
+
         /*
           select_create is currently not re-execution friendly and
           needs to be created for every execution of a PS/SP.

sql/wsrep_applier.cc

@@ -73,6 +73,7 @@ static wsrep_cb_status_t wsrep_apply_events(THD*        thd,
   char *buf= (char *)events_buf;
   int rcode= 0;
   int event= 1;
+  Log_event_type typ;
 
   DBUG_ENTER("wsrep_apply_events");
 
@@ -106,7 +107,10 @@ static wsrep_cb_status_t wsrep_apply_events(THD*        thd,
       rcode= 1;
       goto error;
     }
-    switch (ev->get_type_code()) {
+
+    typ= ev->get_type_code();
+
+    switch (typ) {
     case WRITE_ROWS_EVENT:
     case UPDATE_ROWS_EVENT:
     case DELETE_ROWS_EVENT:
@@ -117,14 +121,20 @@ static wsrep_cb_status_t wsrep_apply_events(THD*        thd,
       break;
     }
 
-    thd->server_id = ev->server_id; // use the original server id for logging
-    thd->set_time();                // time the query
+    /* Use the original server id for logging. */
+    thd->set_server_id(ev->server_id);
+    thd->set_time();                            // time the query
     wsrep_xid_init(&thd->transaction.xid_state.xid,
                    &thd->wsrep_trx_meta.gtid.uuid,
                    thd->wsrep_trx_meta.gtid.seqno);
     thd->lex->current_select= 0;
     if (!ev->when)
       ev->when = time(NULL);
+
+    thd->variables.option_bits=
+      (thd->variables.option_bits & ~OPTION_SKIP_REPLICATION) |
+      (ev->flags & LOG_EVENT_SKIP_REPLICATION_F ?  OPTION_SKIP_REPLICATION : 0);
+
     ev->thd = thd;
     exec_res = ev->apply_event(thd->wsrep_rli);
     DBUG_PRINT("info", ("exec_event result: %d", exec_res));
@@ -172,7 +182,8 @@ static wsrep_cb_status_t wsrep_apply_events(THD*        thd,
         WSREP_ERROR("Error in %s event: commit of row events failed: %lld",
                     ev->get_type_str(), (long long)wsrep_thd_trx_seqno(thd));
     }
-    delete ev;
+
+    delete_or_keep_event_post_apply(thd->wsrep_rli, typ, ev);
   }
 
  error:

sql/wsrep_thd.cc

@@ -94,6 +94,12 @@ static Relay_log_info* wsrep_relay_log_init(const char* log_fname)
   }
 
   rli->sql_thd= current_thd;
+
+  if ((rli->deferred_events_collecting= rpl_filter->is_on()))
+  {
+    rli->deferred_events= new Deferred_log_events(rli);
+  }
+
   return rli;
 }
 
@@ -132,6 +138,10 @@ static void wsrep_return_from_bf_mode(THD *thd, struct wsrep_thd_shadow* shadow)
   thd->net.vio                = shadow->vio;
   thd->variables.tx_isolation = shadow->tx_isolation;
   thd->reset_db(shadow->db, shadow->db_length);
+
+  thd->wsrep_rli->cleanup_after_session();
+  delete thd->wsrep_rli;
+  thd->wsrep_rli= NULL;
 }
 
 void wsrep_replay_transaction(THD *thd)