Zero-pad fixed-length binary fields (#401)

MySQL is zero-padding fixed-length binary fields [1], but is not storing
trailing zeros to the binlog. Consequently, when reading values of
fixed-length binary fields, the value must be padded with zeros up to
the specified length.

The `test_fixed_length_binary` test case exemplifies the issue.
`varbinary` columns are not padded and not affected.

This commit extends the `information_schema` query in
`__get_table_information` to fetch information about the length
limitation of the field and stores this information in the
`fixed_binary_length` attribute of `Column` instances.

Upon decoding of binary fields (which share the same binlog field type
value of `254` as strings), the value is zero padded at the end, when it
does not meet the specified length as retrieved from the information
schema.

Fixes #400.

[1] https://dev.mysql.com/doc/refman/5.7/en/binary-varbinary.html

Author: oseemann

pymysqlreplication/binlogstream.py

@@ -616,7 +616,8 @@ def __get_table_information(self, schema, table):
                 cur.execute("""
                     SELECT
                         COLUMN_NAME, COLLATION_NAME, CHARACTER_SET_NAME,
-                        COLUMN_COMMENT, COLUMN_TYPE, COLUMN_KEY, ORDINAL_POSITION
+                        COLUMN_COMMENT, COLUMN_TYPE, COLUMN_KEY, ORDINAL_POSITION,
+                        DATA_TYPE, CHARACTER_OCTET_LENGTH
                     FROM
                         information_schema.columns
                     WHERE

pymysqlreplication/column.py

@@ -26,6 +26,12 @@ def __parse_column_definition(self, column_type, column_schema, packet):
         self.type_is_bool = False
         self.is_primary = column_schema["COLUMN_KEY"] == "PRI"
 
+        # Check for fixed-length binary type. When that's the case then we need
+        # to zero-pad the values to full length at read time.
+        self.fixed_binary_length = None
+        if column_schema["DATA_TYPE"] == "binary":
+            self.fixed_binary_length = column_schema["CHARACTER_OCTET_LENGTH"]
+
         if self.type == FIELD_TYPE.VARCHAR:
             self.max_length = struct.unpack('<H', packet.read(2))[0]
         elif self.type == FIELD_TYPE.DOUBLE:

pymysqlreplication/row_event.py

@@ -109,6 +109,7 @@ def _read_column_data(self,  cols_bitmap):
             name = self.table_map[self.table_id].columns[i].name
             unsigned = self.table_map[self.table_id].columns[i].unsigned
             zerofill = self.table_map[self.table_id].columns[i].zerofill
+            fixed_binary_length = self.table_map[self.table_id].columns[i].fixed_binary_length
 
             if BitGet(cols_bitmap, i) == 0:
                 values[name] = None
@@ -154,6 +155,14 @@ def _read_column_data(self,  cols_bitmap):
                     values[name] = self.__read_string(2, column)
                 else:
                     values[name] = self.__read_string(1, column)
+
+                if fixed_binary_length and len(values[name]) < fixed_binary_length:
+                    # Fixed-length binary fields are stored in the binlog
+                    # without trailing zeros and must be padded with zeros up
+                    # to the specified length at read time.
+                    nr_pad = fixed_binary_length - len(values[name])
+                    values[name] += b'\x00' * nr_pad
+
             elif column.type == FIELD_TYPE.NEWDECIMAL:
                 values[name] = self.__read_new_decimal(column)
             elif column.type == FIELD_TYPE.BLOB:
@@ -640,6 +649,8 @@ def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs)
                         'COLUMN_NAME': '__dropped_col_{i}__'.format(i=i),
                         'COLLATION_NAME': None,
                         'CHARACTER_SET_NAME': None,
+                        'CHARACTER_OCTET_LENGTH': None,
+                        'DATA_TYPE': 'BLOB',
                         'COLUMN_COMMENT': None,
                         'COLUMN_TYPE': 'BLOB',  # we don't know what it is, so let's not do anything with it.
                         'COLUMN_KEY': '',

pymysqlreplication/tests/test_data_objects.py

@@ -22,6 +22,8 @@ def test_column_is_primary(self):
                      {"COLUMN_NAME": "test",
                       "COLLATION_NAME": "utf8_general_ci",
                       "CHARACTER_SET_NAME": "UTF8",
+                      "CHARACTER_OCTET_LENGTH": None,
+                      "DATA_TYPE": "tinyint",
                       "COLUMN_COMMENT": "",
                       "COLUMN_TYPE": "tinyint(2)",
                       "COLUMN_KEY": "PRI"},
@@ -33,6 +35,8 @@ def test_column_not_primary(self):
                      {"COLUMN_NAME": "test",
                       "COLLATION_NAME": "utf8_general_ci",
                       "CHARACTER_SET_NAME": "UTF8",
+                      "CHARACTER_OCTET_LENGTH": None,
+                      "DATA_TYPE": "tinyint",
                       "COLUMN_COMMENT": "",
                       "COLUMN_TYPE": "tinyint(2)",
                       "COLUMN_KEY": ""},
@@ -44,6 +48,8 @@ def test_column_serializable(self):
                      {"COLUMN_NAME": "test",
                       "COLLATION_NAME": "utf8_general_ci",
                       "CHARACTER_SET_NAME": "UTF8",
+                      "CHARACTER_OCTET_LENGTH": None,
+                      "DATA_TYPE": "tinyint",
                       "COLUMN_COMMENT": "",
                       "COLUMN_TYPE": "tinyint(2)",
                       "COLUMN_KEY": "PRI"},

pymysqlreplication/tests/test_data_type.py

@@ -101,6 +101,18 @@ def create_and_get_tablemap_event(self, bit):
 
         return event
 
+    def test_varbinary(self):
+        create_query = "CREATE TABLE test(b VARBINARY(4))"
+        insert_query = "INSERT INTO test VALUES(UNHEX('ff010000'))"
+        event = self.create_and_insert_value(create_query, insert_query)
+        self.assertEqual(event.rows[0]["values"]["b"], b'\xff\x01\x00\x00')
+
+    def test_fixed_length_binary(self):
+        create_query = "CREATE TABLE test(b BINARY(4))"
+        insert_query = "INSERT INTO test VALUES(UNHEX('ff010000'))"
+        event = self.create_and_insert_value(create_query, insert_query)
+        self.assertEqual(event.rows[0]["values"]["b"], b'\xff\x01\x00\x00')
+
     def test_decimal(self):
         create_query = "CREATE TABLE test (test DECIMAL(2,1))"
         insert_query = "INSERT INTO test VALUES(4.2)"