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

Author: dongwook-chan

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()