Merge pull request julien-duponchelle#117 from vartec/master

Partial fix for dropped columns blowing up replication

Author: julien-duponchelle

pymysqlreplication/row_event.py

@@ -547,7 +547,20 @@ def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs)
             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]
+                try:
+                    column_schema = self.column_schemas[i]
+                except IndexError:
+                    # this a dirty hack to prevent row evens containing columns which have been dropped prior
+                    # to pymysqlreplication start, but replayed from binlog from blowing up the service.
+                    # TODO: this does not address the issue if the column other than the last one is dropped
+                    column_schema = {
+                        'COLUMN_NAME': '__dropped_col_{i}__'.format(i=i),
+                        'COLLATION_NAME': None,
+                        'CHARACTER_SET_NAME': None,
+                        'COLUMN_COMMENT': None,
+                        'COLUMN_TYPE': 'BLOB',  # we don't know what it is, so let's not do anything with it.
+                        'COLUMN_KEY': '',
+                    }
                 col = Column(byte2int(column_type), column_schema, from_packet)
                 self.columns.append(col)
 

pymysqlreplication/tests/test_basic.py

@@ -563,6 +563,28 @@ def test_drop_table(self):
 
         self.assertEqual([], event.rows)
 
+    def test_drop_column(self):
+        self.stream.close()
+        self.execute("CREATE TABLE test_drop_column (id INTEGER(11), data VARCHAR(50))")
+        self.execute("INSERT INTO test_drop_column VALUES (1, 'A value')")
+        self.execute("COMMIT")
+        self.execute("ALTER TABLE test_drop_column DROP COLUMN data")
+        self.execute("INSERT INTO test_drop_column VALUES (2)")
+        self.execute("COMMIT")
+
+        self.stream = BinLogStreamReader(
+            self.database,
+            server_id=1024,
+            only_events=(WriteRowsEvent,),
+            )
+        try:
+            self.stream.fetchone()  # insert with two values
+            self.stream.fetchone()  # insert with one value
+        except Exception as e:
+            self.fail("raised unexpected exception: {exception}".format(exception=e))
+        finally:
+            self.resetBinLog()
+
 
 class TestGtidBinLogStreamReader(base.PyMySQLReplicationTestCase):
     def setUp(self):