Galera GTID support

Support for galera GTID consistency thru cluster. All nodes in cluster
should have same GTID for replicated events which are originating from cluster.
Cluster originating commands need to contain sequential WSREP GTID seqno
Ignore manual setting of gtid_seq_no=X.

In master-slave scenario where master is non galera node replicated GTID is
replicated and is preserved in all nodes.

To have this - domain_id, server_id and seqnos should be same on all nodes.
Node which bootstraps the cluster, to achieve this, sends domain_id and
server_id to other nodes and this combination is used to write GTID for events
that are replicated inside cluster.

Cluster nodes that are executing non replicated events are going to have different
GTID than replicated ones, difference will be visible in domain part of gtid.

With wsrep_gtid_domain_id you can set domain_id for WSREP cluster.

Functions WSREP_LAST_WRITTEN_GTID, WSREP_LAST_SEEN_GTID and
WSREP_SYNC_WAIT_UPTO_GTID now works with "native" GTID format.

Fixed galera tests to reflect this chances.

Add variable to manually update WSREP GTID seqno in cluster

Add variable to manipulate and change WSREP GTID seqno. Next command
originating from cluster and on same thread will have set seqno and
cluster should change their internal counter to it's value.
Behavior is same as using @@gtid_seq_no for non WSREP transaction.

1 files changed, 52 insertions, 14 deletions

diff --git a/sql/wsrep_xid.cc b/sql/wsrep_xid.cc
index d8f6e013820..c84071e13d0 100644
--- a/sql/wsrep_xid.cc
+++ b/sql/wsrep_xid.cc
@@ -33,20 +33,24 @@
 #define WSREP_XID_VERSION_OFFSET WSREP_XID_PREFIX_LEN
 #define WSREP_XID_VERSION_1 'd'
 #define WSREP_XID_VERSION_2 'e'
+#define WSREP_XID_VERSION_3 'f'
 #define WSREP_XID_UUID_OFFSET 8
 #define WSREP_XID_SEQNO_OFFSET (WSREP_XID_UUID_OFFSET + sizeof(wsrep_uuid_t))
-#define WSREP_XID_GTRID_LEN (WSREP_XID_SEQNO_OFFSET + sizeof(wsrep_seqno_t))
+#define WSREP_XID_GTRID_LEN_V_1_2 (WSREP_XID_SEQNO_OFFSET + sizeof(wsrep_seqno_t))
+#define WSREP_XID_RPL_GTID_OFFSET (WSREP_XID_SEQNO_OFFSET + sizeof(wsrep_seqno_t))
+#define WSREP_XID_GTRID_LEN_V_3 (WSREP_XID_RPL_GTID_OFFSET + sizeof(wsrep_server_gtid_t))
 
-void wsrep_xid_init(XID* xid, const wsrep::gtid& wsgtid)
+void wsrep_xid_init(XID* xid, const wsrep::gtid& wsgtid, const wsrep_server_gtid_t& gtid)
 {
   xid->formatID= 1;
-  xid->gtrid_length= WSREP_XID_GTRID_LEN;
+  xid->gtrid_length= WSREP_XID_GTRID_LEN_V_3;
   xid->bqual_length= 0;
   memset(xid->data, 0, sizeof(xid->data));
   memcpy(xid->data, WSREP_XID_PREFIX, WSREP_XID_PREFIX_LEN);
-  xid->data[WSREP_XID_VERSION_OFFSET]= WSREP_XID_VERSION_2;
+  xid->data[WSREP_XID_VERSION_OFFSET]= WSREP_XID_VERSION_3;
   memcpy(xid->data + WSREP_XID_UUID_OFFSET,  wsgtid.id().data(),sizeof(wsrep::id));
   int8store(xid->data + WSREP_XID_SEQNO_OFFSET, wsgtid.seqno().get());
+  memcpy(xid->data + WSREP_XID_RPL_GTID_OFFSET, &gtid, sizeof(wsrep_server_gtid_t));
 }
 
 extern "C"
@@ -54,11 +58,13 @@ int wsrep_is_wsrep_xid(const void* xid_ptr)
 {
   const XID* xid= static_cast<const XID*>(xid_ptr);
   return (xid->formatID      == 1                   &&
-          xid->gtrid_length  == WSREP_XID_GTRID_LEN &&
           xid->bqual_length  == 0                   &&
-          !memcmp(xid->data, WSREP_XID_PREFIX, WSREP_XID_PREFIX_LEN) &&
-          (xid->data[WSREP_XID_VERSION_OFFSET] == WSREP_XID_VERSION_1 ||
-           xid->data[WSREP_XID_VERSION_OFFSET] == WSREP_XID_VERSION_2));
+          !memcmp(xid->data, WSREP_XID_PREFIX, WSREP_XID_PREFIX_LEN)      &&
+          (((xid->data[WSREP_XID_VERSION_OFFSET] == WSREP_XID_VERSION_1   ||
+             xid->data[WSREP_XID_VERSION_OFFSET] == WSREP_XID_VERSION_2)  &&
+            xid->gtrid_length  == WSREP_XID_GTRID_LEN_V_1_2)              ||
+           (xid->data[WSREP_XID_VERSION_OFFSET] == WSREP_XID_VERSION_3    &&
+            xid->gtrid_length  == WSREP_XID_GTRID_LEN_V_3)));
 }
 
 const unsigned char* wsrep_xid_uuid(const xid_t* xid)
@@ -90,6 +96,7 @@ long long wsrep_xid_seqno(const xid_t* xid)
       memcpy(&ret, xid->data + WSREP_XID_SEQNO_OFFSET, sizeof ret);
       break;
     case WSREP_XID_VERSION_2:
+    case WSREP_XID_VERSION_3:
       ret= sint8korr(xid->data + WSREP_XID_SEQNO_OFFSET);
       break;
     default:
@@ -127,10 +134,10 @@ bool wsrep_set_SE_checkpoint(XID& xid)
                         &xid);
 }
 
-bool wsrep_set_SE_checkpoint(const wsrep::gtid& wsgtid)
+bool wsrep_set_SE_checkpoint(const wsrep::gtid& wsgtid, const wsrep_server_gtid_t& gtid)
 {
   XID xid;
-  wsrep_xid_init(&xid, wsgtid);
+  wsrep_xid_init(&xid, wsgtid, gtid);
   return wsrep_set_SE_checkpoint(xid);
 }
 
@@ -158,30 +165,61 @@ bool wsrep_get_SE_checkpoint(XID& xid)
                         &xid);
 }
 
-wsrep::gtid wsrep_get_SE_checkpoint()
+static bool wsrep_get_SE_checkpoint_common(XID& xid)
 {
-  XID xid;
   xid.null();
 
   if (wsrep_get_SE_checkpoint(xid))
   {
-    return wsrep::gtid();
+    return FALSE;
   }
 
   if (xid.is_null())
   {
-    return wsrep::gtid();
+    return FALSE;
   }
 
   if (!wsrep_is_wsrep_xid(&xid))
   {
     WSREP_WARN("Read non-wsrep XID from storage engines.");
+    return FALSE;
+  }
+
+  return TRUE;
+}
+
+template<>
+wsrep::gtid wsrep_get_SE_checkpoint()
+{
+  XID xid;
+
+  if (!wsrep_get_SE_checkpoint_common(xid))
+  {
     return wsrep::gtid();
   }
 
   return wsrep::gtid(wsrep_xid_uuid(xid),wsrep_xid_seqno(xid));
 }
 
+template<>
+wsrep_server_gtid_t wsrep_get_SE_checkpoint()
+{
+  XID xid;
+  wsrep_server_gtid_t gtid= {0,0,0};
+
+  if (!wsrep_get_SE_checkpoint_common(xid))
+  {
+    return gtid;
+  }
+
+  if (xid.data[WSREP_XID_VERSION_OFFSET] == WSREP_XID_VERSION_3)
+  {
+    memcpy(&gtid, &xid.data[WSREP_XID_RPL_GTID_OFFSET], sizeof(wsrep_server_gtid_t));
+  }
+
+  return gtid;
+}
+
 /*
   Sort order for XIDs. Wsrep XIDs are sorted according to
   seqno in ascending order. Non-wsrep XIDs are considered