Merge pull request #602 from dongwook-chan/fix/rows-query-parsing

Fix parsing of query field of RowsQueryEvent

pymysqlreplication/event.py

@@ -872,20 +872,18 @@ class RowsQueryLogEvent(BinLogEvent):
     More details are available in the MySQL Knowledge Base:
     https://dev.mysql.com/doc/dev/mysql-server/latest/classRows__query__log__event.html
 
-    :ivar query_length: uint - Length of the SQL statement
     :ivar query: str - The executed SQL statement
     """
 
     def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs):
         super(RowsQueryLogEvent, self).__init__(
             from_packet, event_size, table_map, ctl_connection, **kwargs
         )
-        self.query_length = self.packet.read_uint8()
-        self.query = self.packet.read(self.query_length).decode("utf-8")
+        self.packet.advance(1)
+        self.query = self.packet.read_available().decode("utf-8")
 
     def dump(self):
         print(f"=== {self.__class__.__name__} ===")
-        print(f"Query length: {self.query_length}")
         print(f"Query: {self.query}")
 
 

pymysqlreplication/packet.py

@@ -1,6 +1,7 @@
 from pymysqlreplication import constants, event, row_event
 from pymysqlreplication.json_binary import parse_json, JsonDiff, JsonDiffOperation
 from pymysqlreplication.util.bytes import *
+from pymysqlreplication.constants import BINLOG
 
 # Constants from PyMYSQL source code
 NULL_COLUMN = 251
@@ -384,3 +385,6 @@ def read_string(self):
 
     def bytes_to_read(self):
         return len(self.packet._data) - self.packet._position
+
+    def read_available(self):
+        return self.packet.read(self.bytes_to_read() - BINLOG.BINLOG_CHECKSUM_LEN)

pymysqlreplication/tests/test_basic.py

@@ -1567,6 +1567,36 @@ def test_rows_query_log_event(self):
         event = self.stream.fetchone()
         self.assertIsInstance(event, RowsQueryLogEvent)
 
+    def test_long_query(self):
+        """
+        Address issue #601
+        Do not use the first byte of the body to determine the length of the query.
+        1 byte can not represent the length of a query that is longer than 255 bytes.
+        """
+
+        self.stream.close()
+        self.stream = BinLogStreamReader(
+            self.database,
+            server_id=1024,
+            only_events=[RowsQueryLogEvent],
+        )
+
+        self.execute(
+            "CREATE TABLE IF NOT EXISTS test (id INT AUTO_INCREMENT PRIMARY KEY, long_text VARCHAR(256))"
+        )
+        long_query = (
+            "INSERT INTO test (long_text) VALUES ('"
+            "What is the longest word in english?"
+            "Pneumonoultramicroscopicsilicovolcanoconiosis is the longest word in the English language."
+            "This text has 256 characters and hence its length can not be represented in a single byte."
+            "')"
+        )
+        self.execute(long_query)
+        self.execute("COMMIT")
+        event = self.stream.fetchone()
+        self.assertIsInstance(event, RowsQueryLogEvent)
+        self.assertEqual(event.query, long_query)
+
 
 class TestLatin1(base.PyMySQLReplicationTestCase):
     def setUp(self):