1 files changed, 20 insertions, 36 deletions

diff --git a/sql/transaction.cc b/sql/transaction.cc
index 1c2820200d1..4d61d2a120d 100644
--- a/sql/transaction.cc
+++ b/sql/transaction.cc
@@ -24,6 +24,9 @@
 #include "debug_sync.h"         // DEBUG_SYNC
 #include "sql_acl.h"
 #include "semisync_master.h"
+#ifdef WITH_WSREP
+#include "wsrep_trans_observer.h"
+#endif /* WITH_WSREP */
 
 #ifndef EMBEDDED_LIBRARY
 /**
@@ -135,8 +138,6 @@ static bool xa_trans_force_rollback(THD *thd)
     by ha_rollback()/THD::transaction::cleanup().
   */
   thd->transaction.xid_state.rm_error= 0;
-  if (WSREP_ON)
-    wsrep_register_hton(thd, TRUE);
   if (ha_rollback_trans(thd, true))
   {
     my_error(ER_XAER_RMERR, MYF(0));
@@ -184,14 +185,16 @@ bool trans_begin(THD *thd, uint flags)
       (thd->variables.option_bits & OPTION_TABLE_LOCK))
   {
     thd->variables.option_bits&= ~OPTION_TABLE_LOCK;
-    if (WSREP_ON)
-      wsrep_register_hton(thd, TRUE);
     thd->server_status&=
       ~(SERVER_STATUS_IN_TRANS | SERVER_STATUS_IN_TRANS_READONLY);
     DBUG_PRINT("info", ("clearing SERVER_STATUS_IN_TRANS"));
     res= MY_TEST(ha_commit_trans(thd, TRUE));
-    if (WSREP_ON)
-      wsrep_post_commit(thd, TRUE);
+#ifdef WITH_WSREP
+    if (wsrep_thd_is_local(thd))
+    {
+      res= res || wsrep_after_statement(thd);
+    }
+#endif /* WITH_WSREP */
   }
 
   thd->variables.option_bits&= ~(OPTION_BEGIN | OPTION_KEEP_LOG);
@@ -252,9 +255,14 @@ bool trans_begin(THD *thd, uint flags)
   }
 
 #ifdef WITH_WSREP
-  thd->wsrep_PA_safe= true;
-  if (WSREP_CLIENT(thd) && wsrep_sync_wait(thd))
-    DBUG_RETURN(TRUE);
+  if (wsrep_thd_is_local(thd))
+  {
+    if (wsrep_sync_wait(thd))
+      DBUG_RETURN(TRUE);
+    if (!thd->tx_read_only &&
+        wsrep_start_transaction(thd, thd->wsrep_next_trx_id()))
+      DBUG_RETURN(TRUE);
+  }
 #endif /* WITH_WSREP */
 
   thd->variables.option_bits|= OPTION_BEGIN;
@@ -299,8 +307,6 @@ bool trans_commit(THD *thd)
   if (trans_check(thd))
     DBUG_RETURN(TRUE);
 
-  if (WSREP_ON)
-    wsrep_register_hton(thd, TRUE);
   thd->server_status&=
     ~(SERVER_STATUS_IN_TRANS | SERVER_STATUS_IN_TRANS_READONLY);
   DBUG_PRINT("info", ("clearing SERVER_STATUS_IN_TRANS"));
@@ -311,8 +317,6 @@ bool trans_commit(THD *thd)
   mysql_mutex_assert_not_owner(&LOCK_after_binlog_sync);
   mysql_mutex_assert_not_owner(&LOCK_commit_ordered);
 
-  if (WSREP_ON)
-    wsrep_post_commit(thd, TRUE);
     /*
       if res is non-zero, then ha_commit_trans has rolled back the
       transaction, so the hooks for rollback will be called.
@@ -368,14 +372,10 @@ bool trans_commit_implicit(THD *thd)
     /* Safety if one did "drop table" on locked tables */
     if (!thd->locked_tables_mode)
       thd->variables.option_bits&= ~OPTION_TABLE_LOCK;
-    if (WSREP_ON)
-      wsrep_register_hton(thd, TRUE);
     thd->server_status&=
       ~(SERVER_STATUS_IN_TRANS | SERVER_STATUS_IN_TRANS_READONLY);
     DBUG_PRINT("info", ("clearing SERVER_STATUS_IN_TRANS"));
     res= MY_TEST(ha_commit_trans(thd, TRUE));
-    if (WSREP_ON)
-      wsrep_post_commit(thd, TRUE);
   }
 
   thd->variables.option_bits&= ~(OPTION_BEGIN | OPTION_KEEP_LOG);
@@ -409,14 +409,9 @@ bool trans_rollback(THD *thd)
   int res;
   DBUG_ENTER("trans_rollback");
 
-#ifdef WITH_WSREP
-  thd->wsrep_PA_safe= true;
-#endif /* WITH_WSREP */
   if (trans_check(thd))
     DBUG_RETURN(TRUE);
 
-  if (WSREP_ON)
-    wsrep_register_hton(thd, TRUE);
   thd->server_status&=
     ~(SERVER_STATUS_IN_TRANS | SERVER_STATUS_IN_TRANS_READONLY);
   DBUG_PRINT("info", ("clearing SERVER_STATUS_IN_TRANS"));
@@ -515,14 +510,10 @@ bool trans_commit_stmt(THD *thd)
 
   if (thd->transaction.stmt.ha_list)
   {
-    if (WSREP_ON)
-      wsrep_register_hton(thd, FALSE);
     res= ha_commit_trans(thd, FALSE);
     if (! thd->in_active_multi_stmt_transaction())
     {
       trans_reset_one_shot_chistics(thd);
-      if (WSREP_ON)
-        wsrep_post_commit(thd, FALSE);
     }
   }
 
@@ -578,8 +569,6 @@ bool trans_rollback_stmt(THD *thd)
 
   if (thd->transaction.stmt.ha_list)
   {
-    if (WSREP_ON)
-      wsrep_register_hton(thd, FALSE);
     ha_rollback_trans(thd, FALSE);
     if (! thd->in_active_multi_stmt_transaction())
       trans_reset_one_shot_chistics(thd);
@@ -733,7 +722,8 @@ bool trans_rollback_to_savepoint(THD *thd, LEX_CSTRING name)
     logging is off.
   */
   bool mdl_can_safely_rollback_to_savepoint=
-                (!(mysql_bin_log.is_open() && thd->variables.sql_log_bin) ||
+                (!((WSREP_EMULATE_BINLOG_NNULL(thd) || mysql_bin_log.is_open())
+                 && thd->variables.sql_log_bin) ||
                  ha_rollback_to_savepoint_can_release_mdl(thd));
 
   if (ha_rollback_to_savepoint(thd, sv))
@@ -944,13 +934,9 @@ bool trans_xa_commit(THD *thd)
   }
   else if (xa_state == XA_IDLE && thd->lex->xa_opt == XA_ONE_PHASE)
   {
-    if (WSREP_ON)
-      wsrep_register_hton(thd, TRUE);
     int r= ha_commit_trans(thd, TRUE);
     if ((res= MY_TEST(r)))
       my_error(r == 1 ? ER_XA_RBROLLBACK : ER_XAER_RMERR, MYF(0));
-    if (WSREP_ON)
-      wsrep_post_commit(thd, TRUE);
   }
   else if (xa_state == XA_PREPARED && thd->lex->xa_opt == XA_NONE)
   {
@@ -963,14 +949,12 @@ bool trans_xa_commit(THD *thd)
 
       We allow FLUSHer to COMMIT; we assume FLUSHer knows what it does.
     */
-    mdl_request.init(MDL_key::COMMIT, "", "", MDL_INTENTION_EXCLUSIVE,
+    mdl_request.init(MDL_key::BACKUP, "", "", MDL_BACKUP_COMMIT,
                      MDL_TRANSACTION);
 
     if (thd->mdl_context.acquire_lock(&mdl_request,
                                       thd->variables.lock_wait_timeout))
     {
-      if (WSREP_ON)
-        wsrep_register_hton(thd, TRUE);
       ha_rollback_trans(thd, TRUE);
       my_error(ER_XAER_RMERR, MYF(0));
     }