MDEV-19157 Optimistic Parallel Replication fails when gtid_slave_pos table is non transnational and binlog events are transnationalbb-10.2-sachin

Problem:- In Optimistic/Aggressive mode of parallel replication when we have
    conflict, we retry the transaction, But if gtid_slave_pos is non transactional
    In that case retry may result into duplicate key error for table because
    the conflicted transaction can not rollback the changes in gtid_slave_pos table
    So when we try the same trans again, We get the Duplicate Key error

Solution:- Users are advised to use transactional storage engine for gtid_slave_pos.
    In this patch we are throwing warning when we see that gtid_slave_pos in non
    transactional. In addition to throwing warnings Transactional event groups
    from master are not scheduled in Optimistic/Aggressive mode.

10 files changed, 217 insertions, 1 deletions

diff --git a/mysql-test/suite/rpl/include/rpl_mdev_19157_failing.inc b/mysql-test/suite/rpl/include/rpl_mdev_19157_failing.inc
new file mode 100644
index 00000000000..3db4faebef0
--- /dev/null
+++ b/mysql-test/suite/rpl/include/rpl_mdev_19157_failing.inc
@@ -0,0 +1,27 @@
+#
+# This is failing test case from the MDEV-19157
+#  We will make some inserts/updates in the master which are conflicting
+#  And gtid_slave_pos is non transactional
+#
+
+# Create some conflicting  events on master
+connection master;
+CREATE TABLE t1(a INT NOT NULL AUTO_INCREMENT PRIMARY KEY) engine=innodb;
+INSERT INTO t1 VALUES (1),(2),(3),(4);
+--disable_query_log
+--let $count=5
+while ($count)
+{
+update t1 set a=a+5 where a=1;
+update t1 set a=a+5 where a=2;
+update t1 set a=a+5 where a=3;
+update t1 set a=a+5 where a=4;
+update t1 set a=a-5 where a=5;
+update t1 set a=a-5 where a=6;
+--dec $count
+}
+--enable_query_log
+select * from t1 order by a;
+drop table t1;
+--sync_slave_with_master
+connection master;
diff --git a/mysql-test/suite/rpl/r/mdev_19157.result b/mysql-test/suite/rpl/r/mdev_19157.result
new file mode 100644
index 00000000000..8bb09ac727c
--- /dev/null
+++ b/mysql-test/suite/rpl/r/mdev_19157.result
@@ -0,0 +1,70 @@
+include/master-slave.inc
+[connection master]
+call mtr.add_suppression('Non transactional gtid_slave_pos table with *');
+connection slave;
+include/stop_slave.inc
+SET GLOBAL slave_parallel_threads=8;
+set global slave_parallel_mode=optimistic;
+alter table mysql.gtid_slave_pos engine=myisam;
+connection slave;
+include/start_slave.inc
+Warnings:
+Warning	4079	Non transactional gtid_slave_pos table with Optimistic parallel mode detected, Transactions will be applied in conservative mode.
+#Only 1 warning
+FOUND 1 /Non transactional gtid_slave_pos table with/ in mysqld.2.err
+include/stop_slave.inc
+include/start_slave.inc
+Warnings:
+Warning	4079	Non transactional gtid_slave_pos table with Optimistic parallel mode detected, Transactions will be applied in conservative mode.
+#2 warning
+FOUND 2 /Non transactional gtid_slave_pos table with/ in mysqld.2.err
+#Restart , There should be one more warning
+include/rpl_restart_server.inc [server_number=2]
+SET GLOBAL slave_parallel_threads=8;
+set global slave_parallel_mode=optimistic;
+alter table mysql.gtid_slave_pos engine=myisam;
+include/start_slave.inc
+Warnings:
+Warning	4079	Non transactional gtid_slave_pos table with Optimistic parallel mode detected, Transactions will be applied in conservative mode.
+#3 warning
+FOUND 3 /Non transactional gtid_slave_pos table with/ in mysqld.2.err
+##load the failing test on Optimistic mode
+connection master;
+connection master;
+CREATE TABLE t1(a INT NOT NULL AUTO_INCREMENT PRIMARY KEY) engine=innodb;
+INSERT INTO t1 VALUES (1),(2),(3),(4);
+select * from t1 order by a;
+a
+1
+7
+8
+9
+drop table t1;
+connection slave;
+connection master;
+##load the failing test on Aggressive mode
+connection slave;
+include/stop_slave.inc
+set global slave_parallel_mode=aggressive;
+include/start_slave.inc
+Warnings:
+Warning	4079	Non transactional gtid_slave_pos table with Aggressive parallel mode detected, Transactions will be applied in conservative mode.
+connection master;
+connection master;
+CREATE TABLE t1(a INT NOT NULL AUTO_INCREMENT PRIMARY KEY) engine=innodb;
+INSERT INTO t1 VALUES (1),(2),(3),(4);
+select * from t1 order by a;
+a
+1
+7
+8
+9
+drop table t1;
+connection slave;
+connection master;
+connection slave;
+include/stop_slave.inc
+SET GLOBAL slave_parallel_threads= 0;
+set global slave_parallel_mode=conservative;
+include/start_slave.inc
+include/rpl_end.inc
diff --git a/mysql-test/suite/rpl/t/mdev_19157.test b/mysql-test/suite/rpl/t/mdev_19157.test
new file mode 100644
index 00000000000..906d32cf397
--- /dev/null
+++ b/mysql-test/suite/rpl/t/mdev_19157.test
@@ -0,0 +1,60 @@
+#
+# MDEV-19157 -Optimistic Parallel Replication fails when gtid_slave_pos table is non
+#  transactional and binlog events are transactional
+# Test if gtid_slave_pos is non transactional we are giving warning to user
+#
+--source include/have_binlog_format_row.inc
+--source include/have_innodb.inc
+--source include/master-slave.inc
+
+call mtr.add_suppression('Non transactional gtid_slave_pos table with *');
+--sync_slave_with_master
+--source include/stop_slave.inc
+--let $old_threads= `select @@GLOBAL.slave_parallel_threads`
+--let $old_mode= `select @@GLOBAL.slave_parallel_mode`
+SET GLOBAL slave_parallel_threads=8;
+set global slave_parallel_mode=optimistic;
+alter table mysql.gtid_slave_pos engine=myisam;
+
+
+--connection slave
+--source include/start_slave.inc
+--let SEARCH_FILE= $MYSQLTEST_VARDIR/log/mysqld.2.err
+--let SEARCH_PATTERN= Non transactional gtid_slave_pos table with
+--echo #Only 1 warning
+source include/search_pattern_in_file.inc;
+
+--source include/stop_slave.inc
+--source include/start_slave.inc
+--echo #2 warning
+source include/search_pattern_in_file.inc;
+
+--echo #Restart , There should be one more warning
+--let $rpl_server_number= 2
+--source include/rpl_restart_server.inc
+SET GLOBAL slave_parallel_threads=8;
+set global slave_parallel_mode=optimistic;
+alter table mysql.gtid_slave_pos engine=myisam;
+--source include/start_slave.inc
+--echo #3 warning
+source include/search_pattern_in_file.inc;
+
+--echo ##load the failing test on Optimistic mode
+--connection master
+--source include/rpl_mdev_19157_failing.inc
+
+--echo ##load the failing test on Aggressive mode
+--connection slave
+--source include/stop_slave.inc
+set global slave_parallel_mode=aggressive;
+--source include/start_slave.inc
+--connection master
+--source include/rpl_mdev_19157_failing.inc
+
+--connection slave
+--source include/stop_slave.inc
+--eval SET GLOBAL slave_parallel_threads= $old_threads
+--eval set global slave_parallel_mode=$old_mode
+
+--source include/start_slave.inc
+--source include/rpl_end.inc
diff --git a/sql/rpl_gtid.cc b/sql/rpl_gtid.cc
index a8f7641fce4..88a95914c24 100644
--- a/sql/rpl_gtid.cc
+++ b/sql/rpl_gtid.cc
@@ -244,7 +244,7 @@ rpl_slave_state_free_element(void *arg)
 
 
 rpl_slave_state::rpl_slave_state()
-  : last_sub_id(0), loaded(false)
+  : last_sub_id(0), loaded(false), gtid_slave_pos_transactional_cache(-1)
 {
   mysql_mutex_init(key_LOCK_slave_state, &LOCK_slave_state,
                    MY_MUTEX_INIT_SLOW);
diff --git a/sql/rpl_gtid.h b/sql/rpl_gtid.h
index e4949ff0d39..9dd4771fafd 100644
--- a/sql/rpl_gtid.h
+++ b/sql/rpl_gtid.h
@@ -172,6 +172,13 @@ struct rpl_slave_state
 
   uint64 last_sub_id;
   bool loaded;
+  /*
+   -1 refresh required(maybe mysql_gtid_slave_pos_transactional is called first
+                       time , or gtid_slave_pos engine changed)
+    0 gtid_slave_pos not transactional
+    1       "         transactional
+  */
+  int gtid_slave_pos_transactional_cache;
 
   rpl_slave_state();
   ~rpl_slave_state();
diff --git a/sql/rpl_parallel.cc b/sql/rpl_parallel.cc
index 4503f6ed9be..1dcf705c0fa 100644
--- a/sql/rpl_parallel.cc
+++ b/sql/rpl_parallel.cc
@@ -2815,6 +2815,7 @@ rpl_parallel::do_event(rpl_group_info *serial_rgi, Log_event *ev,
         new_gco= false;
         if (!(gtid_flags & Gtid_log_event::FL_TRANSACTIONAL) ||
             ( (!(gtid_flags & Gtid_log_event::FL_ALLOW_PARALLEL) ||
+           (rpl_global_gtid_slave_state->gtid_slave_pos_transactional_cache==0)||
                (gtid_flags & Gtid_log_event::FL_WAITED)) &&
               (mode < SLAVE_PARALLEL_AGGRESSIVE)))
         {
diff --git a/sql/rpl_rli.cc b/sql/rpl_rli.cc
index 5273b33c728..2b4426719c3 100644
--- a/sql/rpl_rli.cc
+++ b/sql/rpl_rli.cc
@@ -1568,6 +1568,32 @@ Relay_log_info::update_relay_log_state(rpl_gtid *gtid_list, uint32 count)
 
 
 #if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
+/*
+  Check whether gtid_slave_pos is transactional or not, and load the result
+  into rpl_global_gtid_slave_state->is_gtid_slave_pos_transactional 
+  @@returns
+      0      Success
+      >0      Error while opening gtid_slave_pos table
+*/
+int mysql_gtid_slave_pos_transactional(THD *thd)
+{
+  if(rpl_global_gtid_slave_state->gtid_slave_pos_transactional_cache != -1)
+    return 0;
+  TABLE_LIST tlist;
+  int error;
+  tlist.init_one_table(STRING_WITH_LEN("mysql"),
+                       rpl_gtid_slave_state_table_name.str,
+                       rpl_gtid_slave_state_table_name.length,
+                       NULL, TL_READ);
+  if ((error= open_and_lock_tables(thd, &tlist, FALSE, 0)))
+    return error;
+  rpl_global_gtid_slave_state->gtid_slave_pos_transactional_cache=
+                    !(tlist.table->file->ha_table_flags() & HA_NO_TRANSACTIONS);
+  close_thread_tables(thd);
+  thd->release_transactional_locks();
+  return 0;
+}
+
 int
 rpl_load_gtid_slave_state(THD *thd)
 {
diff --git a/sql/rpl_rli.h b/sql/rpl_rli.h
index 2bc0a80268a..671c4fa8903 100644
--- a/sql/rpl_rli.h
+++ b/sql/rpl_rli.h
@@ -992,6 +992,7 @@ public:
 extern struct rpl_slave_state *rpl_global_gtid_slave_state;
 extern gtid_waiting rpl_global_gtid_waiting;
 
+int mysql_gtid_slave_pos_transactional(THD *thd);
 int rpl_load_gtid_slave_state(THD *thd);
 int event_group_new_gtid(rpl_group_info *rgi, Gtid_log_event *gev);
 void delete_or_keep_event_post_apply(rpl_group_info *rgi,
diff --git a/sql/share/errmsg-utf8.txt b/sql/share/errmsg-utf8.txt
index 2d4e0562d8f..52328436a1c 100644
--- a/sql/share/errmsg-utf8.txt
+++ b/sql/share/errmsg-utf8.txt
@@ -7744,3 +7744,5 @@ ER_MYROCKS_CANT_NOPAD_COLLATION
   eng "MyRocks doesn't currently support collations with \"No pad\" attribute."
 ER_TRANSACTIONAL_ARIA_LOG_ENGINE
         eng "Only non-transactional Aria table can be used for logging"
+ER_GTID_SLAVE_POS_NON_TRANS
+        eng "Non transactional gtid_slave_pos table with %s parallel mode detected, Transactions will be applied in conservative mode."
diff --git a/sql/sql_repl.cc b/sql/sql_repl.cc
index 7ff0e27b008..42531d8d6e6 100644
--- a/sql/sql_repl.cc
+++ b/sql/sql_repl.cc
@@ -15,6 +15,7 @@
    Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1335  USA */
 
 #include <my_global.h>
+#include "mysqld.h"
 #include "sql_priv.h"
 #include "unireg.h"
 #include "sql_base.h"
@@ -3163,6 +3164,27 @@ int start_slave(THD* thd , Master_info* mi,  bool net_report)
                      Sql_condition::WARN_LEVEL_NOTE, ER_UNTIL_COND_IGNORED,
                      ER_THD(thd, ER_UNTIL_COND_IGNORED));
 
+      int result= mysql_gtid_slave_pos_transactional(thd);
+      if(result)
+      {
+        slave_errno= result;
+        goto err;
+      }
+      if(!(rpl_global_gtid_slave_state->gtid_slave_pos_transactional_cache)&&
+              mi->parallel_mode > SLAVE_PARALLEL_CONSERVATIVE &&
+              opt_slave_parallel_threads > 1)
+      {
+        const char *mode= mi->parallel_mode == SLAVE_PARALLEL_OPTIMISTIC ?
+                     "Optimistic" : "Aggressive";
+        sql_print_warning("Non transactional gtid_slave_pos table with %s "
+                          "parallel mode detected, Transactions will be applied "
+                          "in conservative mode.", mode);
+        push_warning_printf(thd, Sql_condition::WARN_LEVEL_WARN,
+                                ER_GTID_SLAVE_POS_NON_TRANS,
+                                ER_THD(thd, ER_GTID_SLAVE_POS_NON_TRANS),
+                                mode);
+      }
+
       if (!slave_errno)
         slave_errno = start_slave_threads(thd,
                                           1,