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))")