- fix bug when long json string type

- change json array type format
- add Json testcases

add update Json testcase

add test datetime,decimal,time cases

Author: sean-k1

pymysqlreplication/packet.py

@@ -133,7 +133,7 @@ def parse_json_object_or_array(bytes, is_small, is_object):
             return None
         values[i] = parse_json(json_type, bytes[value_offset:])
     if not is_object:
-        return values
+        return list(values.values())
     out = {}
     for i in range(count):
         out[keys[i]] = values[i]

pymysqlreplication/tests/test_basic.py

@@ -596,6 +596,33 @@ def create_binlog_packet_wrapper(pkt):
         self.assertEqual(binlog_event.event._is_event_valid, True)
         self.assertNotEqual(wrong_event.event._is_event_valid, True)
 
+    def test_json_update(self):
+        self.stream.close()
+        self.stream = BinLogStreamReader(
+            self.database, server_id=1024, only_events=[UpdateRowsEvent]
+        )
+        create_query = (
+            "CREATE TABLE setting_table( id SERIAL AUTO_INCREMENT, setting JSON);"
+        )
+        insert_query = """INSERT INTO setting_table (setting) VALUES ('{"btn": true, "model": false}');"""
+
+        update_query = """  UPDATE setting_table
+                            SET setting = JSON_REMOVE(setting, '$.model')
+                            WHERE id=1;
+                        """
+        self.execute(create_query)
+        self.execute(insert_query)
+        self.execute(update_query)
+        self.execute("COMMIT;")
+        event = self.stream.fetchone()
+
+        if event.table_map[event.table_id].column_name_flag:
+            self.assertEqual(
+                event.rows[0]["before_values"]["setting"],
+                {b"btn": True, b"model": False},
+            ),
+            self.assertEqual(event.rows[0]["after_values"]["setting"], {b"btn": True}),
+
 
 class TestMultipleRowBinLogStreamReader(base.PyMySQLReplicationTestCase):
     def setUp(self):

pymysqlreplication/tests/test_data_type.py

@@ -588,8 +588,6 @@ def test_geometry(self):
             )
 
     def test_json(self):
-        if not self.isMySQL57():
-            self.skipTest("Json is only supported in mysql 5.7")
         create_query = "CREATE TABLE test (id int, value json);"
         insert_query = """INSERT INTO test (id, value) VALUES (1, '{"my_key": "my_val", "my_key2": "my_val2"}');"""
         event = self.create_and_insert_value(create_query, insert_query)
@@ -600,8 +598,6 @@ def test_json(self):
             )
 
     def test_json_array(self):
-        if not self.isMySQL57():
-            self.skipTest("Json is only supported in mysql 5.7")
         create_query = "CREATE TABLE test (id int, value json);"
         insert_query = (
             """INSERT INTO test (id, value) VALUES (1, '["my_val", "my_val2"]');"""
@@ -611,8 +607,6 @@ def test_json_array(self):
             self.assertEqual(event.rows[0]["values"]["value"], [b"my_val", b"my_val2"])
 
     def test_json_large(self):
-        if not self.isMySQL57():
-            self.skipTest("Json is only supported in mysql 5.7")
         data = dict(
             [("foooo%i" % i, "baaaaar%i" % i) for i in range(2560)]
         )  # Make it large enough to reach 2^16 length
@@ -626,8 +620,6 @@ def test_json_large(self):
 
     def test_json_large_array(self):
         "Test json array larger than 64k bytes"
-        if not self.isMySQL57():
-            self.skipTest("Json is only supported in mysql 5.7")
         create_query = "CREATE TABLE test (id int, value json);"
         large_array = dict(my_key=[i for i in range(100000)])
         insert_query = "INSERT INTO test (id, value) VALUES (1, '%s');" % (
@@ -640,8 +632,6 @@ def test_json_large_array(self):
             )
 
     def test_json_large_with_literal(self):
-        if not self.isMySQL57():
-            self.skipTest("Json is only supported in mysql 5.7")
         data = dict(
             [("foooo%i" % i, "baaaaar%i" % i) for i in range(2560)], literal=True
         )  # Make it large with literal
@@ -654,9 +644,6 @@ def test_json_large_with_literal(self):
             self.assertEqual(event.rows[0]["values"]["value"], to_binary_dict(data))
 
     def test_json_types(self):
-        if not self.isMySQL57():
-            self.skipTest("Json is only supported in mysql 5.7")
-
         types = [
             True,
             False,
@@ -685,9 +672,6 @@ def test_json_types(self):
             self.setUp()
 
     def test_json_basic(self):
-        if not self.isMySQL57():
-            self.skipTest("Json is only supported in mysql 5.7")
-
         types = [
             True,
             False,
@@ -714,8 +698,6 @@ def test_json_basic(self):
             self.setUp()
 
     def test_json_unicode(self):
-        if not self.isMySQL57():
-            self.skipTest("Json is only supported in mysql 5.7")
         create_query = "CREATE TABLE test (id int, value json);"
         insert_query = """INSERT INTO test (id, value) VALUES (1, '{"miam": "🍔"}');"""
         event = self.create_and_insert_value(create_query, insert_query)
@@ -725,8 +707,6 @@ def test_json_unicode(self):
             )
 
     def test_json_long_string(self):
-        if not self.isMySQL57():
-            self.skipTest("Json is only supported in mysql 5.7")
         create_query = "CREATE TABLE test (id int, value json);"
         # The string length needs to be larger than what can fit in a single byte.
         string_value = "super_long_string" * 100
@@ -735,12 +715,31 @@ def test_json_long_string(self):
             % (string_value,)
         )
         event = self.create_and_insert_value(create_query, insert_query)
+        print(event.rows[0])
         if event.table_map[event.table_id].column_name_flag:
             self.assertEqual(
                 event.rows[0]["values"]["value"],
                 to_binary_dict({"my_key": string_value}),
             )
 
+    def test_json_deciaml_time_datetime(self):
+        create_query = """CREATE TABLE json_deciaml_time_datetime_test (
+                            id INT PRIMARY KEY AUTO_INCREMENT,
+                            json_data JSON
+                        );"""
+        insert_query = """
+                INSERT INTO json_deciaml_time_datetime_test (json_data) VALUES (JSON_OBJECT('time_key', CAST('18:54:12' AS TIME), 'datetime_key', CAST('2023-09-24 18:54:12' AS DATETIME) ,'decimal', CAST('99.99' AS DECIMAL(10, 2))));"""
+        event = self.create_and_insert_value(create_query, insert_query)
+        if event.table_map[event.table_id].column_name_flag:
+            self.assertEqual(
+                event.rows[0]["values"]["json_data"],
+                {
+                    b"decimal": Decimal("99.99"),
+                    b"time_key": datetime.time(18, 54, 12),
+                    b"datetime_key": datetime.datetime(2023, 9, 24, 18, 54, 12),
+                },
+            )
+
     def test_null(self):
         create_query = "CREATE TABLE test ( \
             test TINYINT NULL DEFAULT NULL, \

pymysqlreplication/util/bytes.py

@@ -27,18 +27,15 @@ def decode_variable_length(data: bytes):
     max_count = 5
     if len(data) < max_count:
         max_count = len(data)
-
     pos = 0
     length = 0
     for _ in range(max_count):
         v = data[pos]
         length |= (v & 0x7F) << (7 * pos)
-
+        pos += 1
         if v & 0x80 == 0:
             if length > sys.maxsize - 1:
                 return 0, 0
-
-            pos += 1
             return int(length), pos
 
     return 0, 0
@@ -103,11 +100,8 @@ def decode_time(data: bytes):
     if v == 0:
         return datetime.time(hour=0, minute=0, second=0)
 
-    sign = ""
     if v < 0:
-        sign = "-"
         v = -v
-
     int_part = v >> 24
     hour = (int_part >> 12) % (1 << 10)
     min = (int_part >> 6) % (1 << 6)