Merge branch 'main' into feature/rand-event

Author: mikaniz

.github/workflows/pytest.yml

@@ -28,9 +28,17 @@ jobs:
         run: |
           docker compose create
           docker compose start
-          # Wait for the services to accept connections,
-          # TODO: do that smarter, poll connection attempt until it succeeds
-          sleep 30
+          echo "wait mysql server"
+        
+          while :
+          do
+            if mysql -h 127.0.0.1 --user=root --execute "SELECT version();" 2>&1 >/dev/null && mysql -h 127.0.0.1 --port=3307  --user=root --execute "SELECT version();" 2>&1 >/dev/null; then
+              break
+            fi
+            sleep 1
+          done
+          
+          echo "run pytest"
 
       - name: Install dependencies
         run: |

README.md

@@ -4,7 +4,7 @@ python-mysql-replication
 <a href="https://travis-ci.org/noplay/python-mysql-replication"><img src="https://travis-ci.org/noplay/python-mysql-replication.svg?branch=master"></a>&nbsp;
 <a href="https://pypi.python.org/pypi/mysql-replication"><img src="http://img.shields.io/pypi/dm/mysql-replication.svg"></a>
 
-Pure Python Implementation of MySQL replication protocol build on top of PyMYSQL. This allow you to receive event like insert, update, delete with their datas and raw SQL queries.
+Pure Python Implementation of MySQL replication protocol build on top of PyMYSQL. This allows you to receive event like insert, update, delete with their datas and raw SQL queries.
 
 Use cases
 ===========
@@ -56,6 +56,11 @@ Limitations
 
 https://python-mysql-replication.readthedocs.org/en/latest/limitations.html
 
+Featured Books
+=============
+
+[Data Pipelines Pocket Reference](https://www.oreilly.com/library/view/data-pipelines-pocket/9781492087823/) (by James Densmore, O'Reilly): Introduced and exemplified in Chapter 4: Data Ingestion: Extracting Data.
+
 Projects using this library
 ===========================
 

docker-compose-test.yml

@@ -0,0 +1,59 @@
+version: '3.2'
+services:
+  percona-5.7:
+    platform: linux/amd64
+    image: percona:5.7
+    environment:
+      MYSQL_ALLOW_EMPTY_PASSWORD: true
+      MYSQL_DATABASE: pymysqlreplication_test
+    ports:
+      - 3306:3306
+    command: mysqld --log-bin=mysql-bin.log --server-id 1 --binlog-format=row --gtid_mode=on --enforce-gtid-consistency=on --log_slave_updates
+    restart: always
+    networks:
+      - default
+
+  percona-5.7-ctl:
+    image: percona:5.7
+    environment:
+      MYSQL_ALLOW_EMPTY_PASSWORD: true
+      MYSQL_DATABASE: pymysqlreplication_test
+    ports:
+      - 3307:3307
+    command: mysqld --log-bin=mysql-bin.log --server-id 1 --binlog-format=row --gtid_mode=on --enforce-gtid-consistency=on --log_slave_updates -P 3307
+
+  pymysqlreplication:
+    build:
+      context: .
+      dockerfile: test.Dockerfile
+      args:
+        BASE_IMAGE: python:3.11-alpine
+        MYSQL_5_7: percona-5.7
+        MYSQL_5_7_CTL: percona-5.7-ctl
+
+    command:
+      - /bin/sh
+      - -ce
+      - |
+        echo "wait mysql server"
+        
+        while :
+        do
+          if mysql -h percona-5.7 --user=root --execute "USE pymysqlreplication_test;" 2>&1 >/dev/null && mysql -h percona-5.7-ctl --port=3307  --user=root --execute "USE pymysqlreplication_test;" 2>&1 >/dev/null; then
+            break
+          fi
+          sleep 1
+        done
+        
+        echo "run pytest"
+        pytest -k "not test_no_trailing_rotate_event and not test_end_log_pos"
+
+    working_dir: /pymysqlreplication
+    networks:
+      - default
+    depends_on:
+      - percona-5.7
+      - percona-5.7-ctl
+
+networks:
+  default: {}

docker-compose.yml

@@ -15,3 +15,17 @@ services:
     ports:
       - 3307:3307
     command: mysqld --log-bin=mysql-bin.log --server-id 1 --binlog-format=row --gtid_mode=on --enforce-gtid-consistency=on --log_slave_updates -P 3307
+
+  mariadb-10.6:
+    image: mariadb:10.6
+    environment:
+      MARIADB_ALLOW_EMPTY_ROOT_PASSWORD: 1
+    ports:
+      - "3308:3306"
+    command: |
+      --server-id=1
+      --default-authentication-plugin=mysql_native_password
+      --log-bin=master-bin
+      --binlog-format=row
+      --log-slave-updates=on
+      

docs/developement.rst

@@ -23,7 +23,7 @@ When it's possible we have an unit test.
 *pymysqlreplication/tests/* contains the test suite. The test suite
 use the standard *unittest* Python module.
 
-**Be carefull** tests will reset the binary log of your MySQL server.
+**Be careful** tests will reset the binary log of your MySQL server.
 
 Make sure you have the following configuration set in your mysql config file (usually my.cnf on development env):
 

examples/mariadb_gtid/read_event.py

@@ -1,7 +1,7 @@
 import pymysql
 
 from pymysqlreplication import BinLogStreamReader, gtid
-from pymysqlreplication.event import GtidEvent, RotateEvent, MariadbGtidEvent, QueryEvent
+from pymysqlreplication.event import GtidEvent, RotateEvent, MariadbGtidEvent, QueryEvent,MariadbAnnotateRowsEvent
 from pymysqlreplication.row_event import WriteRowsEvent, UpdateRowsEvent, DeleteRowsEvent
 
 MARIADB_SETTINGS = {
@@ -65,10 +65,12 @@ def query_server_id(self):
             RotateEvent,
             WriteRowsEvent,
             UpdateRowsEvent,
-            DeleteRowsEvent
+            DeleteRowsEvent,
+            MariadbAnnotateRowsEvent
         ],
         auto_position=gtid,
-        is_mariadb=True
+        is_mariadb=True,
+        annotate_rows_event=True
     )
 
     print('Starting reading events from GTID ', gtid)

pymysqlreplication/binlogstream.py

@@ -14,8 +14,8 @@
     QueryEvent, RotateEvent, FormatDescriptionEvent,
     XidEvent, GtidEvent, StopEvent, XAPrepareEvent,
     BeginLoadQueryEvent, ExecuteLoadQueryEvent,
-    HeartbeatLogEvent, NotImplementedEvent, 
-    MariadbGtidEvent, RandEvent)
+    HeartbeatLogEvent, NotImplementedEvent, MariadbGtidEvent,
+    MariadbAnnotateRowsEvent, RandEvent)
 from .exceptions import BinLogNotEnabled
 from .row_event import (
     UpdateRowsEvent, WriteRowsEvent, DeleteRowsEvent, TableMapEvent)
@@ -142,6 +142,7 @@ def __init__(self, connection_settings, server_id,
                  fail_on_table_metadata_unavailable=False,
                  slave_heartbeat=None,
                  is_mariadb=False,
+                 annotate_rows_event=False,
                  ignore_decode_errors=False):
         """
         Attributes:
@@ -167,7 +168,8 @@ def __init__(self, connection_settings, server_id,
             skip_to_timestamp: Ignore all events until reaching specified
                                timestamp.
             report_slave: Report slave in SHOW SLAVE HOSTS.
-            slave_uuid: Report slave_uuid in SHOW SLAVE HOSTS.
+            slave_uuid: Report slave_uuid or replica_uuid in SHOW SLAVE HOSTS(MySQL 8.0.21-) or
+                        SHOW REPLICAS(MySQL 8.0.22+) depends on your MySQL version.
             fail_on_table_metadata_unavailable: Should raise exception if we
                                                 can't get table information on
                                                 row_events
@@ -179,6 +181,8 @@ def __init__(self, connection_settings, server_id,
                              for semantics
             is_mariadb: Flag to indicate it's a MariaDB server, used with auto_position
                     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 
                                   when reading column data will be ignored.
         """
@@ -220,6 +224,7 @@ def __init__(self, connection_settings, server_id,
         self.auto_position = auto_position
         self.skip_to_timestamp = skip_to_timestamp
         self.is_mariadb = is_mariadb
+        self.__annotate_rows_event = annotate_rows_event
 
         if end_log_pos:
             self.is_past_end_log_pos = False
@@ -302,7 +307,7 @@ def __connect_to_stream(self):
 
         if self.slave_uuid:
             cur = self._stream_connection.cursor()
-            cur.execute("set @slave_uuid= '%s'" % self.slave_uuid)
+            cur.execute("SET @slave_uuid = %s, @replica_uuid = %s", (self.slave_uuid, self.slave_uuid))
             cur.close()
 
         if self.slave_heartbeat:
@@ -332,67 +337,39 @@ def __connect_to_stream(self):
         self._register_slave()
 
         if not self.auto_position:
-            # only when log_file and log_pos both provided, the position info is
-            # valid, if not, get the current position from master
-            if self.log_file is None or self.log_pos is None:
-                cur = self._stream_connection.cursor()
-                cur.execute("SHOW MASTER STATUS")
-                master_status = cur.fetchone()
-                if master_status is None:
-                    raise BinLogNotEnabled()
-                self.log_file, self.log_pos = master_status[:2]
-                cur.close()
-
-            prelude = struct.pack('<i', len(self.log_file) + 11) \
-                + bytes(bytearray([COM_BINLOG_DUMP]))
-
-            if self.__resume_stream:
-                prelude += struct.pack('<I', self.log_pos)
-            else:
-                prelude += struct.pack('<I', 4)
-
-            flags = 0
-            if not self.__blocking:
-                flags |= 0x01  # BINLOG_DUMP_NON_BLOCK
-            prelude += struct.pack('<H', flags)
-
-            prelude += struct.pack('<I', self.__server_id)
-            prelude += self.log_file.encode()
-        else:
             if self.is_mariadb:
-                # https://mariadb.com/kb/en/5-slave-registration/
-                cur = self._stream_connection.cursor()
-                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()
-
-                # 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
-                )
-
-                prelude = struct.pack('<i', header_size) + bytes(bytearray([COM_BINLOG_DUMP]))
-
-                # binlog pos
-                prelude += struct.pack('<i', 4)
+                prelude = self.__set_mariadb_settings()
+            else:
+                # only when log_file and log_pos both provided, the position info is
+                # valid, if not, get the current position from master
+                if self.log_file is None or self.log_pos is None:
+                    cur = self._stream_connection.cursor()
+                    cur.execute("SHOW MASTER STATUS")
+                    master_status = cur.fetchone()
+                    if master_status is None:
+                        raise BinLogNotEnabled()
+                    self.log_file, self.log_pos = master_status[:2]
+                    cur.close()
+
+                prelude = struct.pack('<i', len(self.log_file) + 11) \
+                    + bytes(bytearray([COM_BINLOG_DUMP]))
+
+                if self.__resume_stream:
+                    prelude += struct.pack('<I', self.log_pos)
+                else:
+                    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'
-                
+                prelude += self.log_file.encode()
+        else:
+            if self.is_mariadb:
+                prelude = self.__set_mariadb_settings()        
             else:
                 # Format for mysql packet master_auto_position
                 #
@@ -474,6 +451,48 @@ def __connect_to_stream(self):
             self._stream_connection._next_seq_id = 1
         self.__connected_stream = True
 
+    def __set_mariadb_settings(self):
+        # https://mariadb.com/kb/en/5-slave-registration/
+        cur = self._stream_connection.cursor()
+        if self.auto_position != None : 
+            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()
+
+        # 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
+        )
+
+        prelude = struct.pack('<i', header_size) + bytes(bytearray([COM_BINLOG_DUMP]))
+
+        # binlog pos
+        prelude += struct.pack('<i', 4)
+
+        flags = 0
+
+        # Enable annotate rows event 
+        if self.__annotate_rows_event:
+            flags |= 0x02 # BINLOG_SEND_ANNOTATE_ROWS_EVENT
+
+        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'
+
+        return prelude
+
     def fetchone(self):
         while True:
             if self.end_log_pos and self.is_past_end_log_pos:
@@ -602,6 +621,7 @@ def _allowed_event_list(self, only_events, ignored_events,
                 HeartbeatLogEvent,
                 NotImplementedEvent,
                 MariadbGtidEvent,
+                MariadbAnnotateRowsEvent,
                 RandEvent
                 ))
         if ignored_events is not None:

pymysqlreplication/bitmap.py

@@ -19,7 +19,7 @@
     4, 5, 5, 6, 5, 6, 6, 7, 5, 6, 6, 7, 6, 7, 7, 8,
 ]
 
-# Calculate totol bit counts in a bitmap
+# Calculate total bit counts in a bitmap
 def BitCount(bitmap):
     n = 0
     for i in range(0, len(bitmap)):