Allow to force the encoding for old MySQL release

For MySQL before 5.XX we guess the encoding, this allow to manually set
it.

Author: julien-duponchelle

pymysqlreplication/binlogstream.py

@@ -1,43 +1,43 @@
-import struct
 import logging
-from packaging.version import Version
+import struct
 
 import pymysql
+from packaging.version import Version
 from pymysql.constants.COMMAND import COM_BINLOG_DUMP, COM_REGISTER_SLAVE
 from pymysql.cursors import DictCursor
 
-from .constants.BINLOG import TABLE_MAP_EVENT, ROTATE_EVENT, FORMAT_DESCRIPTION_EVENT
+from .constants.BINLOG import FORMAT_DESCRIPTION_EVENT, ROTATE_EVENT, TABLE_MAP_EVENT
 from .event import (
-    QueryEvent,
-    RotateEvent,
-    FormatDescriptionEvent,
-    XidEvent,
-    GtidEvent,
-    StopEvent,
-    XAPrepareEvent,
     BeginLoadQueryEvent,
     ExecuteLoadQueryEvent,
+    FormatDescriptionEvent,
+    GtidEvent,
     HeartbeatLogEvent,
-    NotImplementedEvent,
-    MariadbGtidEvent,
     MariadbAnnotateRowsEvent,
-    RandEvent,
+    MariadbBinLogCheckPointEvent,
+    MariadbGtidEvent,
+    MariadbGtidListEvent,
     MariadbStartEncryptionEvent,
+    NotImplementedEvent,
+    PreviousGtidsEvent,
+    QueryEvent,
+    RandEvent,
+    RotateEvent,
     RowsQueryLogEvent,
-    MariadbGtidListEvent,
-    MariadbBinLogCheckPointEvent,
+    StopEvent,
     UserVarEvent,
-    PreviousGtidsEvent,
+    XAPrepareEvent,
+    XidEvent,
 )
 from .exceptions import BinLogNotEnabled
 from .gtid import GtidSet
 from .packet import BinLogPacketWrapper
 from .row_event import (
-    UpdateRowsEvent,
-    WriteRowsEvent,
     DeleteRowsEvent,
-    TableMapEvent,
     PartialUpdateRowsEvent,
+    TableMapEvent,
+    UpdateRowsEvent,
+    WriteRowsEvent,
 )
 
 try:
@@ -185,6 +185,7 @@ def __init__(
         slave_heartbeat=None,
         is_mariadb=False,
         annotate_rows_event=False,
+        force_encoding=None,
         ignore_decode_errors=False,
         verify_checksum=False,
         enable_logging=True,
@@ -225,6 +226,8 @@ def __init__(
                     to point to Mariadb specific GTID.
             annotate_rows_event: Parameter value to enable annotate rows event in mariadb,
                     used with 'is_mariadb'
+            force_encoding: Force the encoding to decode a string this for MySQL 5.XXX This is the charset
+                to use example latin-1
             ignore_decode_errors: If true, any decode errors encountered
                                   when reading column data will be ignored.
             verify_checksum: If true, verify events read from the binary log by examining checksums.
@@ -252,6 +255,7 @@ def __init__(
             only_events, ignored_events, filter_non_implemented_events
         )
         self.__ignore_decode_errors = ignore_decode_errors
+        self.__force_encoding = force_encoding
         self.__verify_checksum = verify_checksum
         self.__optional_meta_data = False
 
@@ -628,6 +632,7 @@ def fetchone(self):
                 self.__ignored_schemas,
                 self.__freeze_schema,
                 self.__ignore_decode_errors,
+                self.__force_encoding,
                 self.__verify_checksum,
                 self.__optional_meta_data,
             )

pymysqlreplication/event.py

@@ -1,15 +1,15 @@
 import binascii
-import struct
 import datetime
 import decimal
-import zlib
+import json
 import logging
+import struct
+import zlib
+from typing import Optional, Union
 
 from pymysqlreplication.constants.STATUS_VAR_KEY import *
 from pymysqlreplication.exceptions import StatusVariableMismatch
 from pymysqlreplication.util.bytes import parse_decimal_from_bytes
-from typing import Union, Optional
-import json
 
 
 class BinLogEvent(object):
@@ -26,6 +26,7 @@ def __init__(
         ignored_schemas=None,
         freeze_schema=False,
         ignore_decode_errors=False,
+        force_encoding=None,
         verify_checksum=False,
         optional_meta_data=False,
     ):
@@ -37,6 +38,7 @@ def __init__(
         self._ctl_connection = ctl_connection
         self.mysql_version = mysql_version
         self._ignore_decode_errors = ignore_decode_errors
+        self._force_encoding = force_encoding
         self._verify_checksum = verify_checksum
         self._is_event_valid = None
         # The event have been fully processed, if processed is false

pymysqlreplication/packet.py

@@ -1,7 +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
+from pymysqlreplication.json_binary import JsonDiff, JsonDiffOperation, parse_json
+from pymysqlreplication.util.bytes import *
 
 # Constants from PyMYSQL source code
 NULL_COLUMN = 251
@@ -72,6 +72,7 @@ def __init__(
         ignored_schemas,
         freeze_schema,
         ignore_decode_errors,
+        force_encoding,
         verify_checksum,
         optional_meta_data,
     ):
@@ -125,6 +126,7 @@ def __init__(
             ignored_schemas=ignored_schemas,
             freeze_schema=freeze_schema,
             ignore_decode_errors=ignore_decode_errors,
+            force_encoding=force_encoding,
             verify_checksum=verify_checksum,
             optional_meta_data=optional_meta_data,
         )

pymysqlreplication/row_event.py

@@ -1,18 +1,15 @@
-import struct
-import decimal
 import datetime
+import decimal
+import struct
+from enum import Enum
 
 from pymysql.charset import charset_by_name
-from enum import Enum
 
-from .event import BinLogEvent
-from .constants import FIELD_TYPE
-from .constants import BINLOG
-from .constants import CHARSET
-from .constants import NONE_SOURCE
+from .bitmap import BitCount, BitGet
 from .column import Column
+from .constants import BINLOG, CHARSET, FIELD_TYPE, NONE_SOURCE
+from .event import BinLogEvent
 from .table import Table
-from .bitmap import BitCount, BitGet
 
 
 class RowsEvent(BinLogEvent):
@@ -332,7 +329,10 @@ def __read_string(self, size, column):
         else:
             # MYSQL 5.xx Version  Goes Here
             # We don't know encoding type So apply Default Utf-8
-            string = string.decode(errors=decode_errors)
+            if self._force_encoding:
+                string = string.decode(encoding=self._force_encoding, errors=decode_errors)
+            else:
+                string = string.decode(errors=decode_errors)
         return string
 
     def __read_bit(self, column):

pymysqlreplication/tests/test_basic.py

@@ -1,19 +1,20 @@
+import copy
 import io
 import time
 import unittest
+from unittest.mock import patch
+
+from pymysql.protocol import MysqlPacket
 
-from pymysqlreplication.json_binary import JsonDiff, JsonDiffOperation
-from pymysqlreplication.tests import base
 from pymysqlreplication import BinLogStreamReader
-from pymysqlreplication.gtid import GtidSet, Gtid
-from pymysqlreplication.event import *
 from pymysqlreplication.constants.BINLOG import *
 from pymysqlreplication.constants.NONE_SOURCE import *
-from pymysqlreplication.row_event import *
+from pymysqlreplication.event import *
+from pymysqlreplication.gtid import Gtid, GtidSet
+from pymysqlreplication.json_binary import JsonDiff, JsonDiffOperation
 from pymysqlreplication.packet import BinLogPacketWrapper
-from pymysql.protocol import MysqlPacket
-from unittest.mock import patch
-
+from pymysqlreplication.row_event import *
+from pymysqlreplication.tests import base
 
 __all__ = [
     "TestBasicBinLogStreamReader",
@@ -284,43 +285,11 @@ def test_delete_row_event(self):
         self.execute("COMMIT")
 
         self.assertIsInstance(self.stream.fetchone(), RotateEvent)
-        self.assertIsInstance(self.stream.fetchone(), FormatDescriptionEvent)
-
-        # QueryEvent for the BEGIN
-        if not self.isMariaDB():
-            self.assertIsInstance(self.stream.fetchone(), QueryEvent)
-
-        self.assertIsInstance(self.stream.fetchone(), TableMapEvent)
-
-        event = self.stream.fetchone()
-        if self.isMySQL56AndMore():
-            self.assertEqual(event.event_type, DELETE_ROWS_EVENT_V2)
-        else:
-            self.assertEqual(event.event_type, DELETE_ROWS_EVENT_V1)
-        self.assertIsInstance(event, DeleteRowsEvent)
-        if event.table_map[event.table_id].column_name_flag:
-            self.assertEqual(event.rows[0]["values"]["id"], 1)
-            self.assertEqual(event.rows[0]["values"]["data"], "Hello World")
-
-    def test_update_row_event(self):
-        query = "CREATE TABLE test (id INT NOT NULL AUTO_INCREMENT, data VARCHAR (50) NOT NULL, PRIMARY KEY (id))"
-        self.execute(query)
-        query = "INSERT INTO test (data) VALUES('Hello')"
-        self.execute(query)
-
-        self.resetBinLog()
-
-        query = "UPDATE test SET data = 'World' WHERE id = 1"
-        self.execute(query)
+        self.execute(
+            "INSERT INTO test (data) VALUES ('ó')".encode("latin-1")
+        )
         self.execute("COMMIT")
 
-        self.assertIsInstance(self.stream.fetchone(), RotateEvent)
-        self.assertIsInstance(self.stream.fetchone(), FormatDescriptionEvent)
-
-        # QueryEvent for the BEGIN
-        if not self.isMariaDB():
-            self.assertIsInstance(self.stream.fetchone(), QueryEvent)
-
         self.assertIsInstance(self.stream.fetchone(), TableMapEvent)
 
         event = self.stream.fetchone()
@@ -617,6 +586,7 @@ def create_binlog_packet_wrapper(pkt):
                 self.stream._BinLogStreamReader__ignored_schemas,
                 self.stream._BinLogStreamReader__freeze_schema,
                 self.stream._BinLogStreamReader__ignore_decode_errors,
+                self.stream._BinLogStreamReader__force_encoding,
                 self.stream._BinLogStreamReader__verify_checksum,
                 self.stream._BinLogStreamReader__optional_meta_data,
             )
@@ -837,42 +807,83 @@ 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_force_encoding(self):
+        if self.isMySQL80AndMore():
+            self.skipTest("MYSQL 8 Version don't need force encoding")
+        self.stream.close()
+
+        db = copy.copy(self.database)
+        db["charset"] = "latin1"
+        self.connect_conn_control(db)
+
+        string = "\u00e9"
+
+        create_query = (
+            "CREATE TABLE test (test CHAR(12)) CHARACTER SET latin1 COLLATE latin1_bin;"
+        )
+        insert_query = b"INSERT INTO test VALUES('" + string.encode("latin-1") + b"');"
+        self.execute(create_query)
+        self.execute(insert_query)
+        self.execute("COMMIT")
+
+        self.stream = BinLogStreamReader(
+            self.database,
+            server_id=1024,
+            only_events=(WriteRowsEvent,),
+            force_encoding="utf-8",
+        )
+        with self.assertRaises(UnicodeError):
+            event = self.stream.fetchone()
+            data = event.rows[0]["values"]["data"]
+
+        self.stream = BinLogStreamReader(
+            self.database,
+            server_id=1024,
+            only_events=(WriteRowsEvent,),
+            force_encoding="latin-1",
+        )
+        event = self.stream.fetchone()
+        if event.table_map[event.table_id].column_name_flag:
+            data = event.rows[0]["values"]["data"]
+            self.assertEqual(data, '[{"text":"  Some string"}]')
+
+
     def test_ignore_decode_errors(self):
         if self.isMySQL80AndMore():
             self.skipTest("MYSQL 8 Version Pymysql Data Error Incorrect string value")
+        problematic_unicode_string = "ó".encode("latin-1")
+        self.stream.close()
         problematic_unicode_string = (
             b'[{"text":"\xed\xa0\xbd \xed\xb1\x8d Some string"}]'
         )
-        self.stream.close()
         self.execute("CREATE TABLE test (data VARCHAR(50) CHARACTER SET utf8mb4)")
         self.execute_with_args(
             "INSERT INTO test (data) VALUES (%s)", (problematic_unicode_string)
         )
         self.execute("COMMIT")
 
-        # Initialize with ignore_decode_errors=False
         self.stream = BinLogStreamReader(
             self.database,
             server_id=1024,
             only_events=(WriteRowsEvent,),
-            ignore_decode_errors=False,
+            force_encoding=None,
         )
         with self.assertRaises(UnicodeError):
             event = self.stream.fetchone()
             data = event.rows[0]["values"]["data"]
 
-        # Initialize with ignore_decode_errors=True
         self.stream = BinLogStreamReader(
             self.database,
             server_id=1024,
             only_events=(WriteRowsEvent,),
-            ignore_decode_errors=True,
+            force_encoding="latin-1",
         )
         event = self.stream.fetchone()
         if event.table_map[event.table_id].column_name_flag:
             data = event.rows[0]["values"]["data"]
             self.assertEqual(data, '[{"text":"  Some string"}]')
 
+
     def test_drop_column(self):
         self.stream.close()
         self.execute("CREATE TABLE test_drop_column (id INTEGER(11), data VARCHAR(50))")