Add support for MariaDB GTID (julien-duponchelle#366)

Samira-El · web-flow · commit 7fd706d1686d · 2022-01-23T21:08:22.000+01:00
* Add MariaDBGtidEvent class
* Add MariaDB specific events
diff --git a/examples/mariadb_gtid/docker-compose.yml b/examples/mariadb_gtid/docker-compose.yml
@@ -0,0 +1,31 @@
+version: '3'
+
+services:
+  testdb:
+    container_name: "testdb"
+    image: mariadb:10.6
+    environment:
+      MARIADB_ALLOW_EMPTY_ROOT_PASSWORD: 1
+      MARIADB_DATABASE: mydb
+      MARIADB_USER: replication_user
+      MARIADB_PASSWORD: secret123passwd
+    ports:
+      - "3306:3306"
+    command: |
+      --server-id=1
+      --default-authentication-plugin=mysql_native_password
+      --log-bin=master-bin
+      --binlog-format=row
+      --log-slave-updates=on
+      --binlog-do-db=mydb
+    volumes:
+      - ./queries.sql:/docker-entrypoint-initdb.d/queries.sql
+    networks:
+      - mariadb-cluster
+
+networks:
+  mariadb-cluster:
+    ipam:
+      driver: default
+      config:
+        - subnet: 172.200.0.0/24
diff --git a/examples/mariadb_gtid/queries.sql b/examples/mariadb_gtid/queries.sql
@@ -0,0 +1,29 @@
+# configure replication user
+grant replication slave on *.* to 'replication_user'@'%';
+flush privileges;
+
+# create objects
+create table r1 (
+    i1 int auto_increment primary key,
+    c1 varchar(10),
+    d1 datetime default current_timestamp()
+);
+
+insert into r1 (c1) values ('#1'),('#2'),('#3'),('#4'),('#5'),('#6'),('#7');
+
+create table r2 (i2 int primary key, d2 datetime) ;
+insert into r2 (i2, d2) values (1, now());
+insert into r2 (i2, d2) values (2, now());
+insert into r2 (i2, d2) values (3, now());
+insert into r2 (i2, d2) values (4, now());
+
+update r1 set c1=concat(c1, '-up');
+
+select * from r2;
+
+delete from r1 where i1 < 4;
+
+drop table r2;
+
+alter table r1 add column b1 bool default False;
+insert into r1 (c1, b1) values ('#8', True);
diff --git a/examples/mariadb_gtid/read_event.py b/examples/mariadb_gtid/read_event.py
@@ -0,0 +1,83 @@
+import pymysql
+
+from pymysqlreplication import BinLogStreamReader, gtid
+from pymysqlreplication.event import GtidEvent, RotateEvent, MariadbGtidEvent, QueryEvent
+from pymysqlreplication.row_event import WriteRowsEvent, UpdateRowsEvent, DeleteRowsEvent
+
+MARIADB_SETTINGS = {
+    "host": "127.0.0.1",
+    "port": 3306,
+    "user": "replication_user",
+    "passwd": "secret123passwd",
+}
+
+
+class MariaDbGTID:
+    def __init__(self, conn_config):
+        self.connection = pymysql.connect(**conn_config)
+
+    def query_single_value(self, sql: str):
+        res = None
+
+        with self.connection.cursor() as cursor:
+            cursor.execute(sql)
+            row = cursor.fetchone()
+            res = str(row[0])
+
+        return res
+
+    def extract_gtid(self, gtid: str, server_id: str):
+        if gtid is None or server_id is None:
+            return None
+
+        gtid_parts = gtid.split("-")
+
+        if len(gtid_parts) != 3:
+            return None
+
+        if gtid_parts[1] == server_id:
+            return gtid
+
+        return None
+
+    def query_gtid_current_pos(self, server_id: str):
+        return self.extract_gtid(self.query_single_value("SELECT @@gtid_current_pos"), server_id)
+
+    def query_server_id(self):
+        return int(self.query_single_value("SELECT @@server_id"))
+
+
+if __name__ == "__main__":
+    db = MariaDbGTID(MARIADB_SETTINGS)
+
+    server_id = db.query_server_id()
+    print('Server ID: ', server_id)
+
+    # gtid = db.query_gtid_current_pos(server_id)
+    gtid = '0-1-1'  # initial pos
+
+    stream = BinLogStreamReader(
+        connection_settings=MARIADB_SETTINGS,
+        server_id=server_id,
+        blocking=False,
+        only_events=[
+            MariadbGtidEvent,
+            RotateEvent,
+            WriteRowsEvent,
+            UpdateRowsEvent,
+            DeleteRowsEvent
+        ],
+        auto_position=gtid,
+        is_mariadb=True
+    )
+
+    print('Starting reading events from GTID ', gtid)
+    for binlogevent in stream:
+        binlogevent.dump()
+
+        if isinstance(binlogevent, MariadbGtidEvent):
+            gtid = binlogevent.gtid
+
+    print('Last encountered GTID: ', gtid)
+
+    stream.close()
diff --git a/pymysqlreplication/binlogstream.py b/pymysqlreplication/binlogstream.py
@@ -14,7 +14,7 @@
     QueryEvent, RotateEvent, FormatDescriptionEvent,
     XidEvent, GtidEvent, StopEvent,
     BeginLoadQueryEvent, ExecuteLoadQueryEvent,
-    HeartbeatLogEvent, NotImplementedEvent)
+    HeartbeatLogEvent, NotImplementedEvent, MariadbGtidEvent)
 from .exceptions import BinLogNotEnabled
 from .row_event import (
     UpdateRowsEvent, WriteRowsEvent, DeleteRowsEvent, TableMapEvent)
@@ -139,7 +139,8 @@ def __init__(self, connection_settings, server_id,
                  report_slave=None, slave_uuid=None,
                  pymysql_wrapper=None,
                  fail_on_table_metadata_unavailable=False,
-                 slave_heartbeat=None):
+                 slave_heartbeat=None,
+                 is_mariadb=False):
         """
         Attributes:
             ctl_connection_settings: Connection settings for cluster holding
@@ -174,6 +175,8 @@ def __init__(self, connection_settings, server_id,
                              many event to skip in binlog). See
                              MASTER_HEARTBEAT_PERIOD in mysql documentation
                              for semantics
+            is_mariadb: Flag to indicate it's a MariaDB server, used with auto_position
+                    to point to Mariadb specific GTID.
         """
 
         self.__connection_settings = connection_settings
@@ -211,6 +214,7 @@ def __init__(self, connection_settings, server_id,
         self.log_file = log_file
         self.auto_position = auto_position
         self.skip_to_timestamp = skip_to_timestamp
+        self.is_mariadb = is_mariadb
 
         if end_log_pos:
             self.is_past_end_log_pos = False
@@ -341,77 +345,114 @@ def __connect_to_stream(self):
             prelude += struct.pack('<I', self.__server_id)
             prelude += self.log_file.encode()
         else:
-            # Format for mysql packet master_auto_position
-            #
-            # All fields are little endian
-            # All fields are unsigned
-
-            # Packet length   uint   4bytes
-            # Packet type     byte   1byte   == 0x1e
-            # Binlog flags    ushort 2bytes  == 0 (for retrocompatibilty)
-            # Server id       uint   4bytes
-            # binlognamesize  uint   4bytes
-            # binlogname      str    Nbytes  N = binlognamesize
-            #                                Zeroified
-            # binlog position uint   4bytes  == 4
-            # payload_size    uint   4bytes
-
-            # What come next, is the payload, where the slave gtid_executed
-            # is sent to the master
-            # n_sid           ulong  8bytes  == which size is the gtid_set
-            # | sid           uuid   16bytes UUID as a binary
-            # | n_intervals   ulong  8bytes  == how many intervals are sent
-            # |                                 for this gtid
-            # | | start       ulong  8bytes  Start position of this interval
-            # | | stop        ulong  8bytes  Stop position of this interval
-
-            # A gtid set looks like:
-            #   19d69c1e-ae97-4b8c-a1ef-9e12ba966457:1-3:8-10,
-            #   1c2aad49-ae92-409a-b4df-d05a03e4702e:42-47:80-100:130-140
-            #
-            # In this particular gtid set,
-            # 19d69c1e-ae97-4b8c-a1ef-9e12ba966457:1-3:8-10
-            # is the first member of the set, it is called a gtid.
-            # In this gtid, 19d69c1e-ae97-4b8c-a1ef-9e12ba966457 is the sid
-            # and have two intervals, 1-3 and 8-10, 1 is the start position of
-            # the first interval 3 is the stop position of the first interval.
-
-            gtid_set = GtidSet(self.auto_position)
-            encoded_data_size = gtid_set.encoded_length
-
-            header_size = (2 +  # binlog_flags
-                           4 +  # server_id
-                           4 +  # binlog_name_info_size
-                           4 +  # empty binlog name
-                           8 +  # binlog_pos_info_size
-                           4)  # encoded_data_size
-
-            prelude = b'' + struct.pack('<i', header_size + encoded_data_size)\
-                + bytes(bytearray([COM_BINLOG_DUMP_GTID]))
+            if self.is_mariadb:
+                # https://mariadb.com/kb/en/5-slave-registration/
+                cur = self._stream_connection.cursor()
 
-            flags = 0
-            if not self.__blocking:
-                flags |= 0x01  # BINLOG_DUMP_NON_BLOCK
-            flags |= 0x04  # BINLOG_THROUGH_GTID
+                cur.execute("SET @mariadb_slave_capability=4")
+                cur.execute("SET @slave_connect_state='%s'" % self.auto_position)
+                cur.execute("SET @slave_gtid_strict_mode=1")
+                cur.execute("SET @slave_gtid_ignore_duplicates=0")
+                cur.close()
 
-            # binlog_flags (2 bytes)
-            # see:
-            #  https://dev.mysql.com/doc/internals/en/com-binlog-dump-gtid.html
-            prelude += struct.pack('<H', flags)
+                # https://mariadb.com/kb/en/com_binlog_dump/
+                header_size = (
+                        4 +  # binlog pos
+                        2 +  # binlog flags
+                        4 +  # slave server_id,
+                        4    # requested binlog file name , set it to empty
+                )
 
-            # server_id (4 bytes)
-            prelude += struct.pack('<I', self.__server_id)
-            # binlog_name_info_size (4 bytes)
-            prelude += struct.pack('<I', 3)
-            # empty_binlog_name (4 bytes)
-            prelude += b'\0\0\0'
-            # binlog_pos_info (8 bytes)
-            prelude += struct.pack('<Q', 4)
-
-            # encoded_data_size (4 bytes)
-            prelude += struct.pack('<I', gtid_set.encoded_length)
-            # encoded_data
-            prelude += gtid_set.encoded()
+                prelude = struct.pack('<i', header_size) + bytes(bytearray([COM_BINLOG_DUMP]))
+
+                # binlog pos
+                prelude += struct.pack('<i', 4)
+
+                flags = 0
+                if not self.__blocking:
+                    flags |= 0x01  # BINLOG_DUMP_NON_BLOCK
+                
+                # binlog flags
+                prelude += struct.pack('<H', flags)
+
+                # server id (4 bytes)
+                prelude += struct.pack('<I', self.__server_id)
+
+                # empty_binlog_name (4 bytes)
+                prelude += b'\0\0\0\0'
+                
+            else:
+                # Format for mysql packet master_auto_position
+                #
+                # All fields are little endian
+                # All fields are unsigned
+
+                # Packet length   uint   4bytes
+                # Packet type     byte   1byte   == 0x1e
+                # Binlog flags    ushort 2bytes  == 0 (for retrocompatibilty)
+                # Server id       uint   4bytes
+                # binlognamesize  uint   4bytes
+                # binlogname      str    Nbytes  N = binlognamesize
+                #                                Zeroified
+                # binlog position uint   4bytes  == 4
+                # payload_size    uint   4bytes
+
+                # What come next, is the payload, where the slave gtid_executed
+                # is sent to the master
+                # n_sid           ulong  8bytes  == which size is the gtid_set
+                # | sid           uuid   16bytes UUID as a binary
+                # | n_intervals   ulong  8bytes  == how many intervals are sent
+                # |                                 for this gtid
+                # | | start       ulong  8bytes  Start position of this interval
+                # | | stop        ulong  8bytes  Stop position of this interval
+
+                # A gtid set looks like:
+                #   19d69c1e-ae97-4b8c-a1ef-9e12ba966457:1-3:8-10,
+                #   1c2aad49-ae92-409a-b4df-d05a03e4702e:42-47:80-100:130-140
+                #
+                # In this particular gtid set,
+                # 19d69c1e-ae97-4b8c-a1ef-9e12ba966457:1-3:8-10
+                # is the first member of the set, it is called a gtid.
+                # In this gtid, 19d69c1e-ae97-4b8c-a1ef-9e12ba966457 is the sid
+                # and have two intervals, 1-3 and 8-10, 1 is the start position of
+                # the first interval 3 is the stop position of the first interval.
+
+                gtid_set = GtidSet(self.auto_position)
+                encoded_data_size = gtid_set.encoded_length
+
+                header_size = (2 +  # binlog_flags
+                               4 +  # server_id
+                               4 +  # binlog_name_info_size
+                               4 +  # empty binlog name
+                               8 +  # binlog_pos_info_size
+                               4)  # encoded_data_size
+
+                prelude = b'' + struct.pack('<i', header_size + encoded_data_size)\
+                    + bytes(bytearray([COM_BINLOG_DUMP_GTID]))
+
+                flags = 0
+                if not self.__blocking:
+                    flags |= 0x01  # BINLOG_DUMP_NON_BLOCK
+                flags |= 0x04  # BINLOG_THROUGH_GTID
+
+                # binlog_flags (2 bytes)
+                # see:
+                #  https://dev.mysql.com/doc/internals/en/com-binlog-dump-gtid.html
+                prelude += struct.pack('<H', flags)
+
+                # server_id (4 bytes)
+                prelude += struct.pack('<I', self.__server_id)
+                # binlog_name_info_size (4 bytes)
+                prelude += struct.pack('<I', 3)
+                # empty_binlog_namapprovale (4 bytes)
+                prelude += b'\0\0\0'
+                # binlog_pos_info (8 bytes)
+                prelude += struct.pack('<Q', 4)
+
+                # encoded_data_size (4 bytes)
+                prelude += struct.pack('<I', gtid_set.encoded_length)
+                # encoded_data
+                prelude += gtid_set.encoded()
 
         if pymysql.__version__ < LooseVersion("0.6"):
             self._stream_connection.wfile.write(prelude)
@@ -542,6 +583,7 @@ def _allowed_event_list(self, only_events, ignored_events,
                 TableMapEvent,
                 HeartbeatLogEvent,
                 NotImplementedEvent,
+                MariadbGtidEvent
                 ))
         if ignored_events is not None:
             for e in ignored_events:
diff --git a/pymysqlreplication/constants/BINLOG.py b/pymysqlreplication/constants/BINLOG.py
@@ -41,3 +41,10 @@
 INTVAR_INVALID_INT_EVENT = 0x00
 INTVAR_LAST_INSERT_ID_EVENT = 0x01
 INTVAR_INSERT_ID_EVENT = 0x02
+
+# MariaDB events
+MARIADB_ANNOTATE_ROWS_EVENT = 0xa0
+MARIADB_BINLOG_CHECKPOINT_EVENT = 0xa1
+MARIADB_GTID_EVENT = 0xa2
+MARIADB_GTID_GTID_LIST_EVENT = 0xa3
+MARIADB_START_ENCRYPTION_EVENT = 0xa4
diff --git a/pymysqlreplication/event.py b/pymysqlreplication/event.py
diff --git a/pymysqlreplication/packet.py b/pymysqlreplication/packet.py
