Merge pull request #111 from noplay/indexerror

Prevent invalid table-map-entries to crash the whole app

Author: julien-duponchelle

pymysqlreplication/binlogstream.py

@@ -28,6 +28,7 @@
 MYSQL_EXPECTED_ERROR_CODES = [2013, 2006]
 
 
+
 class BinLogStreamReader(object):
     """Connect to replication stream and read event
     """

pymysqlreplication/event.py

@@ -20,6 +20,7 @@ def __init__(self, from_packet, event_size, table_map, ctl_connection,
         # The event have been fully processed, if processed is false
         # the event will be skipped
         self._processed = True
+        self.complete = True
 
     def _read_table_id(self):
         # Table ID is 6 byte

pymysqlreplication/row_event.py

@@ -54,6 +54,9 @@ def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs)
         self.number_of_columns = self.packet.read_length_coded_binary()
         self.columns = self.table_map[self.table_id].columns
 
+        if len(self.columns) == 0:  # could not read the table metadata, probably already dropped
+            self.complete = False
+
     def __is_null(self, null_bitmap, position):
         bit = null_bitmap[int(position / 8)]
         if type(bit) is str:
@@ -381,6 +384,10 @@ def _dump(self):
 
     def _fetch_rows(self):
         self.__rows = []
+
+        if not self.complete:
+            return
+
         while self.packet.read_bytes + 1 < self.event_size:
             self.__rows.append(self._fetch_one_row())
 
@@ -534,14 +541,15 @@ def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs)
         else:
             self.column_schemas = self._ctl_connection._get_table_information(self.schema, self.table)
 
-        # Read columns meta data
-        column_types = list(self.packet.read(self.column_count))
-        self.packet.read_length_coded_binary()
-        for i in range(0, len(column_types)):
-            column_type = column_types[i]
-            column_schema = self.column_schemas[i]
-            col = Column(byte2int(column_type), column_schema, from_packet)
-            self.columns.append(col)
+        if len(self.column_schemas) != 0:
+            # Read columns meta data
+            column_types = list(self.packet.read(self.column_count))
+            self.packet.read_length_coded_binary()
+            for i in range(0, len(column_types)):
+                column_type = column_types[i]
+                column_schema = self.column_schemas[i]
+                col = Column(byte2int(column_type), column_schema, from_packet)
+                self.columns.append(col)
 
         self.table_obj = Table(self.column_schemas, self.table_id, self.schema,
                                self.table, self.columns)

pymysqlreplication/tests/test_basic.py

@@ -486,6 +486,34 @@ def test_delete_multiple_row_event(self):
         self.assertEqual(event.rows[1]["values"]["id"], 2)
         self.assertEqual(event.rows[1]["values"]["data"], "World")
 
+    def test_drop_table(self):
+        self.execute("CREATE TABLE test (id INTEGER(11))")
+        self.execute("INSERT INTO test VALUES (1)")
+        self.execute("DROP TABLE test")
+        self.execute("COMMIT")
+
+        #RotateEvent
+        self.stream.fetchone()
+        #FormatDescription
+        self.stream.fetchone()
+        #QueryEvent for the Create Table
+        self.stream.fetchone()
+
+        #QueryEvent for the BEGIN
+        self.stream.fetchone()
+
+        event = self.stream.fetchone()
+        self.assertIsInstance(event, TableMapEvent)
+
+        event = self.stream.fetchone()
+        if self.isMySQL56AndMore():
+            self.assertEqual(event.event_type, WRITE_ROWS_EVENT_V2)
+        else:
+            self.assertEqual(event.event_type, WRITE_ROWS_EVENT_V1)
+        self.assertIsInstance(event, WriteRowsEvent)
+
+        self.assertEqual([], event.rows)
+
 
 class TestGtidBinLogStreamReader(base.PyMySQLReplicationTestCase):
     def test_read_query_event(self):