Merge pull request #351 from dongwook-chan/zerofill

Support 'ZEROFILL' attribute for numeric datatypes

Author: julien-duponchelle

pymysqlreplication/column.py

@@ -22,6 +22,7 @@ def __parse_column_definition(self, column_type, column_schema, packet):
         self.character_set_name = column_schema["CHARACTER_SET_NAME"]
         self.comment = column_schema["COLUMN_COMMENT"]
         self.unsigned = column_schema["COLUMN_TYPE"].find("unsigned") != -1
+        self.zerofill = column_schema["COLUMN_TYPE"].find("zerofill") != -1
         self.type_is_bool = False
         self.is_primary = column_schema["COLUMN_KEY"] == "PRI"
 

pymysqlreplication/row_event.py

@@ -92,6 +92,7 @@ def _read_column_data(self,  cols_bitmap):
             column = self.columns[i]
             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
 
             if BitGet(cols_bitmap, i) == 0:
                 values[name] = None
@@ -102,21 +103,29 @@ def _read_column_data(self,  cols_bitmap):
             elif column.type == FIELD_TYPE.TINY:
                 if unsigned:
                     values[name] = struct.unpack("<B", self.packet.read(1))[0]
+                    if zerofill:
+                        values[name] = format(values[name], '03d')
                 else:
                     values[name] = struct.unpack("<b", self.packet.read(1))[0]
             elif column.type == FIELD_TYPE.SHORT:
                 if unsigned:
                     values[name] = struct.unpack("<H", self.packet.read(2))[0]
+                    if zerofill:
+                        values[name] = format(values[name], '05d')
                 else:
                     values[name] = struct.unpack("<h", self.packet.read(2))[0]
             elif column.type == FIELD_TYPE.LONG:
                 if unsigned:
                     values[name] = struct.unpack("<I", self.packet.read(4))[0]
+                    if zerofill:
+                        values[name] = format(values[name], '010d')
                 else:
                     values[name] = struct.unpack("<i", self.packet.read(4))[0]
             elif column.type == FIELD_TYPE.INT24:
                 if unsigned:
                     values[name] = self.packet.read_uint24()
+                    if zerofill:
+                        values[name] = format(values[name], '08d')
                 else:
                     values[name] = self.packet.read_int24()
             elif column.type == FIELD_TYPE.FLOAT:
@@ -155,6 +164,8 @@ def _read_column_data(self,  cols_bitmap):
             elif column.type == FIELD_TYPE.LONGLONG:
                 if unsigned:
                     values[name] = self.packet.read_uint64()
+                    if zerofill:
+                        values[name] = format(values[name], '020d')
                 else:
                     values[name] = self.packet.read_int64()
             elif column.type == FIELD_TYPE.YEAR:

pymysqlreplication/tests/test_data_objects.py

@@ -56,6 +56,7 @@ def test_column_serializable(self):
         self.assertIn("character_set_name", serialized)
         self.assertIn("comment", serialized)
         self.assertIn("unsigned", serialized)
+        self.assertIn("zerofill", serialized)
         self.assertIn("type_is_bool", serialized)
         self.assertIn("is_primary", serialized)
 

pymysqlreplication/tests/test_data_type.py

@@ -609,6 +609,21 @@ def test_encoding_utf8(self):
         event = self.create_and_insert_value(create_query, insert_query)
         self.assertMultiLineEqual(event.rows[0]["values"]["test"], string)
 
+    def test_zerofill(self):
+        create_query = "CREATE TABLE test ( \
+            test TINYINT UNSIGNED ZEROFILL DEFAULT NULL, \
+            test2 SMALLINT UNSIGNED ZEROFILL DEFAULT NULL, \
+            test3 MEDIUMINT UNSIGNED ZEROFILL DEFAULT NULL, \
+            test4 INT UNSIGNED ZEROFILL DEFAULT NULL, \
+            test5 BIGINT UNSIGNED ZEROFILL DEFAULT NULL \
+            )"
+        insert_query = "INSERT INTO test (test, test2, test3, test4, test5) VALUES(1, 1, 1, 1, 1)"
+        event = self.create_and_insert_value(create_query, insert_query)
+        self.assertEqual(event.rows[0]["values"]["test"], '001')
+        self.assertEqual(event.rows[0]["values"]["test2"], '00001')
+        self.assertEqual(event.rows[0]["values"]["test3"], '00000001')
+        self.assertEqual(event.rows[0]["values"]["test4"], '0000000001')
+        self.assertEqual(event.rows[0]["values"]["test5"], '00000000000000000001')
 
 if __name__ == "__main__":
     unittest.main()