Store wsrep transaction seqno and UUID into Xid_log_event

Prior to this change, wsrep overwrote xid used to prepare
a transaction with a xid containing wsrep seqno and UUID.
This however is not fully compatible with partition engine
and atomic DDL recovery.

In order to avoid overwriting xid during prepare, store the
wsrep transaction seqno and UUID into Xid_log_event if the
transaction is wsrep transaction. This way the seqno and UUID
will be available to reconstruct the wsrep XID for commit
during binlog coordinated recovery.

This changes on-disk format of Xid_log_event to contain
additional 24 bytes for wsrep seqno and UUID if the
transaction is wsrep transaction. As Xid_log_event has no
meaning in replication and used only in recovery, the change
is backwards compatible.

The Xid_log_event output in mariadb-binlog is extended to
print also wsrep_seqno and wsrep_uuid if they are set.

Author: temeo

include/mysql/service_wsrep.h

@@ -54,6 +54,7 @@ extern struct wsrep_service_st {
   int                         (*wsrep_is_wsrep_xid_func)(const void *xid);
   long long                   (*wsrep_xid_seqno_func)(const struct xid_t *xid);
   const unsigned char*        (*wsrep_xid_uuid_func)(const struct xid_t *xid);
+  const struct xid_t*          (*wsrep_commit_xid_func)(const MYSQL_THD thd);
   my_bool                     (*wsrep_on_func)(const MYSQL_THD thd);
   bool                        (*wsrep_prepare_key_for_innodb_func)(MYSQL_THD thd, const unsigned char*, size_t, const unsigned char*, size_t, struct wsrep_buf*, size_t*);
   void                        (*wsrep_thd_LOCK_func)(const MYSQL_THD thd);
@@ -63,6 +64,7 @@ extern struct wsrep_service_st {
   int                         (*wsrep_thd_retry_counter_func)(const MYSQL_THD thd);
   bool                        (*wsrep_thd_ignore_table_func)(MYSQL_THD thd);
   long long                   (*wsrep_thd_trx_seqno_func)(const MYSQL_THD thd);
+  const unsigned char*        (*wsrep_thd_trx_uuid_func)(const MYSQL_THD thd);
   my_bool                     (*wsrep_thd_is_aborting_func)(const MYSQL_THD thd);
   void                        (*wsrep_set_data_home_dir_func)(const char *data_dir);
   my_bool                     (*wsrep_thd_is_BF_func)(const MYSQL_THD thd, my_bool sync);
@@ -109,6 +111,7 @@ extern struct wsrep_service_st {
 #define wsrep_is_wsrep_xid(X) wsrep_service->wsrep_is_wsrep_xid_func(X)
 #define wsrep_xid_seqno(X) wsrep_service->wsrep_xid_seqno_func(X)
 #define wsrep_xid_uuid(X) wsrep_service->wsrep_xid_uuid_func(X)
+#define wsrep_commit_xid(thd) wsrep_service->wsrep_commit_xid_func(thd)
 #define wsrep_on(thd) (thd) && WSREP_ON && wsrep_service->wsrep_on_func(thd)
 #define wsrep_prepare_key_for_innodb(A,B,C,D,E,F,G) wsrep_service->wsrep_prepare_key_for_innodb_func(A,B,C,D,E,F,G)
 #define wsrep_thd_LOCK(T) wsrep_service->wsrep_thd_LOCK_func(T)
@@ -120,6 +123,7 @@ extern struct wsrep_service_st {
 #define wsrep_thd_retry_counter(T) wsrep_service->wsrep_thd_retry_counter_func(T)
 #define wsrep_thd_ignore_table(T) wsrep_service->wsrep_thd_ignore_table_func(T)
 #define wsrep_thd_trx_seqno(T) wsrep_service->wsrep_thd_trx_seqno_func(T)
+#define wsrep_thd_trx_uuid(T) wsrep_service->wsrep_thd_trx_uuid_func(T)
 #define wsrep_set_data_home_dir(A) wsrep_service->wsrep_set_data_home_dir_func(A)
 #define wsrep_thd_is_BF(T,S) wsrep_service->wsrep_thd_is_BF_func(T,S)
 #define wsrep_thd_is_aborting(T) wsrep_service->wsrep_thd_is_aborting_func(T)
@@ -163,7 +167,9 @@ extern "C" const char *wsrep_thd_query(const MYSQL_THD thd);
 extern "C" int wsrep_is_wsrep_xid(const void* xid);
 extern "C" long long wsrep_xid_seqno(const struct xid_t* xid);
 const unsigned char* wsrep_xid_uuid(const struct xid_t* xid);
+extern "C" const struct xid_t* wsrep_commit_xid(const MYSQL_THD thd);
 extern "C" long long wsrep_thd_trx_seqno(const MYSQL_THD thd);
+extern "C" const unsigned char* wsrep_thd_trx_uuid(const MYSQL_THD thd);
 my_bool get_wsrep_recovery();
 bool wsrep_thd_ignore_table(MYSQL_THD thd);
 void wsrep_set_data_home_dir(const char *data_dir);

sql/handler.cc

@@ -1999,15 +1999,6 @@ int ha_commit_trans(THD *thd, bool all)
   xid= thd->transaction->implicit_xid.quick_get_my_xid();
 
 #ifdef WITH_WSREP
-  if (run_wsrep_hooks && !error)
-  {
-    wsrep::seqno const s= wsrep_xid_seqno(thd->wsrep_xid);
-    if (!s.is_undefined())
-    {
-      // xid was rewritten by wsrep
-      xid= s.get();
-    }
-  }
   if (run_wsrep_hooks && (error = wsrep_before_commit(thd, all)))
     goto wsrep_err;
 #endif /* WITH_WSREP */
@@ -2692,7 +2683,26 @@ static void xarecover_do_commit_or_rollback(transaction_participant *hton,
     x= *member->full_xid;
 
   if (xarecover_decide_to_commit(member, ptr_commit_max))
+  {
+#ifdef WITH_WSREP
+    XID wsrep_commit_xid;
+    if (!member->wsrep_seqno.is_undefined())
+    {
+      wsrep::gtid wsrep_gtid{
+          member->wsrep_uuid,
+          member->wsrep_seqno};
+      wsrep_server_gtid_t server_gtid{member->wsrep_gtid_domain_id,
+                                      member->wsrep_gtid_server_id,
+                                      member->wsrep_gtid_seq_no};
+      wsrep_xid_init(&wsrep_commit_xid, wsrep_gtid, server_gtid);
+      wsrep_recovery_commit_xid= &wsrep_commit_xid;
+    }
+#endif /* WITH_WSREP */
     rc= hton->commit_by_xid(&x);
+#ifdef WITH_WSREP
+    wsrep_recovery_commit_xid = nullptr;
+#endif /* WITH_WSREP */
+  }
   else if (hton->recover_rollback_by_xid &&
            IF_WSREP(!(WSREP_ON || wsrep_recovery), true))
     rc= hton->recover_rollback_by_xid(&x);

sql/handler.h

@@ -44,6 +44,10 @@
 #include "mem_root_array.h"
 #include <utility>     // pair
 #include <my_attribute.h> /* __attribute__ */
+#ifdef WITH_WSREP
+#include "wsrep/id.hpp"
+#include "wsrep/seqno.hpp"
+#endif /* WITH_WSREP */
 
 class Alter_info;
 class Virtual_column_info;
@@ -1028,12 +1032,29 @@ struct xid_recovery_member
   XID *full_xid;           // needed by wsrep or past it recovery
   decltype(::server_id) server_id;         // server id of orginal server
 
+#ifdef WITH_WSREP
+  /* wsrep specific fields to reconstruct wsrep_xid for commit.
+     If wsrep_seqno is undefined, the transaction is not a wsrep transaction. */
+  wsrep::seqno wsrep_seqno;
+  wsrep::id wsrep_uuid;
+  uint32 wsrep_gtid_domain_id;
+  uint32 wsrep_gtid_server_id;
+  uint64 wsrep_gtid_seq_no;
+#endif /* WITH_WSREP */
+
   xid_recovery_member(my_xid xid_arg, uint prepare_arg, bool decided_arg,
                       XID *full_xid_arg, decltype(::server_id) server_id_arg)
     : xid(xid_arg), in_engine_prepare(prepare_arg),
       decided_to_commit(decided_arg),
       binlog_coord(Binlog_offset(MAX_binlog_id, MAX_off_t)),
-      full_xid(full_xid_arg), server_id(server_id_arg) {};
+      full_xid(full_xid_arg), server_id(server_id_arg)
+#ifdef WITH_WSREP
+      ,
+      wsrep_seqno(wsrep::seqno::undefined()),
+      wsrep_uuid(wsrep::id::undefined()), wsrep_gtid_domain_id(0),
+      wsrep_gtid_server_id(0), wsrep_gtid_seq_no(0)
+#endif /* WITH_WSREP */
+  {}
 };
 
 /* for recover() handlerton call */

sql/log.cc

@@ -2071,6 +2071,14 @@ binlog_commit_flush_xid_caches(THD *thd, binlog_cache_mngr *cache_mngr,
   DBUG_ASSERT(xid); // replaced former treatment of ONE-PHASE XA
 
   Xid_log_event end_evt(thd, xid, TRUE);
+#ifdef WITH_WSREP
+  if (WSREP(thd))
+  {
+    end_evt.wsrep_seqno= wsrep_thd_trx_seqno(thd);
+    memcpy(end_evt.wsrep_uuid, wsrep_thd_trx_uuid(thd),
+           sizeof(end_evt.wsrep_uuid));
+  }
+#endif
   if (!thd->rgi_slave && !thd->user_time.val)
   {
     /*
@@ -11791,6 +11799,16 @@ class Recovery_context
   bool last_gtid_no2pc; // true when the group does not end with Xid event
   uint last_gtid_engines;
   Binlog_offset last_gtid_coord; // <binlog id, binlog offset>
+#ifdef WITH_WSREP
+  /*
+    Wsrep gtid specific fields are initialized from SE checkpoint and
+    updated when Gtid_log_event with domain id matching last_wsrep_gtid_domain_id
+    is processed.
+  */
+  uint32 last_wsrep_gtid_domain_id;
+  uint32 last_wsrep_gtid_server_id;
+  uint64 last_wsrep_gtid_seq_no;
+#endif /* WITH_WSREP */
   /*
     When true, it's semisync slave recovery mode
     rolls back transactions in doubt and wipes them off from binlog.
@@ -11884,6 +11902,10 @@ class Recovery_context
   */
   void process_gtid(int round, Gtid_log_event *gev, LOG_INFO *linfo);
 
+#ifdef WITH_WSREP
+  void process_wsrep(xid_recovery_member *member, const Xid_log_event *xid_ev);
+#endif /* WITH_WSREP */
+
   /*
     Compute next action at the end of processing of the current binlog file.
     It may increment the round.
@@ -12027,6 +12049,13 @@ Recovery_context::Recovery_context() :
   binlog_unsafe_gtid= truncate_gtid= truncate_gtid_1st_round= rpl_gtid();
   if (do_truncate)
     gtid_maybe_to_truncate= new Dynamic_array<rpl_gtid>(16, 16);
+#ifdef WITH_WSREP
+  wsrep_server_gtid_t wsrep_server_gtid=
+      wsrep_get_SE_checkpoint<wsrep_server_gtid_t>();
+  last_wsrep_gtid_domain_id= wsrep_server_gtid.domain_id;
+  last_wsrep_gtid_server_id= wsrep_server_gtid.server_id;
+  last_wsrep_gtid_seq_no= wsrep_server_gtid.seqno;
+#endif /* WITH_WSREP */
 }
 
 bool Recovery_context::reset_truncate_coord(my_off_t pos)
@@ -12232,8 +12261,30 @@ void Recovery_context::process_gtid(int round, Gtid_log_event *gev,
     /* Update the binlog state with any 'valid' GTID logged after Gtid_list. */
     last_gtid_valid= true;    // may flip at Xid when falls to truncate
   }
+#ifdef WITH_WSREP
+  if (last_gtid.domain_id == last_wsrep_gtid_domain_id)
+  {
+    last_wsrep_gtid_server_id= last_gtid.server_id;
+    last_wsrep_gtid_seq_no= last_gtid.seq_no;
+  }
+#endif /* WITH_WSREP */
 }
 
+#ifdef WITH_WSREP
+void Recovery_context::process_wsrep(xid_recovery_member *member,
+                                     const Xid_log_event *xid_ev)
+{
+  if (member && xid_ev->wsrep_seqno != Xid_log_event::wsrep_seqno_undefined)
+  {
+    member->wsrep_seqno= wsrep::seqno{xid_ev->wsrep_seqno};
+    member->wsrep_uuid= wsrep::id{xid_ev->wsrep_uuid, sizeof(member->wsrep_uuid)};
+    member->wsrep_gtid_domain_id= last_wsrep_gtid_domain_id;
+    member->wsrep_gtid_server_id= last_wsrep_gtid_server_id;
+    member->wsrep_gtid_seq_no= last_wsrep_gtid_seq_no;
+  }
+}
+#endif /* WITH_WSREP */
+
 int Recovery_context::next_binlog_or_round(int& round,
                                            const char *last_log_name,
                                            const char *binlog_checkpoint_name,
@@ -12381,9 +12432,10 @@ int TC_LOG_BINLOG::recover(LOG_INFO *linfo, const char *last_log_name,
       case XID_EVENT:
       if (do_xa)
       {
+        const Xid_log_event *xid_ev= static_cast<const Xid_log_event*>(ev);
         xid_recovery_member *member=
           (xid_recovery_member*)
-          my_hash_search(&xids, (uchar*) &static_cast<Xid_log_event*>(ev)->xid,
+          my_hash_search(&xids, (uchar*) &xid_ev->xid,
                          sizeof(my_xid));
 #ifndef HAVE_REPLICATION
         {
@@ -12393,6 +12445,9 @@ int TC_LOG_BINLOG::recover(LOG_INFO *linfo, const char *last_log_name,
 #else
         if (ctx.decide_or_assess(member, round, fdle, linfo, end_pos))
           goto err2;
+#ifdef WITH_WSREP
+        ctx.process_wsrep(member, xid_ev);
+#endif /* WITH_WSREP */
 #endif
       }
       break;

sql/log_event.cc

@@ -1139,7 +1139,7 @@ Log_event* Log_event::read_log_event(const uchar *buf, uint event_len,
       ev= new Intvar_log_event(buf, fdle);
       break;
     case XID_EVENT:
-      ev= new Xid_log_event(buf, fdle);
+      ev= new Xid_log_event(buf, event_len, fdle);
       break;
     case XA_PREPARE_LOG_EVENT:
       ev= new XA_prepare_log_event(buf, fdle);
@@ -2743,13 +2743,29 @@ Rand_log_event::Rand_log_event(const uchar *buf,
 
 Xid_log_event::
 Xid_log_event(const uchar *buf,
+              uint event_len,
               const Format_description_log_event *description_event)
   :Xid_apply_log_event(buf, description_event)
 {
   /* The Post-Header is empty. The Variable Data part begins immediately. */
   buf+= description_event->common_header_len +
     description_event->post_header_len[XID_EVENT-1];
   memcpy((char*) &xid, buf, sizeof(xid));
+  const uint len_with_wsrep= description_event->common_header_len +
+    description_event->post_header_len[XID_EVENT-1] +
+    sizeof(xid) + sizeof(wsrep_seqno) + sizeof(wsrep_uuid);
+  if (event_len == len_with_wsrep)
+  {
+    buf += sizeof(xid);
+    wsrep_seqno= sint8korr(buf);
+    buf += sizeof(wsrep_seqno);
+    memcpy(wsrep_uuid, buf, sizeof(wsrep_uuid));
+  }
+  else
+  {
+    wsrep_seqno= wsrep_seqno_undefined;
+    memset(wsrep_uuid, 0, sizeof(wsrep_uuid));
+  }
 }
 
 /**************************************************************************

sql/log_event.h

@@ -2823,6 +2823,9 @@ class Xid_apply_log_event: public Log_event
   Logs xid of the transaction-to-be-committed in the 2pc protocol.
   Has no meaning in replication, slaves ignore it.
 
+  Stores wsrep_seqno and wsrep_uuid in the event body if the event
+  was produced by server with wsrep enabled.
+
   @section Xid_log_event_binary_format Binary Format  
 */
 #ifdef MYSQL_CLIENT
@@ -2833,10 +2836,16 @@ class Xid_log_event: public Xid_apply_log_event
 {
 public:
   my_xid xid;
+  /* This matches the definition of WSREP_SEQNO_UNDEFINED and
+     wsrep::seqno::undefined() */
+  constexpr static int64 wsrep_seqno_undefined= -1;
+  int64 wsrep_seqno;
+  uchar wsrep_uuid[16];
 
 #ifdef MYSQL_SERVER
   Xid_log_event(THD* thd_arg, my_xid x, bool direct):
-   Xid_apply_log_event(thd_arg), xid(x)
+   Xid_apply_log_event(thd_arg), xid(x),
+     wsrep_seqno(wsrep_seqno_undefined), wsrep_uuid()
    {
      if (direct)
        cache_type= Log_event::EVENT_NO_CACHE;
@@ -2853,10 +2862,16 @@ class Xid_log_event: public Xid_apply_log_event
 #endif
 
   Xid_log_event(const uchar *buf,
+                uint event_len,
                 const Format_description_log_event *description_event);
   ~Xid_log_event() = default;
   Log_event_type get_type_code() override { return XID_EVENT;}
-  int get_data_size() override { return sizeof(xid); }
+  int get_data_size() override
+  {
+    return sizeof(xid) +
+           (wsrep_seqno != wsrep_seqno_undefined ?
+            sizeof(wsrep_seqno) + sizeof(wsrep_uuid) : 0);
+  }
 #ifdef MYSQL_SERVER
   bool write(Log_event_writer *writer) override;
 #endif

sql/log_event_client.cc

@@ -2464,9 +2464,24 @@ bool Xid_log_event::print(FILE* file, PRINT_EVENT_INFO* print_event_info)
     char buf[64];
     longlong10_to_str(xid, buf, 10);
 
-    if (print_header(&cache, print_event_info, FALSE) ||
-        my_b_printf(&cache, "\tXid = %s\n", buf))
-      goto err;
+    if (wsrep_seqno == wsrep_seqno_undefined)
+    {
+      if (print_header(&cache, print_event_info, FALSE) ||
+          my_b_printf(&cache, "\tXid = %s\n", buf))
+        goto err;
+    }
+    else
+    {
+      char wsrep_seqno_buf[22];
+      longlong10_to_str(wsrep_seqno, wsrep_seqno_buf, 10);
+      char wsrep_uuid_buf[MY_UUID_STRING_LENGTH + 1];
+      my_uuid2str(wsrep_uuid, wsrep_uuid_buf, 1);
+      wsrep_uuid_buf[MY_UUID_STRING_LENGTH]= '\0';
+      if (print_header(&cache, print_event_info, FALSE) ||
+          my_b_printf(&cache, "\tXid = %s, wsrep_seqno = %s, wsrep_uuid = %s\n",
+                      buf, wsrep_seqno_buf, wsrep_uuid_buf))
+        goto err;
+    }
   }
   if (my_b_printf(&cache, is_flashback ? "START TRANSACTION%s\n" : "COMMIT%s\n",
                   print_event_info->delimiter))

sql/log_event_server.cc

@@ -3846,9 +3846,21 @@ int Xid_log_event::do_commit()
 bool Xid_log_event::write(Log_event_writer *writer)
 {
   DBUG_EXECUTE_IF("do_not_write_xid", return 0;);
-  return write_header(writer, sizeof(xid)) ||
-         write_data(writer, (uchar*)&xid, sizeof(xid)) ||
-         write_footer(writer);
+  if (wsrep_seqno == wsrep_seqno_undefined)
+    return write_header(writer, sizeof(xid)) ||
+           write_data(writer, (uchar*)&xid, sizeof(xid)) ||
+           write_footer(writer);
+  else
+  {
+    uchar data[sizeof(xid) + sizeof(wsrep_seqno) + sizeof(wsrep_uuid)];
+    memcpy(data, &xid, sizeof(xid));
+    int8store(data + sizeof(xid), wsrep_seqno);
+    memcpy(data + sizeof(xid) + sizeof(wsrep_seqno), wsrep_uuid,
+           sizeof(wsrep_uuid));
+    return write_header(writer, sizeof(xid) + sizeof(wsrep_seqno) +
+                                    sizeof(wsrep_uuid)) ||
+           write_data(writer, data, sizeof(data)) || write_footer(writer);
+  }
 }
 
 /**************************************************************************