Merge pull request julien-duponchelle#395 from paulvic/paul/decode-errors

Add support for ignoring string decode errors

Author: julien-duponchelle

pymysqlreplication/binlogstream.py

@@ -140,7 +140,8 @@ def __init__(self, connection_settings, server_id,
                  pymysql_wrapper=None,
                  fail_on_table_metadata_unavailable=False,
                  slave_heartbeat=None,
-                 is_mariadb=False):
+                 is_mariadb=False,
+                 ignore_decode_errors=False):
         """
         Attributes:
             ctl_connection_settings: Connection settings for cluster holding
@@ -177,6 +178,8 @@ def __init__(self, connection_settings, server_id,
                              for semantics
             is_mariadb: Flag to indicate it's a MariaDB server, used with auto_position
                     to point to Mariadb specific GTID.
+            ignore_decode_errors: If true, any decode errors encountered 
+                                  when reading column data will be ignored.
         """
 
         self.__connection_settings = connection_settings
@@ -198,6 +201,7 @@ def __init__(self, connection_settings, server_id,
         self.__allowed_events = self._allowed_event_list(
             only_events, ignored_events, filter_non_implemented_events)
         self.__fail_on_table_metadata_unavailable = fail_on_table_metadata_unavailable
+        self.__ignore_decode_errors = ignore_decode_errors
 
         # We can't filter on packet level TABLE_MAP and rotate event because
         # we need them for handling other operations
@@ -502,7 +506,8 @@ def fetchone(self):
                                                self.__only_schemas,
                                                self.__ignored_schemas,
                                                self.__freeze_schema,
-                                               self.__fail_on_table_metadata_unavailable)
+                                               self.__fail_on_table_metadata_unavailable,
+                                               self.__ignore_decode_errors)
 
             if binlog_event.event_type == ROTATE_EVENT:
                 self.log_pos = binlog_event.event.position

pymysqlreplication/event.py

@@ -13,14 +13,16 @@ def __init__(self, from_packet, event_size, table_map, ctl_connection,
                  only_schemas=None,
                  ignored_schemas=None,
                  freeze_schema=False,
-                 fail_on_table_metadata_unavailable=False):
+                 fail_on_table_metadata_unavailable=False,
+                 ignore_decode_errors=False):
         self.packet = from_packet
         self.table_map = table_map
         self.event_type = self.packet.event_type
         self.timestamp = self.packet.timestamp
         self.event_size = event_size
         self._ctl_connection = ctl_connection
         self._fail_on_table_metadata_unavailable = fail_on_table_metadata_unavailable
+        self._ignore_decode_errors = ignore_decode_errors
         # The event have been fully processed, if processed is false
         # the event will be skipped
         self._processed = True

pymysqlreplication/packet.py

@@ -96,7 +96,8 @@ def __init__(self, from_packet, table_map, ctl_connection, use_checksum,
                  only_schemas,
                  ignored_schemas,
                  freeze_schema,
-                 fail_on_table_metadata_unavailable):
+                 fail_on_table_metadata_unavailable,
+                 ignore_decode_errors):
         # -1 because we ignore the ok byte
         self.read_bytes = 0
         # Used when we want to override a value in the data buffer
@@ -140,7 +141,8 @@ def __init__(self, from_packet, table_map, ctl_connection, use_checksum,
                                  only_schemas=only_schemas,
                                  ignored_schemas=ignored_schemas,
                                  freeze_schema=freeze_schema,
-                                 fail_on_table_metadata_unavailable=fail_on_table_metadata_unavailable)
+                                 fail_on_table_metadata_unavailable=fail_on_table_metadata_unavailable,
+                                 ignore_decode_errors=ignore_decode_errors)
         if self.event._processed == False:
             self.event = None
 

pymysqlreplication/row_event.py

@@ -247,7 +247,8 @@ def __read_string(self, size, column):
         string = self.packet.read_length_coded_pascal_string(size)
         if column.character_set_name is not None:
             encoding = self.charset_to_encoding(column.character_set_name)
-            string = string.decode(encoding)
+            decode_errors = "ignore" if self._ignore_decode_errors else "strict"
+            string = string.decode(encoding, decode_errors)
         return string
 
     def __read_bit(self, column):

pymysqlreplication/tests/base.py

@@ -85,6 +85,11 @@ def execute(self, query):
         c = self.conn_control.cursor()
         c.execute(query)
         return c
+    
+    def execute_with_args(self, query, args):
+        c = self.conn_control.cursor()
+        c.execute(query, args)
+        return c
 
     def resetBinLog(self):
         self.execute("RESET MASTER")

pymysqlreplication/tests/test_basic.py

@@ -678,6 +678,39 @@ def test_drop_table_tablemetadata_unavailable(self):
             self.resetBinLog()
             assert had_error
 
+    def test_ignore_decode_errors(self):
+        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
+        )
+        event = self.stream.fetchone()
+        event = self.stream.fetchone()   
+        with self.assertRaises(UnicodeError) as exception:
+            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
+        )
+        self.stream.fetchone()
+        self.stream.fetchone()        
+        event = self.stream.fetchone()
+        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))")