Merge branch 'main' of https://github.com/23-OSSCA-python-mysql-replication/python-mysql-replication into feature/typing-packet-gtid-binlogstream

Author: starcat37

README.md

@@ -63,6 +63,7 @@ Featured
 
 [Streaming Changes in a Database with Amazon Kinesis](https://aws.amazon.com/blogs/database/streaming-changes-in-a-database-with-amazon-kinesis/) (by Emmanuel Espina, Amazon Web Services)
 
+[Near Zero Downtime Migration from MySQL to DynamoDB](https://aws.amazon.com/ko/blogs/big-data/near-zero-downtime-migration-from-mysql-to-dynamodb/) (by YongSeong Lee, Amazon Web Services)
 
 Projects using this library
 ===========================

pymysqlreplication/binlogstream.py

@@ -16,7 +16,8 @@
     BeginLoadQueryEvent, ExecuteLoadQueryEvent,
     HeartbeatLogEvent, NotImplementedEvent, MariadbGtidEvent,
     MariadbAnnotateRowsEvent, RandEvent, MariadbStartEncryptionEvent, RowsQueryLogEvent,
-    MariadbGtidListEvent, MariadbBinLogCheckPointEvent)
+    MariadbGtidListEvent, MariadbBinLogCheckPointEvent, UserVarEvent,
+    PreviousGtidsEvent)
 from .exceptions import BinLogNotEnabled
 from .gtid import GtidSet
 from .packet import BinLogPacketWrapper
@@ -148,7 +149,8 @@ def __init__(self, connection_settings: Dict, server_id: int,
                  slave_heartbeat: Optional[float] = None,
                  is_mariadb: bool = False,
                  annotate_rows_event: bool = False,
-                 ignore_decode_errors: bool = False) -> None:
+                 ignore_decode_errors: bool = False,
+                 verify_checksum = False) -> None:
         """
         Attributes:
             ctl_connection_settings[Dict]: Connection settings for cluster holding
@@ -188,6 +190,7 @@ def __init__(self, connection_settings: Dict, server_id: int,
                     used with 'is_mariadb'
             ignore_decode_errors[bool]: If true, any decode errors encountered
                                   when reading column data will be ignored.
+            verify_checksum[bool]: If true, verify events read from the binary log by examining checksums.
         """
 
         self.__connection_settings: Dict = connection_settings
@@ -210,6 +213,7 @@ def __init__(self, connection_settings: Dict, server_id: int,
             only_events, ignored_events, filter_non_implemented_events)
         self.__fail_on_table_metadata_unavailable: bool = fail_on_table_metadata_unavailable
         self.__ignore_decode_errors: bool = ignore_decode_errors
+        self.__verify_checksum: bool = verify_checksum
 
         # We can't filter on packet level TABLE_MAP and rotate event because
         # we need them for handling other operations
@@ -541,7 +545,8 @@ def fetchone(self) -> Union[BinLogPacketWrapper, None]:
                                                self.__ignored_schemas,
                                                self.__freeze_schema,
                                                self.__fail_on_table_metadata_unavailable,
-                                               self.__ignore_decode_errors)
+                                               self.__ignore_decode_errors,
+                                               self.__verify_checksum,)
 
             if binlog_event.event_type == ROTATE_EVENT:
                 self.log_pos = binlog_event.event.position
@@ -632,7 +637,9 @@ def _allowed_event_list(self, only_events: Optional[List[str]], ignored_events:
                 RandEvent,
                 MariadbStartEncryptionEvent,
                 MariadbGtidListEvent,
-                MariadbBinLogCheckPointEvent
+                MariadbBinLogCheckPointEvent,
+                UserVarEvent,
+                PreviousGtidsEvent
             ))
         if ignored_events is not None:
             for e in ignored_events:

pymysqlreplication/event.py

@@ -65,13 +65,15 @@ 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,
         constants.HEARTBEAT_LOG_EVENT: event.HeartbeatLogEvent,
         constants.XA_PREPARE_EVENT: event.XAPrepareEvent,
         constants.ROWS_QUERY_LOG_EVENT: event.RowsQueryLogEvent,
         constants.RAND_EVENT: event.RandEvent,
+        constants.USER_VAR_EVENT: event.UserVarEvent,
         # row_event
         constants.UPDATE_ROWS_EVENT_V1: row_event.UpdateRowsEvent,
         constants.WRITE_ROWS_EVENT_V1: row_event.WriteRowsEvent,
@@ -83,11 +85,8 @@ 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.NotImplementedEvent,
         constants.MARIADB_BINLOG_CHECKPOINT_EVENT: event.MariadbBinLogCheckPointEvent,
         constants.MARIADB_GTID_EVENT: event.MariadbGtidEvent,
         constants.MARIADB_GTID_GTID_LIST_EVENT: event.MariadbGtidListEvent,
@@ -105,7 +104,8 @@ def __init__(self, from_packet, table_map,
                  ignored_schemas,
                  freeze_schema,
                  fail_on_table_metadata_unavailable,
-                 ignore_decode_errors):
+                 ignore_decode_errors,
+                 verify_checksum,):
         # -1 because we ignore the ok byte
         self.read_bytes = 0
         # Used when we want to override a value in the data buffer
@@ -135,6 +135,7 @@ def __init__(self, from_packet, table_map,
         if use_checksum:
             event_size_without_header = self.event_size - 23
         else:
+            verify_checksum = False
             event_size_without_header = self.event_size - 19
 
         self.event = None
@@ -151,7 +152,8 @@ def __init__(self, from_packet, table_map,
                                  ignored_schemas=ignored_schemas,
                                  freeze_schema=freeze_schema,
                                  fail_on_table_metadata_unavailable=fail_on_table_metadata_unavailable,
-                                 ignore_decode_errors=ignore_decode_errors)
+                                 ignore_decode_errors=ignore_decode_errors,
+                                 verify_checksum=verify_checksum)
         if self.event._processed == False:
             self.event = None
 

pymysqlreplication/packet.py

@@ -18,15 +18,15 @@ class PyMySQLReplicationTestCase(base):
     def ignoredEvents(self):
         return []
 
-    def setUp(self):
+    def setUp(self, charset="utf8"):
         # default
         self.database = {
             "host": os.environ.get("MYSQL_5_7") or "localhost",
             "user": "root",
             "passwd": "",
             "port": 3306,
             "use_unicode": True,
-            "charset": "utf8",
+            "charset": charset,
             "db": "pymysqlreplication_test"
         }
 

pymysqlreplication/tests/base.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)