Merge branch 'feature/previous-gtid-event' of github.com:23-OSSCA-python-mysql-replication/python-mysql-replication into 23-OSSCA-python-mysql-replication-feature/previous-gtid-event

Author: julien-duponchelle

pymysqlreplication/binlogstream.py

@@ -14,7 +14,8 @@
     BeginLoadQueryEvent, ExecuteLoadQueryEvent,
     HeartbeatLogEvent, NotImplementedEvent, MariadbGtidEvent,
     MariadbAnnotateRowsEvent, RandEvent, MariadbStartEncryptionEvent, RowsQueryLogEvent,
-    MariadbGtidListEvent, MariadbBinLogCheckPointEvent, UserVarEvent)
+    MariadbGtidListEvent, MariadbBinLogCheckPointEvent, UserVarEvent,
+    PreviousGtidsEvent)
 from .exceptions import BinLogNotEnabled
 from .gtid import GtidSet
 from .packet import BinLogPacketWrapper
@@ -183,7 +184,7 @@ def __init__(self, connection_settings, server_id,
                     to point to Mariadb specific GTID.
             annotate_rows_event: Parameter value to enable annotate rows event in mariadb,
                     used with 'is_mariadb'
-            ignore_decode_errors: If true, any decode errors encountered 
+            ignore_decode_errors: If true, any decode errors encountered
                                   when reading column data will be ignored.
             verify_checksum: If true, verify events read from the binary log by examining checksums.
         """
@@ -478,7 +479,7 @@ def __set_mariadb_settings(self):
 
         flags = 0
 
-        # Enable annotate rows event 
+        # Enable annotate rows event
         if self.__annotate_rows_event:
             flags |= 0x02  # BINLOG_SEND_ANNOTATE_ROWS_EVENT
 
@@ -631,7 +632,8 @@ def _allowed_event_list(self, only_events, ignored_events,
                 MariadbStartEncryptionEvent,
                 MariadbGtidListEvent,
                 MariadbBinLogCheckPointEvent,
-                UserVarEvent
+                UserVarEvent,
+                PreviousGtidsEvent
             ))
         if ignored_events is not None:
             for e in ignored_events:

pymysqlreplication/event.py

@@ -121,6 +121,40 @@ def __repr__(self):
         return '<GtidEvent "%s">' % self.gtid
 
 
+class PreviousGtidsEvent(BinLogEvent):
+    """
+    PreviousGtidEvent is contains the Gtids executed in the last binary log file.
+    Attributes:
+        n_sid: which size is the gtid_set
+        sid: 16bytes UUID as a binary
+        n_intervals: how many intervals are sent
+    Eg: [4c9e3dfc-9d25-11e9-8d2e-0242ac1cfd7e:1-100, 4c9e3dfc-9d25-11e9-8d2e-0242ac1cfd7e:1-10:20-30]
+    """
+    def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs):
+        super(PreviousGtidsEvent, self).__init__(from_packet, event_size, table_map,
+                                                ctl_connection, **kwargs)
+
+        self._n_sid = self.packet.read_int64()
+        self._gtids = []
+
+        for _ in range(self._n_sid):
+            sid = self.packet.read(16)
+            n_intervals = self.packet.read_uint64()
+            intervals = [f"{self.packet.read_int64()}-{self.packet.read_uint64()}" for _ in range(n_intervals)]
+            nibbles = binascii.hexlify(sid).decode('ascii')
+            gtid = '%s-%s-%s-%s-%s:%s' % (
+                nibbles[:8], nibbles[8:12], nibbles[12:16], nibbles[16:20], nibbles[20:], ':'.join(intervals))
+            self._gtids.append(gtid)
+
+        self._previous_gtids = ','.join(self._gtids)
+
+    def _dump(self):
+        print("previous_gtids: %s" % self._previous_gtids)
+
+    def __repr__(self):
+        return '<PreviousGtidsEvent "%s">' % self._previous_gtids
+
+
 class MariadbGtidEvent(BinLogEvent):
     """
     GTID(Global Transaction Identifier) change in binlog event in MariaDB

pymysqlreplication/packet.py

@@ -65,6 +65,7 @@ class BinLogPacketWrapper(object):
         constants.XID_EVENT: event.XidEvent,
         constants.INTVAR_EVENT: event.IntvarEvent,
         constants.GTID_LOG_EVENT: event.GtidEvent,
+        constants.PREVIOUS_GTIDS_LOG_EVENT: event.PreviousGtidsEvent,
         constants.STOP_EVENT: event.StopEvent,
         constants.BEGIN_LOAD_QUERY_EVENT: event.BeginLoadQueryEvent,
         constants.EXECUTE_LOAD_QUERY_EVENT: event.ExecuteLoadQueryEvent,
@@ -84,8 +85,6 @@ class BinLogPacketWrapper(object):
 
         #5.6 GTID enabled replication events
         constants.ANONYMOUS_GTID_LOG_EVENT: event.NotImplementedEvent,
-        constants.ANONYMOUS_GTID_LOG_EVENT: event.NotImplementedEvent,
-        constants.PREVIOUS_GTIDS_LOG_EVENT: event.NotImplementedEvent,
         # MariaDB GTID
         constants.MARIADB_ANNOTATE_ROWS_EVENT: event.MariadbAnnotateRowsEvent,
         constants.MARIADB_BINLOG_CHECKPOINT_EVENT: event.MariadbBinLogCheckPointEvent,

pymysqlreplication/tests/test_basic.py

@@ -28,7 +28,7 @@
 
 class TestBasicBinLogStreamReader(base.PyMySQLReplicationTestCase):
     def ignoredEvents(self):
-        return [GtidEvent]
+        return [GtidEvent, PreviousGtidsEvent]
 
     def test_allowed_event_list(self):
         self.assertEqual(len(self.stream._allowed_event_list(None, None, False)), 23)
@@ -577,7 +577,7 @@ def create_binlog_packet_wrapper(pkt):
 
 class TestMultipleRowBinLogStreamReader(base.PyMySQLReplicationTestCase):
     def ignoredEvents(self):
-        return [GtidEvent]
+        return [GtidEvent, PreviousGtidsEvent]
 
     def test_insert_multiple_row_event(self):
         query = "CREATE TABLE test (id INT NOT NULL AUTO_INCREMENT, data VARCHAR (50) NOT NULL, PRIMARY KEY (id))"
@@ -902,6 +902,7 @@ def test_read_query_event(self):
         query = "COMMIT;"
         self.execute(query)
 
+        self.assertIsInstance(self.stream.fetchone(), PreviousGtidsEvent)
         firstevent = self.stream.fetchone()
         self.assertIsInstance(firstevent, GtidEvent)
 
@@ -951,6 +952,7 @@ def test_position_gtid(self):
 
         self.assertIsInstance(self.stream.fetchone(), RotateEvent)
         self.assertIsInstance(self.stream.fetchone(), FormatDescriptionEvent)
+        self.assertIsInstance(self.stream.fetchone(), PreviousGtidsEvent)
         self.assertIsInstance(self.stream.fetchone(), GtidEvent)
         event = self.stream.fetchone()
 

pymysqlreplication/tests/test_data_type.py

@@ -32,7 +32,7 @@ def encode_value(v):
 
 class TestDataType(base.PyMySQLReplicationTestCase):
     def ignoredEvents(self):
-        return [GtidEvent]
+        return [GtidEvent, PreviousGtidsEvent]
 
     def create_and_insert_value(self, create_query, insert_query):
         self.execute(create_query)