Merge remote-tracking branch 'origin/main'

Author: soulee-dev

Diff for: pymysqlreplication/binlogstream.py

@@ -1,22 +1,22 @@
 # -*- coding: utf-8 -*-
 
-import pymysql
 import struct
 from distutils.version import LooseVersion
 
+import pymysql
 from pymysql.constants.COMMAND import COM_BINLOG_DUMP, COM_REGISTER_SLAVE
 from pymysql.cursors import DictCursor
 
-from .packet import BinLogPacketWrapper
 from .constants.BINLOG import TABLE_MAP_EVENT, ROTATE_EVENT, FORMAT_DESCRIPTION_EVENT
-from .gtid import GtidSet
 from .event import (
     QueryEvent, RotateEvent, FormatDescriptionEvent,
     XidEvent, GtidEvent, StopEvent, XAPrepareEvent,
     BeginLoadQueryEvent, ExecuteLoadQueryEvent,
     HeartbeatLogEvent, NotImplementedEvent, MariadbGtidEvent,
-    MariadbAnnotateRowsEvent, RandEvent, MariadbStartEncryptionEvent)
+    MariadbAnnotateRowsEvent, RandEvent, MariadbStartEncryptionEvent, RowsQueryLogEvent)
 from .exceptions import BinLogNotEnabled
+from .gtid import GtidSet
+from .packet import BinLogPacketWrapper
 from .row_event import (
     UpdateRowsEvent, WriteRowsEvent, DeleteRowsEvent, TableMapEvent)
 
@@ -33,7 +33,6 @@
 
 
 class ReportSlave(object):
-
     """Represent the values that you may report when connecting as a slave
     to a master. SHOW SLAVE HOSTS related"""
 
@@ -68,7 +67,7 @@ def __init__(self, value):
             self.hostname = value
 
     def __repr__(self):
-        return '<ReportSlave hostname=%s username=%s password=%s port=%d>' %\
+        return '<ReportSlave hostname=%s username=%s password=%s port=%d>' % \
             (self.hostname, self.username, self.password, self.port)
 
     def encoded(self, server_id, master_id=0):
@@ -123,7 +122,6 @@ def encoded(self, server_id, master_id=0):
 
 
 class BinLogStreamReader(object):
-
     """Connect to replication stream and read event
     """
     report_slave = None
@@ -256,6 +254,7 @@ def __connect_to_ctl(self):
             self._ctl_connection_settings = dict(self.__connection_settings)
         self._ctl_connection_settings["db"] = "information_schema"
         self._ctl_connection_settings["cursorclass"] = DictCursor
+        self._ctl_connection_settings["autocommit"] = True
         self._ctl_connection = self.pymysql_wrapper(**self._ctl_connection_settings)
         self._ctl_connection._get_table_information = self.__get_table_information
         self.__connected_ctl = True
@@ -316,7 +315,7 @@ def __connect_to_stream(self):
                                                                4294967))
             # If heartbeat is too low, the connection will disconnect before,
             # this is also the behavior in mysql
-            heartbeat = float(min(net_timeout/2., self.slave_heartbeat))
+            heartbeat = float(min(net_timeout / 2., self.slave_heartbeat))
             if heartbeat > 4294967:
                 heartbeat = 4294967
 
@@ -352,7 +351,7 @@ def __connect_to_stream(self):
                     cur.close()
 
                 prelude = struct.pack('<i', len(self.log_file) + 11) \
-                    + bytes(bytearray([COM_BINLOG_DUMP]))
+                          + bytes(bytearray([COM_BINLOG_DUMP]))
 
                 if self.__resume_stream:
                     prelude += struct.pack('<I', self.log_pos)
@@ -369,7 +368,7 @@ def __connect_to_stream(self):
                 prelude += self.log_file.encode()
         else:
             if self.is_mariadb:
-                prelude = self.__set_mariadb_settings()        
+                prelude = self.__set_mariadb_settings()
             else:
                 # Format for mysql packet master_auto_position
                 #
@@ -416,8 +415,8 @@ def __connect_to_stream(self):
                                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]))
+                prelude = b'' + struct.pack('<i', header_size + encoded_data_size) \
+                          + bytes(bytearray([COM_BINLOG_DUMP_GTID]))
 
                 flags = 0
                 if not self.__blocking:
@@ -454,7 +453,7 @@ def __connect_to_stream(self):
     def __set_mariadb_settings(self):
         # https://mariadb.com/kb/en/5-slave-registration/
         cur = self._stream_connection.cursor()
-        if self.auto_position != None : 
+        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")
@@ -465,7 +464,7 @@ def __set_mariadb_settings(self):
                 4 +  # binlog pos
                 2 +  # binlog flags
                 4 +  # slave server_id,
-                4    # requested binlog file name , set it to empty
+                4  # requested binlog file name , set it to empty
         )
 
         prelude = struct.pack('<i', header_size) + bytes(bytearray([COM_BINLOG_DUMP]))
@@ -477,11 +476,11 @@ def __set_mariadb_settings(self):
 
         # Enable annotate rows event 
         if self.__annotate_rows_event:
-            flags |= 0x02 # BINLOG_SEND_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)
 
@@ -621,10 +620,11 @@ def _allowed_event_list(self, only_events, ignored_events,
                 HeartbeatLogEvent,
                 NotImplementedEvent,
                 MariadbGtidEvent,
+                RowsQueryLogEvent,
                 MariadbAnnotateRowsEvent,
                 RandEvent,
-                MariadbStartEncryptionEvent
-                ))
+                MariadbStartEncryptionEvent,
+            ))
         if ignored_events is not None:
             for e in ignored_events:
                 events.remove(e)
@@ -653,8 +653,10 @@ def __get_table_information(self, schema, table):
                         table_schema = %s AND table_name = %s
                     ORDER BY ORDINAL_POSITION
                     """, (schema, table))
+                result = cur.fetchall()
+                cur.close()
 
-                return cur.fetchall()
+                return result
             except pymysql.OperationalError as error:
                 code, message = error.args
                 if code in MYSQL_EXPECTED_ERROR_CODES:

Diff for: pymysqlreplication/event.py

@@ -56,7 +56,7 @@ class GtidEvent(BinLogEvent):
     """GTID change in binlog event
     """
     def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs):
-        super(GtidEvent, self).__init__(from_packet, event_size, table_map,
+        super().__init__(from_packet, event_size, table_map,
                                           ctl_connection, **kwargs)
 
         self.commit_flag = struct.unpack("!B", self.packet.read(1))[0] == 1
@@ -97,7 +97,7 @@ class MariadbGtidEvent(BinLogEvent):
     """
     def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs):
 
-        super(MariadbGtidEvent, self).__init__(from_packet, event_size, table_map, ctl_connection, **kwargs)
+        super().__init__(from_packet, event_size, table_map, ctl_connection, **kwargs)
 
         self.server_id = self.packet.server_id
         self.gtid_seq_no = self.packet.read_uint64()
@@ -106,7 +106,7 @@ def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs)
         self.gtid = "%d-%d-%d" % (self.domain_id, self.server_id, self.gtid_seq_no)
 
     def _dump(self):
-        super(MariadbGtidEvent, self)._dump()
+        super()._dump()
         print("Flags:", self.flags)
         print('GTID:', self.gtid)
 
@@ -121,11 +121,11 @@ class MariadbAnnotateRowsEvent(BinLogEvent):
         sql_statement: The SQL statement
     """
     def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs):
-        super(MariadbAnnotateRowsEvent, self).__init__(from_packet, event_size, table_map, ctl_connection, **kwargs)
+        super().__init__(from_packet, event_size, table_map, ctl_connection, **kwargs)
         self.sql_statement = self.packet.read(event_size)
 
     def _dump(self):
-        super(MariadbAnnotateRowsEvent, self)._dump()
+        super()._dump()
         print("SQL statement :", self.sql_statement)   
 
 
@@ -137,7 +137,7 @@ class RotateEvent(BinLogEvent):
         next_binlog: Name of next binlog file
     """
     def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs):
-        super(RotateEvent, self).__init__(from_packet, event_size, table_map,
+        super().__init__(from_packet, event_size, table_map,
                                           ctl_connection, **kwargs)
         self.position = struct.unpack('<Q', self.packet.read(8))[0]
         self.next_binlog = self.packet.read(event_size - 8).decode()
@@ -158,7 +158,7 @@ class XAPrepareEvent(BinLogEvent):
         xid: serialized XID representation of XA transaction
     """
     def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs):
-        super(XAPrepareEvent, self).__init__(from_packet, event_size, table_map,
+        super().__init__(from_packet, event_size, table_map,
                                           ctl_connection, **kwargs)
 
         # one_phase is True: XA COMMIT ... ONE PHASE
@@ -182,7 +182,7 @@ def _dump(self):
 
 class FormatDescriptionEvent(BinLogEvent):
     def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs):
-        super(FormatDescriptionEvent, self).__init__(from_packet, event_size, table_map,
+        super().__init__(from_packet, event_size, table_map,
                                           ctl_connection, **kwargs)
         self.binlog_version = struct.unpack('<H', self.packet.read(2))
         self.mysql_version_str = self.packet.read(50).rstrip(b'\0').decode()
@@ -206,12 +206,12 @@ class XidEvent(BinLogEvent):
     """
 
     def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs):
-        super(XidEvent, self).__init__(from_packet, event_size, table_map,
+        super().__init__(from_packet, event_size, table_map,
                                        ctl_connection, **kwargs)
         self.xid = struct.unpack('<Q', self.packet.read(8))[0]
 
     def _dump(self):
-        super(XidEvent, self)._dump()
+        super()._dump()
         print("Transaction ID: %d" % (self.xid))
 
 
@@ -236,21 +236,21 @@ class HeartbeatLogEvent(BinLogEvent):
     """
 
     def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs):
-        super(HeartbeatLogEvent, self).__init__(from_packet, event_size,
+        super().__init__(from_packet, event_size,
                                                 table_map, ctl_connection,
                                                 **kwargs)
         self.ident = self.packet.read(event_size).decode()
 
     def _dump(self):
-        super(HeartbeatLogEvent, self)._dump()
+        super()._dump()
         print("Current binlog: %s" % (self.ident))
 
 
 class QueryEvent(BinLogEvent):
     '''This event is trigger when a query is run of the database.
     Only replicated queries are logged.'''
     def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs):
-        super(QueryEvent, self).__init__(from_packet, event_size, table_map,
+        super().__init__(from_packet, event_size, table_map,
                                          ctl_connection, **kwargs)
 
         # Post-header
@@ -276,7 +276,7 @@ def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs)
         #string[EOF]    query
 
     def _dump(self):
-        super(QueryEvent, self)._dump()
+        super()._dump()
         print("Schema: %s" % (self.schema))
         print("Execution time: %d" % (self.execution_time))
         print("Query: %s" % (self.query))
@@ -376,15 +376,15 @@ class BeginLoadQueryEvent(BinLogEvent):
         block-data
     """
     def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs):
-        super(BeginLoadQueryEvent, self).__init__(from_packet, event_size, table_map,
+        super().__init__(from_packet, event_size, table_map,
                                                      ctl_connection, **kwargs)
 
         # Payload
         self.file_id = self.packet.read_uint32()
         self.block_data = self.packet.read(event_size - 4)
 
     def _dump(self):
-        super(BeginLoadQueryEvent, self)._dump()
+        super()._dump()
         print("File id: %d" % (self.file_id))
         print("Block data: %s" % (self.block_data))
 
@@ -405,7 +405,7 @@ class ExecuteLoadQueryEvent(BinLogEvent):
         dup_handling_flags
     """
     def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs):
-        super(ExecuteLoadQueryEvent, self).__init__(from_packet, event_size, table_map,
+        super().__init__(from_packet, event_size, table_map,
                                                         ctl_connection, **kwargs)
 
         # Post-header
@@ -442,18 +442,19 @@ class IntvarEvent(BinLogEvent):
         value
     """
     def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs):
-        super(IntvarEvent, self).__init__(from_packet, event_size, table_map,
+        super().__init__(from_packet, event_size, table_map,
                                           ctl_connection, **kwargs)
 
         # Payload
         self.type = self.packet.read_uint8()
         self.value = self.packet.read_uint32()
 
     def _dump(self):
-        super(IntvarEvent, self)._dump()
+        super()._dump()
         print("type: %d" % (self.type))
         print("Value: %d" % (self.value))
 
+
 class RandEvent(BinLogEvent):
     """
     RandEvent is generated every time a statement uses the RAND() function.
@@ -467,7 +468,7 @@ class RandEvent(BinLogEvent):
     """
 
     def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs):
-        super(RandEvent, self).__init__(from_packet, event_size, table_map,
+        super().__init__(from_packet, event_size, table_map,
                                         ctl_connection, **kwargs)
         # Payload
         self._seed1 = self.packet.read_uint64()
@@ -484,10 +485,11 @@ def seed2(self):
         return self._seed2
 
     def _dump(self):
-        super(RandEvent, self)._dump()
+        super()._dump()
         print("seed1: %d" % (self.seed1))
         print("seed2: %d" % (self.seed2))
 
+
 class MariadbStartEncryptionEvent(BinLogEvent):
     """
     Since MariaDB 10.1.7, 
@@ -505,7 +507,7 @@ class MariadbStartEncryptionEvent(BinLogEvent):
     """
 
     def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs):
-        super(MariadbStartEncryptionEvent, self).__init__(from_packet, event_size, table_map, ctl_connection, **kwargs)
+        super().__init__(from_packet, event_size, table_map, ctl_connection, **kwargs)
 
         self.schema = self.packet.read_uint8()
         self.key_version = self.packet.read_uint32()
@@ -517,8 +519,29 @@ def _dump(self):
         print(f"Nonce: {self.nonce}")
 
 
+class RowsQueryLogEvent(BinLogEvent):
+    """
+    Record original query for the row events in Row-Based Replication
+
+    More details are available in the MySQL Knowledge Base:
+    https://dev.mysql.com/doc/dev/mysql-server/latest/classRows__query__log__event.html
+
+    :ivar query_length: uint - Length of the SQL statement
+    :ivar query: str - The executed SQL statement
+    """
+    def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs):
+        super(RowsQueryLogEvent, self).__init__(from_packet, event_size, table_map,
+                                          ctl_connection, **kwargs)
+        self.query_length = self.packet.read_uint8()
+        self.query = self.packet.read(self.query_length).decode('utf-8')
+    def dump(self):
+        print("=== %s ===" % (self.__class__.__name__))
+        print("Query length: %d" % self.query_length)
+        print("Query: %s" % self.query)
+
+
 class NotImplementedEvent(BinLogEvent):
     def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs):
-        super(NotImplementedEvent, self).__init__(
+        super().__init__(
             from_packet, event_size, table_map, ctl_connection, **kwargs)
         self.packet.advance(event_size)

Diff for: pymysqlreplication/packet.py

@@ -70,6 +70,7 @@ class BinLogPacketWrapper(object):
         constants.EXECUTE_LOAD_QUERY_EVENT: event.ExecuteLoadQueryEvent,
         constants.HEARTBEAT_LOG_EVENT: event.HeartbeatLogEvent,
         constants.XA_PREPARE_EVENT: event.XAPrepareEvent,
+        constants.ROWS_QUERY_LOG_EVENT: event.RowsQueryLogEvent,
         constants.RAND_EVENT: event.RandEvent,
         # row_event
         constants.UPDATE_ROWS_EVENT_V1: row_event.UpdateRowsEvent,
@@ -82,6 +83,7 @@ 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,