Merge branch 'main' of github.com:23-OSSCA-python-mysql-replication/python-mysql-replication into feature/categorize-none-column

Author: heehehe

.github/workflows/pytest.yml

@@ -13,14 +13,14 @@ jobs:
           - {name: 'Pypy 3.9', python: 'pypy-3.9'}
     name: ${{ matrix.name }}
     runs-on: ubuntu-latest
-    timeout-minutes: 2
+    timeout-minutes: 3
 
     steps:
       - name: Check out code
-        uses: actions/checkout@v2
+        uses: actions/checkout@v4
 
       - name: Setup Python
-        uses: actions/setup-python@v2
+        uses: actions/setup-python@v4
         with:
           python-version: ${{ matrix.python }}
 

README.md

@@ -65,6 +65,8 @@ Featured
 
 [Near Zero Downtime Migration from MySQL to DynamoDB](https://aws.amazon.com/ko/blogs/big-data/near-zero-downtime-migration-from-mysql-to-dynamodb/) (by YongSeong Lee, Amazon Web Services)
 
+[Enable change data capture on Amazon RDS for MySQL applications that are using XA transactions](https://aws.amazon.com/ko/blogs/database/enable-change-data-capture-on-amazon-rds-for-mysql-applications-that-are-using-xa-transactions/) (by Baruch Assif, Amazon Web Services)
+
 Projects using this library
 ===========================
 

docker-compose-test.yml

@@ -37,6 +37,15 @@ services:
     networks:
       - default
 
+  percona-8.0:
+    <<: *mysql
+    image: percona:8.0
+    platform: linux/amd64
+    ports:
+      - "3309:3306"
+    networks:
+      - default
+
   mariadb-10.6:
     <<: *mariadb
     image: mariadb:10.6
@@ -61,6 +70,8 @@ services:
         MYSQL_5_7: percona-5.7
         MYSQL_5_7_CTL: percona-5.7-ctl
         MYSQL_5_7_CTL_PORT: 3306
+        MYSQL_8_0: percona-8.0
+        MYSQL_8_0_PORT: 3306
         MARIADB_10_6: mariadb-10.6
         MARIADB_10_6_PORT: 3306
 
@@ -72,7 +83,7 @@ services:
         
         while :
         do
-          if mysql -h percona-5.7 --user=root --execute "SELECT version();" 2>&1 >/dev/null && mysql -h percona-5.7-ctl --user=root --execute "SELECT version();" 2>&1 >/dev/null; then
+          if mysql -h percona-5.7 --user=root --execute "SELECT version();" 2>&1 >/dev/null && mysql -h percona-5.7-ctl --user=root --execute "SELECT version();" 2>&1 >/dev/null && mysql -h percona-8.0 --user=root --execute "SELECT version();" 2>&1 >/dev/null; then
             break
           fi
           sleep 1
@@ -87,6 +98,7 @@ services:
     depends_on:
       - percona-5.7
       - percona-5.7-ctl
+      - percona-8.0
 
 networks:
   default:

docker-compose.yml

@@ -33,6 +33,12 @@ services:
     ports:
       - "3307:3306"
 
+  percona-8.0:
+    <<: *mysql
+    image: percona:8.0
+    ports:
+      - "3309:3306"
+
   mariadb-10.6:
     <<: *mariadb
     image: mariadb:10.6

docs/conf.py

@@ -1,5 +1,3 @@
-# -*- coding: utf-8 -*-
-#
 # Python MySQL Replication documentation build configuration file, created by
 # sphinx-quickstart on Sun Sep 30 15:04:27 2012.
 #

examples/dump_events.py

@@ -1,5 +1,4 @@
 #!/usr/bin/env python
-# -*- coding: utf-8 -*-
 
 #
 # Dump all replication events from a remote mysql server

examples/logstash/mysql_to_logstash.py

@@ -1,5 +1,4 @@
 #!/usr/bin/env python
-# -*- coding: utf-8 -*-
 
 #
 # Output logstash events to the console from MySQL replication stream

examples/redis_cache.py

@@ -1,5 +1,4 @@
 #!/usr/bin/env python
-# -*- coding: utf-8 -*-
 
 #
 # Update a redis server cache when an evenement is trigger

pymysqlreplication/_compat.py

@@ -1,5 +1,3 @@
-# -*- coding: utf-8 -*-
-
 import struct
 import logging
 from distutils.version import LooseVersion

pymysqlreplication/binlogstream.py

@@ -1,5 +1,3 @@
-# -*- coding: utf-8 -*-
-
 bitCountInByte = [
     0,
     1,

pymysqlreplication/bitmap.py

@@ -1,5 +1,3 @@
-# -*- coding: utf-8 -*-
-
 import struct
 
 from .constants import FIELD_TYPE

pymysqlreplication/column.py

@@ -1,5 +1,3 @@
-# -*- coding: utf-8 -*-
-
 UNKNOWN_EVENT = 0x00
 START_EVENT_V3 = 0x01
 QUERY_EVENT = 0x02

pymysqlreplication/constants/BINLOG.py

@@ -1,5 +1,3 @@
-# -*- coding: utf-8 -*-
-
 # Original code from PyMySQL
 # Copyright (c) 2010 PyMySQL contributors
 #

pymysqlreplication/constants/FIELD_TYPE.py

@@ -1,5 +1,3 @@
-# -*- coding: utf-8 -*-
-
 from .BINLOG import *
 from .FIELD_TYPE import *
 from .STATUS_VAR_KEY import *

pymysqlreplication/constants/__init__.py

@@ -1,5 +1,3 @@
-# -*- coding: utf-8 -*-
-
 import binascii
 import struct
 import datetime

pymysqlreplication/event.py

@@ -1,5 +1,3 @@
-# -*- coding: utf-8 -*-
-
 import re
 import struct
 import binascii

pymysqlreplication/gtid.py

@@ -1,5 +1,3 @@
-# -*- coding: utf-8 -*-
-
 import struct
 
 from pymysqlreplication import constants, event, row_event

pymysqlreplication/packet.py

@@ -1,5 +1,3 @@
-# -*- coding: utf-8 -*-
-
 import struct
 import decimal
 import datetime

pymysqlreplication/row_event.py

@@ -1,6 +1,3 @@
-# -*- coding: utf-8 -*-
-
-
 class Table(object):
     def __init__(
         self, table_id, schema, table, columns, primary_key=None, column_name_flag=False

pymysqlreplication/table.py

@@ -1,12 +1,7 @@
-# -*- coding: utf-8 -*-
-
 from pymysqlreplication.tests.test_basic import *
 from pymysqlreplication.tests.test_data_type import *
 from pymysqlreplication.tests.test_data_objects import *
+import unittest
 
 if __name__ == "__main__":
-    if sys.version_info < (2, 7):
-        import unittest2 as unittest
-    else:
-        import unittest
     unittest.main()

pymysqlreplication/tests/__init__.py

@@ -1,15 +1,8 @@
-# -*- coding: utf-8 -*-
-
 import pymysql
 import copy
 from pymysqlreplication import BinLogStreamReader
 import os
-import sys
-
-if sys.version_info < (2, 7):
-    import unittest2 as unittest
-else:
-    import unittest
+import unittest
 
 base = unittest.TestCase
 
@@ -160,3 +153,31 @@ def bin_log_basename(self):
         bin_log_basename = cursor.fetchone()[0]
         bin_log_basename = bin_log_basename.split("/")[-1]
         return bin_log_basename
+
+
+class PyMySQLReplicationVersion8TestCase(PyMySQLReplicationTestCase):
+    def setUp(self):
+        super().setUp()
+        # default
+        self.database = {
+            "host": os.environ.get("MYSQL_8_0") or "localhost",
+            "user": "root",
+            "passwd": "",
+            "port": int(os.environ.get("MYSQL_8_0_PORT") or 3309),
+            "use_unicode": True,
+            "charset": "utf8",
+            "db": "pymysqlreplication_test",
+        }
+
+        self.conn_control = None
+        db = copy.copy(self.database)
+        db["db"] = None
+        self.connect_conn_control(db)
+        self.execute("DROP DATABASE IF EXISTS pymysqlreplication_test")
+        self.execute("CREATE DATABASE pymysqlreplication_test")
+        db = copy.copy(self.database)
+        self.connect_conn_control(db)
+        self.stream = None
+        self.resetBinLog()
+        self.isMySQL80AndMore()
+        self.__is_mariaDB = None

pymysqlreplication/tests/base.py

@@ -1,5 +1,3 @@
-# -*- coding: utf-8 -*-
-#
 # This is a sample script in order to make benchmark
 # on library speed.
 #

pymysqlreplication/tests/benchmark.py

@@ -1,4 +1,3 @@
-# -*- coding: utf-8 -*-
 """Test abnormal conditions, such as caused by a MySQL crash
 """
 import os.path

pymysqlreplication/tests/test_abnormal.py

@@ -1,16 +1,9 @@
-# -*- coding: utf-8 -*-
 import copy
 import io
 import os
-import sys
 import time
-
 import pymysql
-
-if sys.version_info < (2, 7):
-    import unittest2 as unittest
-else:
-    import unittest
+import unittest
 
 from pymysqlreplication.tests import base
 from pymysqlreplication import BinLogStreamReader
@@ -1479,7 +1472,7 @@ def test_query_event_latin1(self):
         assert event.query == r"CREATE TABLE test_latin1_\xd6\xc6\xdb (a INT)"
 
 
-class TestOptionalMetaData(base.PyMySQLReplicationTestCase):
+class TestOptionalMetaData(base.PyMySQLReplicationVersion8TestCase):
     def setUp(self):
         super(TestOptionalMetaData, self).setUp()
         self.stream.close()
@@ -1704,7 +1697,7 @@ def test_sync_drop_table_map_event_table_schema(self):
 
         event = self.stream.fetchone()
         self.assertIsInstance(event, TableMapEvent)
-        self.assertEqual(event.table_obj.data["columns"][0].name, "name")
+        self.assertEqual(event.table_obj.data["columns"][0].name, None)
         self.assertEqual(len(column_schemas), 0)
 
     def test_sync_column_drop_event_table_schema(self):
@@ -1735,9 +1728,9 @@ def test_sync_column_drop_event_table_schema(self):
         self.assertEqual(len(event.table_obj.data["columns"]), 3)
         self.assertEqual(column_schemas[0][0], "drop_column1")
         self.assertEqual(column_schemas[1][0], "drop_column3")
-        self.assertEqual(event.table_obj.data["columns"][0].name, "drop_column1")
-        self.assertEqual(event.table_obj.data["columns"][1].name, "drop_column2")
-        self.assertEqual(event.table_obj.data["columns"][2].name, "drop_column3")
+        self.assertEqual(event.table_obj.data["columns"][0].name, None)
+        self.assertEqual(event.table_obj.data["columns"][1].name, None)
+        self.assertEqual(event.table_obj.data["columns"][2].name, None)
 
     def tearDown(self):
         self.execute("SET GLOBAL binlog_row_metadata='MINIMAL';")

pymysqlreplication/tests/test_basic.py

@@ -1,10 +1,4 @@
-import sys
-
-if sys.version_info < (2, 7):
-    import unittest2 as unittest
-else:
-    import unittest
-
+import unittest
 from pymysqlreplication.column import Column
 from pymysqlreplication.table import Table
 from pymysqlreplication.event import GtidEvent

pymysqlreplication/tests/test_data_objects.py

@@ -1,32 +1,22 @@
-# -*- coding: utf-8 -*-
-
 import copy
 import platform
-import sys
 import json
-
 from pymysqlreplication import BinLogStreamReader
-
-if sys.version_info < (2, 7):
-    import unittest2 as unittest
-else:
-    import unittest
+import unittest
 
 from decimal import Decimal
 
 from pymysqlreplication.tests import base
 from pymysqlreplication.constants.BINLOG import *
 from pymysqlreplication.row_event import *
 from pymysqlreplication.event import *
-from pymysqlreplication._compat import text_type
 
-
-__all__ = ["TestDataType"]
+__all__ = ["TestDataType", "TestDataTypeVersion8"]
 
 
 def to_binary_dict(d):
     def encode_value(v):
-        if isinstance(v, text_type):
+        if isinstance(v, str):
             return v.encode()
         if isinstance(v, list):
             return [encode_value(x) for x in v]
@@ -954,5 +944,49 @@ def test_varbinary(self):
             self.assertEqual(event.rows[0]["values"]["b"], b"\xff\x01\x00\x00")
 
 
+class TestDataTypeVersion8(base.PyMySQLReplicationVersion8TestCase):
+    def ignoredEvents(self):
+        return [GtidEvent, PreviousGtidsEvent]
+
+    def create_and_insert_value(self, create_query, insert_query):
+        self.execute(create_query)
+        self.execute(insert_query)
+        self.execute("COMMIT")
+
+        self.assertIsInstance(self.stream.fetchone(), RotateEvent)
+        self.assertIsInstance(self.stream.fetchone(), FormatDescriptionEvent)
+        # QueryEvent for the Create Table
+        self.assertIsInstance(self.stream.fetchone(), QueryEvent)
+
+        # QueryEvent for the BEGIN
+        self.assertIsInstance(self.stream.fetchone(), QueryEvent)
+
+        self.assertIsInstance(self.stream.fetchone(), TableMapEvent)
+
+        event = self.stream.fetchone()
+        if self.isMySQL56AndMore():
+            self.assertEqual(event.event_type, WRITE_ROWS_EVENT_V2)
+        else:
+            self.assertEqual(event.event_type, WRITE_ROWS_EVENT_V1)
+        self.assertIsInstance(event, WriteRowsEvent)
+        return event
+
+    def test_partition_id(self):
+        if not self.isMySQL80AndMore():
+            self.skipTest("Not supported in this version of MySQL")
+        create_query = "CREATE TABLE test (id INTEGER) \
+            PARTITION BY RANGE (id) ( \
+                PARTITION p0 VALUES LESS THAN (1),   \
+                PARTITION p1 VALUES LESS THAN (2),   \
+                PARTITION p2 VALUES LESS THAN (3),   \
+                PARTITION p3 VALUES LESS THAN (4),   \
+                PARTITION p4 VALUES LESS THAN (5)    \
+            )"
+        insert_query = "INSERT INTO test (id) VALUES(3)"
+        event = self.create_and_insert_value(create_query, insert_query)
+        self.assertEqual(event.extra_data_type, 1)
+        self.assertEqual(event.partition_id, 3)
+
+
 if __name__ == "__main__":
     unittest.main()