Fix row events for MySQL8 partitioned table

Since MySQL 8.0.16, partition info has been [added](mysql/mysql-server@e11a540) into extra_data in row event.

Current code is aware of extra_data but doesn't parse it accurately.
Its length is defined as 'extra_data_length / 8'.
However, the [maximum length of partition info](https://github.com/mysql/mysql-server/blob/beb865a960b9a8a16cf999c323e46c5b0c67f21f/libbinlogevents/include/rows_event.h#L772-L814) is 5.
Wrong definition leads program to read 0 (= 5/8) bytes of extra_data.

This causes any row event on partitioned table leads to packet parse failure (undefined behavior).
In some cases errors occur, and in other cases parse result shows  incorrect filed values as in [issue #354](#354 (comment)).

1. Fix byte-length of extra_data to conform to repl protocol
    extra_data_length / 8 -> extra_data_length - 2
    Refer to [MySQL Document](https://dev.mysql.com/doc/internals/en/rows-event.html) and [MySQL source code]((https://github.com/mysql/mysql-server/blob/beb865a960b9a8a16cf999c323e46c5b0c67f21f/libbinlogevents/src/rows_event.cpp#L415)).

Above fixes failure of packet parsing, but following modification is further required to get partition info.

2. Parse extra_data according to [partition info layout](https://github.com/mysql/mysql-server/blob/beb865a960b9a8a16cf999c323e46c5b0c67f21f/libbinlogevents/include/rows_event.h#L772-L814).

Author: dongwook-chan

pymysqlreplication/row_event.py

@@ -57,7 +57,17 @@ def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs)
                 self.event_type == BINLOG.DELETE_ROWS_EVENT_V2 or \
                 self.event_type == BINLOG.UPDATE_ROWS_EVENT_V2:
                 self.flags, self.extra_data_length = struct.unpack('<HH', self.packet.read(4))
-                self.extra_data = self.packet.read(self.extra_data_length / 8)
+                if self.extra_data_length > 2:
+                    self.extra_data_type = struct.unpack('<B', self.packet.read(1))[0]
+                    # partition information
+                    if self.extra_data_type == 1:
+                        if self.event_type == BINLOG.UPDATE_ROWS_EVENT_V2:
+                            self.partition_id, self.source_partition_id = struct.unpack('<HH', self.packet.read(4))
+                        else:
+                            self.partition_id = struct.unpack('<H', self.packet.read(2))[0]
+                    # NDB information etc.
+                    else:
+                        self.extra_data = self.packet.read(self.extra_data_length - 3)
         else:
             self.flags = struct.unpack('<H', self.packet.read(2))[0]
 

pymysqlreplication/tests/base.py

@@ -60,6 +60,10 @@ def isMySQL57(self):
         version = float(self.getMySQLVersion().rsplit('.', 1)[0])
         return version == 5.7
 
+    def isMySQL80AndMore(self):
+        version = float(self.getMySQLVersion().rsplit('.', 1)[0])
+        return version >= 8.0
+
     @property
     def supportsGTID(self):
         if not self.isMySQL56AndMore():

pymysqlreplication/tests/test_data_type.py

@@ -625,5 +625,21 @@ def test_zerofill(self):
         self.assertEqual(event.rows[0]["values"]["test4"], '0000000001')
         self.assertEqual(event.rows[0]["values"]["test5"], '00000000000000000001')
 
+    def test_extra_data(self):
+        if not self.isMySQL80AndMore():
+            self.skipTest("Not supported in this version of MySQL")
+        create_query = "CREATE TABLE test (id INTEGER) \
+            PARTITION BY RANGE (id) ( \
+                PARTITION p0 VALUES LESS THAN (1),   \
+                PARTITION p1 VALUES LESS THAN (2),   \
+                PARTITION p2 VALUES LESS THAN (3),   \
+                PARTITION p3 VALUES LESS THAN (4),   \
+                PARTITION p4 VALUES LESS THAN (5),   \
+            )"
+        insert_query = "INSERT INTO test (id) VALUES(3)"
+        event = self.create_and_insert_value(create_query, insert_query)
+        self.assertEqual(event.extra_data_type, 1)
+        self.assertEqual(event.partition_id, 3)
+
 if __name__ == "__main__":
     unittest.main()