Merge pull request #90 from noplay/speedup

Speedup event filtering

Author: julien-duponchelle

pymysqlreplication/binlogstream.py

@@ -9,8 +9,9 @@
 
 from .packet import BinLogPacketWrapper
 from .constants.BINLOG import TABLE_MAP_EVENT, ROTATE_EVENT
-from .event import NotImplementedEvent
 from .gtid import GtidSet
+from .event import QueryEvent, RotateEvent, FormatDescriptionEvent, XidEvent, GtidEvent, NotImplementedEvent
+from .row_event import UpdateRowsEvent, WriteRowsEvent, DeleteRowsEvent, TableMapEvent
 
 try:
     from pymysql.constants.COMMAND import COM_BINLOG_DUMP_GTID
@@ -48,9 +49,12 @@ def __init__(self, connection_settings, server_id, resume_stream=False,
         self.__connected_ctl = False
         self.__resume_stream = resume_stream
         self.__blocking = blocking
-        self.__only_events = only_events
-        self.__ignored_events = ignored_events
-        self.__filter_non_implemented_events = filter_non_implemented_events
+        self.__allowed_events = self._allowed_event_list(only_events, ignored_events, filter_non_implemented_events)
+
+        # We can't filter on packet level TABLE_MAP and rotate event because we need
+        # them for handling other operations
+        self.__allowed_events_in_packet = frozenset([TableMapEvent, RotateEvent]).union(self.__allowed_events)
+
         self.__server_id = server_id
         self.__use_checksum = False
 
@@ -156,7 +160,7 @@ def __connect_to_stream(self):
             # 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
@@ -228,7 +232,8 @@ def fetchone(self):
 
             binlog_event = BinLogPacketWrapper(pkt, self.table_map,
                                                self._ctl_connection,
-                                               self.__use_checksum)
+                                               self.__use_checksum,
+                                               self.__allowed_events_in_packet)
             if binlog_event.event_type == TABLE_MAP_EVENT:
                 self.table_map[binlog_event.event.table_id] = \
                     binlog_event.event.get_table()
@@ -242,35 +247,44 @@ def fetchone(self):
                 # wrong table schema.
                 # The fix is to rely on the fact that MySQL will also rotate to a new binlog file every time it
                 # restarts. That means every rotation we see *could* be a sign of restart and so potentially
-                # invalidates all our cached table id to schema mappings. This means we have to load them all 
+                # invalidates all our cached table id to schema mappings. This means we have to load them all
                 # again for each logfile which is potentially wasted effort but we can't really do much better
                 # without being broken in restart case
                 self.table_map = {}
             elif binlog_event.log_pos:
                 self.log_pos = binlog_event.log_pos
 
-            if self.__filter_event(binlog_event.event):
+            # event is none if we have filter it on packet level
+            # we filter also not allowed events
+            if binlog_event.event is None or (binlog_event.event.__class__ not in self.__allowed_events):
                 continue
 
             return binlog_event.event
 
-    def __filter_event(self, event):
-        if self.__filter_non_implemented_events and isinstance(event, NotImplementedEvent):
-            return True
-
-        if self.__ignored_events is not None:
-            for ignored_event in self.__ignored_events:
-                if isinstance(event, ignored_event):
-                    return True
-
-        if self.__only_events is not None:
-            for allowed_event in self.__only_events:
-                if isinstance(event, allowed_event):
-                    return False
-            else:
-                return True
-
-        return False
+    def _allowed_event_list(self, only_events, ignored_events, filter_non_implemented_events):
+        if only_events is not None:
+            events = set(only_events)
+        else:
+            events = set((
+                QueryEvent,
+                RotateEvent,
+                FormatDescriptionEvent,
+                XidEvent,
+                GtidEvent,
+                UpdateRowsEvent,
+                WriteRowsEvent,
+                DeleteRowsEvent,
+                TableMapEvent,
+                NotImplementedEvent))
+        if ignored_events is not None:
+            for e in ignored_events:
+                events.remove(e)
+        if filter_non_implemented_events:
+            try:
+                events.remove(NotImplementedEvent)
+            except KeyError:
+                pass
+        return frozenset(events)
 
     def __get_table_information(self, schema, table):
         for i in range(1, 3):

pymysqlreplication/packet.py

@@ -48,7 +48,7 @@ class BinLogPacketWrapper(object):
 
     }
 
-    def __init__(self, from_packet, table_map, ctl_connection, use_checksum):
+    def __init__(self, from_packet, table_map, ctl_connection, use_checksum, allowed_events = None):
         if not from_packet.is_ok_packet():
             raise ValueError(
                 "Cannot create %s object from invalid packet type" %
@@ -79,8 +79,11 @@ def __init__(self, from_packet, table_map, ctl_connection, use_checksum):
         else:
             event_size_without_header = self.event_size - 19
 
-        event_class = self.__event_map.get(self.event_type,
-            event.NotImplementedEvent)
+        self.event = None
+        event_class = self.__event_map.get(self.event_type, event.NotImplementedEvent)
+
+        if event_class not in allowed_events:
+            return
         self.event = event_class(self, event_size_without_header, table_map,
                                  ctl_connection)
 

pymysqlreplication/tests/benchmark.py

@@ -58,5 +58,6 @@ def consume_events():
     while True:
         execute(conn, "UPDATE test SET i = i + 1;")
 else:
-    cProfile.run('consume_events()')
+    consume_events()
+    #cProfile.run('consume_events()')
 

pymysqlreplication/tests/test_basic.py

@@ -13,6 +13,12 @@ class TestBasicBinLogStreamReader(base.PyMySQLReplicationTestCase):
     def ignoredEvents(self):
         return [GtidEvent]
 
+    def test_allowed_event_list(self):
+        self.assertEqual(len(self.stream._allowed_event_list(None, None, False)), 10)
+        self.assertEqual(len(self.stream._allowed_event_list(None, None, True)), 9)
+        self.assertEqual(len(self.stream._allowed_event_list(None, [RotateEvent], False)), 9)
+        self.assertEqual(len(self.stream._allowed_event_list([RotateEvent], None, False)), 1)
+
     def test_read_query_event(self):
         query = "CREATE TABLE test (id INT NOT NULL AUTO_INCREMENT, data VARCHAR (50) NOT NULL, PRIMARY KEY (id))"
         self.execute(query)