related issue: python-mysql-replication issue#380, pg_cameleon commit…

…#38c7a90

Analysis
- Could not parse status variables in query events for MariaDB
- Because the library was not aware of MariaDB-exclusive status variables

Implementation
- Added MariaDB-only status variables
- Added Exception to raise when status variable parse failed
- Added tests for parsing query events for MariaDB

pymysqlreplication/constants/STATUS_VAR_KEY.py

@@ -27,12 +27,14 @@
 Q_TABLE_MAP_FOR_UPDATE_CODE = 0x09
 Q_MASTER_DATA_WRITTEN_CODE = 0x0A
 Q_INVOKER = 0x0B
-Q_UPDATED_DB_NAMES = 0x0C
-Q_MICROSECONDS = 0x0D
+Q_UPDATED_DB_NAMES = 0x0C   # MySQL only
+Q_MICROSECONDS = 0x0D       # MySQL only
 Q_COMMIT_TS = 0x0E
 Q_COMMIT_TS2 = 0X0F
 Q_EXPLICIT_DEFAULTS_FOR_TIMESTAMP = 0X10
 Q_DDL_LOGGED_WITH_XID = 0X11
 Q_DEFAULT_COLLATION_FOR_UTF8MB4 = 0X12
 Q_SQL_REQUIRE_PRIMARY_KEY = 0X13
 Q_DEFAULT_TABLE_ENCRYPTION = 0X14
+Q_HRNOW = 0x80  # MariaDB only
+Q_XID = 0x81    # MariaDB only

pymysqlreplication/event.py

@@ -4,6 +4,7 @@
 import struct
 import datetime
 from pymysqlreplication.constants.STATUS_VAR_KEY import *
+from pymysqlreplication.exceptions import StatusVariableMismatch
 
 
 class BinLogEvent(object):
@@ -291,6 +292,12 @@ def _read_status_vars_value_for_key(self, key):
             self.sql_require_primary_key = self.packet.read_uint8()
         elif key == Q_DEFAULT_TABLE_ENCRYPTION:       # 0x14
             self.default_table_encryption = self.packet.read_uint8()
+        elif key == Q_HRNOW:
+            self.hrnow = self.packet.read_uint24()
+        elif key == Q_XID:
+            self.xid = self.packet.read_uint64()
+        else:
+            raise StatusVariableMismatch
 
 class BeginLoadQueryEvent(BinLogEvent):
     """

pymysqlreplication/exceptions.py

@@ -6,3 +6,14 @@ def __init__(self, table):
 class BinLogNotEnabled(Exception):
     def __init__(self):
         Exception.__init__(self, "MySQL binary logging is not enabled.")
+
+
+class StatusVariableMismatch(Exception):
+    def __init__(self):
+        Exception.__init__(self, " ".join(
+            "Unknown status variable in query event."
+            , "Possible parse failure in preceding fields"
+            , "or outdated constants.STATUS_VAR_KEY"
+            , "Refer to MySQL documentation/source code"
+            , "or create an issue on GitHub"
+        ))

pymysqlreplication/tests/base.py

@@ -43,6 +43,7 @@ def setUp(self):
         self.stream = None
         self.resetBinLog()
         self.isMySQL56AndMore()
+        self.__is_mariaDB = None
 
     def getMySQLVersion(self):
         """Return the MySQL version of the server
@@ -64,6 +65,11 @@ def isMySQL80AndMore(self):
         version = float(self.getMySQLVersion().rsplit('.', 1)[0])
         return version >= 8.0
 
+    def isMariaDB(self):
+        if self.__is_mariaDB is None:
+            self.__is_mariaDB = "MariaDB" in self.execute("SELECT VERSION()").fetchone()
+        return self.__is_mariaDB
+
     @property
     def supportsGTID(self):
         if not self.isMySQL56AndMore():

pymysqlreplication/tests/test_data_type.py

@@ -773,5 +773,31 @@ def test_null_bitmask(self):
         self.assertEqual(event.event_type, TABLE_MAP_EVENT)
         self.assertEqual(event.null_bitmask, bit_mask)
 
+    def test_mariadb_only_status_vars(self):
+        """Test parse of mariadb exclusive status variables (a field in query event)
+
+        A query event for mariadb must be parsed successfully
+        since mariadb exclusive status variables are now taken to account
+        (Q_HRNOW, Q_XID)
+        Test if was parse successful by asserting the last field of the event,
+        'SQL statement'.
+
+        Raises:
+            StatusVariableMismatch: This is the case where new status variables are added to
+            mysql server. Same set of status variables must be added to the library as well.
+        """
+        if not self.isMariaDB():
+            return
+
+        create_query = "CREATE TABLE test (id INTEGER)"
+        event = self.create_table(create_query)
+
+        # skip dummy events with empty schema
+        while event.schema == b'':
+            event = self.stream.fetchone()
+
+        self.assertEqual(event.query, create_query)
+
+
 if __name__ == "__main__":
     unittest.main()