Filters events by table names

Author: julien-duponchelle

pymysqlreplication/binlogstream.py

@@ -30,7 +30,8 @@ class BinLogStreamReader(object):
     def __init__(self, connection_settings, server_id, resume_stream=False,
                  blocking=False, only_events=None, log_file=None, log_pos=None,
                  filter_non_implemented_events=True,
-                 ignored_events=None, auto_position=None):
+                 ignored_events=None, auto_position=None,
+                 only_tables = None):
         """
         Attributes:
             resume_stream: Start for event from position or the latest event of
@@ -41,6 +42,7 @@ def __init__(self, connection_settings, server_id, resume_stream=False,
             log_file: Set replication start log file
             log_pos: Set replication start log pos
             auto_position: Use master_auto_position gtid to set position
+            only_tables: An array with the tables you want to watch
         """
         self.__connection_settings = connection_settings
         self.__connection_settings["charset"] = "utf8"
@@ -49,6 +51,8 @@ def __init__(self, connection_settings, server_id, resume_stream=False,
         self.__connected_ctl = False
         self.__resume_stream = resume_stream
         self.__blocking = blocking
+
+        self.__only_tables = only_tables
         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
@@ -233,7 +237,8 @@ def fetchone(self):
             binlog_event = BinLogPacketWrapper(pkt, self.table_map,
                                                self._ctl_connection,
                                                self.__use_checksum,
-                                               self.__allowed_events_in_packet)
+                                               allowed_events = self.__allowed_events_in_packet,
+                                               only_tables = self.__only_tables)
             if binlog_event.event_type == TABLE_MAP_EVENT:
                 self.table_map[binlog_event.event.table_id] = \
                     binlog_event.event.get_table()

pymysqlreplication/event.py

@@ -7,13 +7,16 @@
 
 
 class BinLogEvent(object):
-    def __init__(self, from_packet, event_size, table_map, ctl_connection):
+    def __init__(self, from_packet, event_size, table_map, ctl_connection, only_tables = None):
         self.packet = from_packet
         self.table_map = table_map
         self.event_type = self.packet.event_type
         self.timestamp = self.packet.timestamp
         self.event_size = event_size
         self._ctl_connection = ctl_connection
+        # The event have been fully processed, if processed is false
+        # the event will be skipped
+        self._processed = True
 
     def _read_table_id(self):
         # Table ID is 6 byte
@@ -39,9 +42,9 @@ def _dump(self):
 class GtidEvent(BinLogEvent):
     """GTID change in binlog event
     """
-    def __init__(self, from_packet, event_size, table_map, ctl_connection):
+    def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs):
         super(GtidEvent, self).__init__(from_packet, event_size, table_map,
-                                          ctl_connection)
+                                          ctl_connection, **kwargs)
 
         self.commit_flag = byte2int(self.packet.read(1)) == 1
         self.sid = self.packet.read(16)
@@ -72,9 +75,9 @@ class RotateEvent(BinLogEvent):
         position: Position inside next binlog
         next_binlog: Name of next binlog file
     """
-    def __init__(self, from_packet, event_size, table_map, ctl_connection):
+    def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs):
         super(RotateEvent, self).__init__(from_packet, event_size, table_map,
-                                          ctl_connection)
+                                          ctl_connection, **kwargs)
         self.position = struct.unpack('<Q', self.packet.read(8))[0]
         self.next_binlog = self.packet.read(event_size - 8).decode()
 
@@ -96,9 +99,9 @@ class XidEvent(BinLogEvent):
         xid: Transaction ID for 2PC
     """
 
-    def __init__(self, from_packet, event_size, table_map, ctl_connection):
+    def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs):
         super(XidEvent, self).__init__(from_packet, event_size, table_map,
-                                       ctl_connection)
+                                       ctl_connection, **kwargs)
         self.xid = struct.unpack('<Q', self.packet.read(8))[0]
 
     def _dump(self):
@@ -109,9 +112,9 @@ def _dump(self):
 class QueryEvent(BinLogEvent):
     '''This evenement 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):
+    def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs):
         super(QueryEvent, self).__init__(from_packet, event_size, table_map,
-                                         ctl_connection)
+                                         ctl_connection, **kwargs)
 
         # Post-header
         self.slave_proxy_id = self.packet.read_uint32()
@@ -137,7 +140,7 @@ def _dump(self):
 
 
 class NotImplementedEvent(BinLogEvent):
-    def __init__(self, from_packet, event_size, table_map, ctl_connection):
+    def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs):
         super(NotImplementedEvent, self).__init__(
-            from_packet, event_size, table_map, ctl_connection)
+            from_packet, event_size, table_map, ctl_connection, **kwargs)
         self.packet.advance(event_size)

pymysqlreplication/packet.py

@@ -48,7 +48,7 @@ class BinLogPacketWrapper(object):
 
     }
 
-    def __init__(self, from_packet, table_map, ctl_connection, use_checksum, allowed_events = None):
+    def __init__(self, from_packet, table_map, ctl_connection, use_checksum, allowed_events = None, only_tables = None):
         # -1 because we ignore the ok byte
         self.read_bytes = 0
         # Used when we want to override a value in the data buffer
@@ -86,7 +86,9 @@ def __init__(self, from_packet, table_map, ctl_connection, use_checksum, allowed
         if event_class not in allowed_events:
             return
         self.event = event_class(self, event_size_without_header, table_map,
-                                 ctl_connection)
+                                 ctl_connection, only_tables = only_tables)
+        if self.event._processed == False:
+            self.event = None
 
     def read(self, size):
         size = int(size)

pymysqlreplication/row_event.py

@@ -14,15 +14,24 @@
 
 
 class RowsEvent(BinLogEvent):
-    def __init__(self, from_packet, event_size, table_map, ctl_connection):
+    def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs):
         super(RowsEvent, self).__init__(from_packet, event_size, table_map,
-                                        ctl_connection)
+                                        ctl_connection, **kwargs)
         self.__rows = None
+        self.__only_tables = kwargs["only_tables"]
 
         #Header
         self.table_id = self._read_table_id()
         self.primary_key = table_map[self.table_id].data["primary_key"]
 
+        # Additional information
+        self.schema = self.table_map[self.table_id].schema
+        self.table = self.table_map[self.table_id].table
+
+        if self.__only_tables is not None and self.table not in self.__only_tables:
+            self._processed = False
+            return
+
         #Event V2
         if self.event_type == BINLOG.WRITE_ROWS_EVENT_V2 or \
                 self.event_type == BINLOG.DELETE_ROWS_EVENT_V2 or \
@@ -36,10 +45,6 @@ def __init__(self, from_packet, event_size, table_map, ctl_connection):
         self.number_of_columns = self.packet.read_length_coded_binary()
         self.columns = self.table_map[self.table_id].columns
 
-        # Additional information
-        self.schema = self.table_map[self.table_id].schema
-        self.table = self.table_map[self.table_id].table
-
     def __is_null(self, null_bitmap, position):
         bit = null_bitmap[int(position / 8)]
         if type(bit) is str:
@@ -370,11 +375,12 @@ class DeleteRowsEvent(RowsEvent):
     For each row you have a hash with a single key: values which contain the data of the removed line.
     """
 
-    def __init__(self, from_packet, event_size, table_map, ctl_connection):
+    def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs):
         super(DeleteRowsEvent, self).__init__(from_packet, event_size,
-                                              table_map, ctl_connection)
-        self.columns_present_bitmap = self.packet.read(
-            (self.number_of_columns + 7) / 8)
+                                              table_map, ctl_connection, **kwargs)
+        if self._processed:
+            self.columns_present_bitmap = self.packet.read(
+                (self.number_of_columns + 7) / 8)
 
     def _fetch_one_row(self):
         row = {}
@@ -398,11 +404,12 @@ class WriteRowsEvent(RowsEvent):
     For each row you have a hash with a single key: values which contain the data of the new line.
     """
 
-    def __init__(self, from_packet, event_size, table_map, ctl_connection):
+    def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs):
         super(WriteRowsEvent, self).__init__(from_packet, event_size,
-                                             table_map, ctl_connection)
-        self.columns_present_bitmap = self.packet.read(
-            (self.number_of_columns + 7) / 8)
+                                             table_map, ctl_connection, **kwargs)
+        if self._processed:
+            self.columns_present_bitmap = self.packet.read(
+                (self.number_of_columns + 7) / 8)
 
     def _fetch_one_row(self):
         row = {}
@@ -431,14 +438,15 @@ class UpdateRowsEvent(RowsEvent):
     http://dev.mysql.com/doc/refman/5.6/en/replication-options-binary-log.html#sysvar_binlog_row_image
     """
 
-    def __init__(self, from_packet, event_size, table_map, ctl_connection):
+    def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs):
         super(UpdateRowsEvent, self).__init__(from_packet, event_size,
-                                              table_map, ctl_connection)
-        #Body
-        self.columns_present_bitmap = self.packet.read(
-            (self.number_of_columns + 7) / 8)
-        self.columns_present_bitmap2 = self.packet.read(
-            (self.number_of_columns + 7) / 8)
+                                              table_map, ctl_connection, **kwargs)
+        if self._processed:
+            #Body
+            self.columns_present_bitmap = self.packet.read(
+                (self.number_of_columns + 7) / 8)
+            self.columns_present_bitmap2 = self.packet.read(
+                (self.number_of_columns + 7) / 8)
 
     def _fetch_one_row(self):
         row = {}
@@ -468,9 +476,9 @@ class TableMapEvent(BinLogEvent):
     A end user of the lib should have no usage of this
     """
 
-    def __init__(self, from_packet, event_size, table_map, ctl_connection):
+    def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs):
         super(TableMapEvent, self).__init__(from_packet, event_size,
-                                            table_map, ctl_connection)
+                                            table_map, ctl_connection, **kwargs)
 
         # Post-Header
         self.table_id = self._read_table_id()

pymysqlreplication/tests/benchmark.py

@@ -24,7 +24,8 @@ def consume_events():
                                 server_id=3,
                                 resume_stream=False,
                                 blocking=True,
-                                only_events = [UpdateRowsEvent])
+                                only_events = [UpdateRowsEvent],
+                                only_tables = ['test'] )
     start = time.clock()
     i = 0.0
     for binlogevent in stream:
@@ -50,13 +51,16 @@ def consume_events():
 conn = pymysql.connect(**database)
 execute(conn, "CREATE TABLE test (i INT) ENGINE = MEMORY")
 execute(conn, "INSERT INTO test VALUES(1)")
+execute(conn, "CREATE TABLE test2 (i INT) ENGINE = MEMORY")
+execute(conn, "INSERT INTO test2 VALUES(1)")
 execute(conn, "RESET MASTER")
 
 
 if os.fork() != 0:
     print("Start insert data")
     while True:
         execute(conn, "UPDATE test SET i = i + 1;")
+        execute(conn, "UPDATE test2 SET i = i + 1;")
 else:
     consume_events()
     #cProfile.run('consume_events()')

pymysqlreplication/tests/test_basic.py

@@ -111,6 +111,28 @@ def test_filtering_ignore_events(self):
         event = self.stream.fetchone()
         self.assertIsInstance(event, RotateEvent)
 
+    def test_filtering_table_event(self):
+        self.stream.close()
+        self.stream = BinLogStreamReader(
+            self.database,
+            server_id=1024,
+            only_events=[WriteRowsEvent],
+            only_tables = ["test_2"])
+
+        query = "CREATE TABLE test_2 (id INT NOT NULL AUTO_INCREMENT, data VARCHAR (50) NOT NULL, PRIMARY KEY (id))"
+        self.execute(query)
+        query = "CREATE TABLE test_3 (id INT NOT NULL AUTO_INCREMENT, data VARCHAR (50) NOT NULL, PRIMARY KEY (id))"
+        self.execute(query)
+
+        self.execute("INSERT INTO test_2 (data) VALUES ('alpha')")
+        self.execute("INSERT INTO test_3 (data) VALUES ('alpha')")
+        self.execute("INSERT INTO test_2 (data) VALUES ('beta')")
+        self.execute("COMMIT")
+        event = self.stream.fetchone()
+        self.assertEqual(event.table, "test_2")
+        event = self.stream.fetchone()
+        self.assertEqual(event.table, "test_2")
+
     def test_write_row_event(self):
         query = "CREATE TABLE test (id INT NOT NULL AUTO_INCREMENT, data VARCHAR (50) NOT NULL, PRIMARY KEY (id))"
         self.execute(query)