Merge branch 'main' of github.com:noplay/python-mysql-replication

Author: julien-duponchelle

.github/workflows/pytest.yml

@@ -0,0 +1,33 @@
+name: PyTest
+on: [push, pull_request]
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    timeout-minutes: 2
+
+    steps:
+      - name: Check out code
+        uses: actions/checkout@v2
+
+      - name: Setup Python
+        uses: actions/setup-python@v2
+        with:
+          python-version: "3.7"
+
+      - name: Run database server in docker
+        run: |
+          docker compose create
+          docker compose start
+          # Wait for the services to accept connections,
+          # TODO: do that smarter, poll connection attempt until it succeeds
+          sleep 30
+
+      - name: Install dependencies
+        run: |
+          pip install .
+          pip install pytest
+
+      - name: Run test suite
+        run: |
+          pytest -k "not test_no_trailing_rotate_event and not test_end_log_pos"

docker-compose.yml

@@ -0,0 +1,17 @@
+version: '3.2'
+services:
+  percona-5.7:
+    image: percona:5.7
+    environment:
+      MYSQL_ALLOW_EMPTY_PASSWORD: true
+    ports:
+      - 3306:3306
+    command: mysqld --log-bin=mysql-bin.log --server-id 1 --binlog-format=row --gtid_mode=on --enforce-gtid-consistency=on --log_slave_updates
+
+  percona-5.7-ctl:
+    image: percona:5.7
+    environment:
+      MYSQL_ALLOW_EMPTY_PASSWORD: true
+    ports:
+      - 3307:3307
+    command: mysqld --log-bin=mysql-bin.log --server-id 1 --binlog-format=row --gtid_mode=on --enforce-gtid-consistency=on --log_slave_updates -P 3307

pymysqlreplication/packet.py

@@ -349,6 +349,9 @@ def unpack_int32(self, n):
 
     def read_binary_json(self, size):
         length = self.read_uint_by_size(size)
+        if length == 0:
+            # handle NULL value
+            return None
         payload = self.read(length)
         self.unread(payload)
         t = self.read_uint8()
@@ -402,9 +405,9 @@ def read_binary_json_type_inlined(self, t, large):
         elif t == JSONB_TYPE_UINT16:
             return self.read_uint32() if large else self.read_uint16()
         elif t == JSONB_TYPE_INT32:
-            return self.read_int64() if large else self.read_int32()
+            return self.read_int32()
         elif t == JSONB_TYPE_UINT32:
-            return self.read_uint64() if large else self.read_uint32()
+            return self.read_uint32()
 
         raise ValueError('Json type %d is not handled' % t)
 

pymysqlreplication/row_event.py

@@ -85,7 +85,8 @@ def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs)
             if self._fail_on_table_metadata_unavailable:
                 raise TableMetadataUnavailableError(self.table)
 
-    def __is_null(self, null_bitmap, position):
+    @staticmethod
+    def _is_null(null_bitmap, position):
         bit = null_bitmap[int(position / 8)]
         if type(bit) is str:
             bit = ord(bit)
@@ -113,7 +114,7 @@ def _read_column_data(self,  cols_bitmap):
                 values[name] = None
                 continue
 
-            if self.__is_null(null_bitmap, nullBitmapIndex):
+            if self._is_null(null_bitmap, nullBitmapIndex):
                 values[name] = None
             elif column.type == FIELD_TYPE.TINY:
                 if unsigned:
@@ -649,7 +650,7 @@ def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs)
                                self.table, self.columns)
 
         # ith column is nullable if (i - 1)th bit is set to True, not nullable otherwise
-        ## Refer to definition of and call to row.event.__is_null() to interpret bitmap corresponding to columns
+        ## Refer to definition of and call to row.event._is_null() to interpret bitmap corresponding to columns
         self.null_bitmask = self.packet.read((self.column_count + 7) / 8)
 
     def get_table(self):

pymysqlreplication/tests/test_basic.py

@@ -25,9 +25,9 @@ def ignoredEvents(self):
         return [GtidEvent]
 
     def test_allowed_event_list(self):
-        self.assertEqual(len(self.stream._allowed_event_list(None, None, False)), 14)
-        self.assertEqual(len(self.stream._allowed_event_list(None, None, True)), 13)
-        self.assertEqual(len(self.stream._allowed_event_list(None, [RotateEvent], False)), 13)
+        self.assertEqual(len(self.stream._allowed_event_list(None, None, False)), 15)
+        self.assertEqual(len(self.stream._allowed_event_list(None, None, True)), 14)
+        self.assertEqual(len(self.stream._allowed_event_list(None, [RotateEvent], False)), 14)
         self.assertEqual(len(self.stream._allowed_event_list([RotateEvent], None, False)), 1)
 
     def test_read_query_event(self):

pymysqlreplication/tests/test_data_type.py

@@ -513,6 +513,16 @@ def test_json_large(self):
 
         self.assertEqual(event.rows[0]["values"]["value"], to_binary_dict(data))
 
+    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');" % (json.dumps(large_array),)
+        event = self.create_and_insert_value(create_query, insert_query)
+        self.assertEqual(event.rows[0]["values"]["value"], to_binary_dict(large_array))
+
     def test_json_large_with_literal(self):
         if not self.isMySQL57():
             self.skipTest("Json is only supported in mysql 5.7")
@@ -730,7 +740,7 @@ def test_null_bitmask(self):
             column_type = "INT"
             column_definition.append(column_type)
 
-            nullability = "NOT NULL" if not RowsEvent.__is_null(bit_mask, i) else ""
+            nullability = "NOT NULL" if not RowsEvent._is_null(bit_mask, i) else ""
             column_definition.append(nullability)
 
             columns.append(" ".join(column_definition))