MDEV-22021: Galera database could get inconsistent with rollback to savepoint

When binlog is disabled, WSREP will not behave correctly when
SAVEPOINT ROLLBACK is executed since we don't register handlers for such case.
Fixed by registering WSREP handlerton for SAVEPOINT related commands.

mysql-test/suite/galera/r/MDEV-22021.result

+CREATE TABLE t1 (f1 INTEGER PRIMARY KEY);
+START TRANSACTION;
+INSERT INTO t1 VALUES (1);
+SAVEPOINT sp1;
+INSERT INTO t1 VALUES (2);
+ROLLBACK TO SAVEPOINT sp1;
+COMMIT;
+SELECT COUNT(*) = 1 FROM t1;
+COUNT(*) = 1
+1
+connection node_2;
+SELECT COUNT(*) = 1 FROM t1;
+COUNT(*) = 1
+1
+connection node_1;
+DELETE FROM t1;
+START TRANSACTION;
+SAVEPOINT sp1;
+INSERT INTO t1 VALUES (1);
+SAVEPOINT sp2;
+INSERT INTO t1 VALUES (2);
+ROLLBACK TO SAVEPOINT sp2;
+ROLLBACK TO SAVEPOINT sp1;
+COMMIT;
+SELECT COUNT(*) = 0 FROM t1;
+COUNT(*) = 0
+1
+connection node_2;
+SELECT COUNT(*) = 0 FROM t1;
+COUNT(*) = 0
+1
+connection node_1;
+DELETE FROM t1;
+START TRANSACTION;
+SAVEPOINT sp1;
+INSERT INTO t1 VALUES (1);
+INSERT INTO t1 VALUES (2);
+INSERT INTO t1 VALUES (3);
+INSERT INTO t1 VALUES (4);
+SAVEPOINT sp2;
+INSERT INTO t1 VALUES (5);
+ROLLBACK TO SAVEPOINT sp2;
+INSERT INTO t1 VALUES (6);
+INSERT INTO t1 VALUES (7);
+ROLLBACK TO SAVEPOINT sp1;
+INSERT INTO t1 VALUES (8);
+COMMIT;
+SELECT COUNT(*) = 1 FROM t1;
+COUNT(*) = 1
+1
+connection node_2;
+SELECT COUNT(*) = 1 FROM t1;
+COUNT(*) = 1
+1
+DROP TABLE t1;

mysql-test/suite/galera/t/MDEV-22021.combinations

+[binlogoff]
+
+[binlogon]
+log-bin

mysql-test/suite/galera/t/MDEV-22021.test

+#
+# SAVEPOINT ROLLBACK can introduce incosistency in cluster.
+#
+
+--source include/galera_cluster.inc
+
+CREATE TABLE t1 (f1 INTEGER PRIMARY KEY);
+
+START TRANSACTION;
+INSERT INTO t1 VALUES (1);
+SAVEPOINT sp1;
+INSERT INTO t1 VALUES (2);
+ROLLBACK TO SAVEPOINT sp1;
+COMMIT;
+
+SELECT COUNT(*) = 1 FROM t1;
+
+--connection node_2
+SELECT COUNT(*) = 1 FROM t1;
+
+--connection node_1
+DELETE FROM t1;
+
+START TRANSACTION;
+SAVEPOINT sp1;
+INSERT INTO t1 VALUES (1);
+SAVEPOINT sp2;
+INSERT INTO t1 VALUES (2);
+ROLLBACK TO SAVEPOINT sp2;
+ROLLBACK TO SAVEPOINT sp1;
+COMMIT;
+
+SELECT COUNT(*) = 0 FROM t1;
+--connection node_2
+SELECT COUNT(*) = 0 FROM t1;
+
+--connection node_1
+DELETE FROM t1;
+
+START TRANSACTION;
+SAVEPOINT sp1;
+INSERT INTO t1 VALUES (1);
+INSERT INTO t1 VALUES (2);
+INSERT INTO t1 VALUES (3);
+INSERT INTO t1 VALUES (4);
+SAVEPOINT sp2;
+INSERT INTO t1 VALUES (5);
+ROLLBACK TO SAVEPOINT sp2;
+INSERT INTO t1 VALUES (6);
+INSERT INTO t1 VALUES (7);
+ROLLBACK TO SAVEPOINT sp1;
+INSERT INTO t1 VALUES (8);
+COMMIT;
+
+SELECT COUNT(*) = 1 FROM t1;
+
+--connection node_2
+SELECT COUNT(*) = 1 FROM t1;
+
+DROP TABLE t1;

mysql-test/suite/wsrep/r/trans.result

@@ -6,4 +6,5 @@
 #
 START TRANSACTION WITH CONSISTENT SNAPSHOT;
 SAVEPOINT A;
+ERROR HY000: Got error 1 "Operation not permitted" from storage engine binlog
 End of test.

mysql-test/suite/wsrep/t/trans.test

@@ -9,6 +9,7 @@
 --echo #
 
 START TRANSACTION WITH CONSISTENT SNAPSHOT;
+--error 1030
 SAVEPOINT A;
 
 --echo End of test.

sql/log.cc

@@ -2209,9 +2209,6 @@ static int binlog_savepoint_set(handlerton *hton, THD *thd, void *sv)
   int error= 1;
   DBUG_ENTER("binlog_savepoint_set");
 
-  if (wsrep_emulate_bin_log)
-    DBUG_RETURN(0);
-
   char buf[1024];
 
   String log_query(buf, sizeof(buf), &my_charset_bin);
@@ -2245,9 +2242,6 @@ static int binlog_savepoint_rollback(handlerton *hton, THD *thd, void *sv)
 {
   DBUG_ENTER("binlog_savepoint_rollback");
 
-  if (wsrep_emulate_bin_log)
-    DBUG_RETURN(0);
-
   /*
     Write ROLLBACK TO SAVEPOINT to the binlog cache if we have updated some
     non-transactional table. Otherwise, truncate the binlog cache starting

sql/transaction.cc

@@ -614,6 +614,9 @@ bool trans_savepoint(THD *thd, LEX_STRING name)
   if (thd->transaction.xid_state.check_has_uncommitted_xa())
     DBUG_RETURN(TRUE);
 
+  if (WSREP_ON)
+    wsrep_register_hton(thd, thd->in_multi_stmt_transaction_mode());
+
   sv= find_savepoint(thd, name);
 
   if (*sv) /* old savepoint of the same name exists */
@@ -690,6 +693,9 @@ bool trans_rollback_to_savepoint(THD *thd, LEX_STRING name)
   if (thd->transaction.xid_state.check_has_uncommitted_xa())
     DBUG_RETURN(TRUE);
 
+  if (WSREP_ON)
+    wsrep_register_hton(thd, thd->in_multi_stmt_transaction_mode());
+
   /**
     Checking whether it is safe to release metadata locks acquired after
     savepoint, if rollback to savepoint is successful.

sql/wsrep_binlog.cc

@@ -376,20 +376,6 @@ int wsrep_binlog_close_connection(THD* thd)
   DBUG_RETURN(0);
 }
 
-int wsrep_binlog_savepoint_set(THD *thd,  void *sv)
-{
-  if (!wsrep_emulate_bin_log) return 0;
-  int rcode = binlog_hton->savepoint_set(binlog_hton, thd, sv);
-  return rcode;
-}
-
-int wsrep_binlog_savepoint_rollback(THD *thd, void *sv)
-{
-  if (!wsrep_emulate_bin_log) return 0;
-  int rcode = binlog_hton->savepoint_rollback(binlog_hton, thd, sv);
-  return rcode;
-}
-
 #if 0
 void wsrep_dump_rbr_direct(THD* thd, IO_CACHE* cache)
 {

sql/wsrep_binlog.h

@@ -54,7 +54,5 @@ void wsrep_dump_rbr_buf_with_header(THD *thd, const void *rbr_buf,
                                     size_t buf_len);
 
 int wsrep_binlog_close_connection(THD* thd);
-int wsrep_binlog_savepoint_set(THD *thd,  void *sv);
-int wsrep_binlog_savepoint_rollback(THD *thd, void *sv);
 
 #endif /* WSREP_BINLOG_H */

sql/wsrep_hton.cc

@@ -219,32 +219,20 @@ static int wsrep_prepare(handlerton *hton, THD *thd, bool all)
   DBUG_RETURN(0);
 }
 
+/*
+  Empty callbacks to support SAVEPOINT callbacks.
+*/
+
 static int wsrep_savepoint_set(handlerton *hton, THD *thd,  void *sv)
 {
   DBUG_ENTER("wsrep_savepoint_set");
-
-  if (thd->wsrep_exec_mode == REPL_RECV)
-  {
-    DBUG_RETURN(0);
-  }
-
-  if (!wsrep_emulate_bin_log) DBUG_RETURN(0);
-  int rcode = wsrep_binlog_savepoint_set(thd, sv);
-  DBUG_RETURN(rcode);
+  DBUG_RETURN(0);
 }
 
 static int wsrep_savepoint_rollback(handlerton *hton, THD *thd, void *sv)
 {
   DBUG_ENTER("wsrep_savepoint_rollback");
-
-  if (thd->wsrep_exec_mode == REPL_RECV)
-  {
-    DBUG_RETURN(0);
-  }
-
-  if (!wsrep_emulate_bin_log) DBUG_RETURN(0);
-  int rcode = wsrep_binlog_savepoint_rollback(thd, sv);
-  DBUG_RETURN(rcode);
+  DBUG_RETURN(0);
 }
 
 static int wsrep_rollback(handlerton *hton, THD *thd, bool all)