Parse status variables in query event (julien-duponchelle#360)

* Parse status variables in query event packet Status variables contain useful information: 1. List of databases affected by the query 2. Runtime value of system variables (SQL_MODE, AUTOCOMMIT, CHARSET_SERVER, etc.) 3. timezone of the master For further information please refer to; [Syntax of system variables field in query event](https://dev.mysql.com/doc/internals/en/query-event.html#q-microseconds) [Definition of enumeration for system variable keys](https://github.com/mysql/mysql-server/blob/beb865a960b9a8a16cf999c323e46c5b0c67f21f/libbinlogevents/include/statement_events.h#L463-L532) [Semantics of system variable values](https://github.com/mysql/mysql-server/blob/beb865a960b9a8a16cf999c323e46c5b0c67f21f/libbinlogevents/include/statement_events.h#L156-L448)

Author: dongwook-chan

pymysqlreplication/constants/STATUS_VAR_KEY.py

@@ -0,0 +1,38 @@
+#from enum import IntEnum
+
+#class StatusVarsKey(IntEnum):
+"""List of Query_event_status_vars
+ 
+ A status variable in query events is a sequence of status KEY-VALUE pairs.
+ The class variables enumerated below are KEYs.
+ Each KEY determines the length of corresponding VALUE.
+
+ For further details refer to:
+ mysql-server: https://github.com/mysql/mysql-server/blob/beb865a960b9a8a16cf999c323e46c5b0c67f21f/libbinlogevents/include/statement_events.h#L463-L532
+ MySQL Documentation: https://dev.mysql.com/doc/internals/en/query-event.html
+
+ Status variable key names From mysql-server source code, edited by dongwook-chan
+"""
+
+# KEY
+Q_FLAGS2_CODE = 0x00
+Q_SQL_MODE_CODE = 0X01
+Q_CATALOG_CODE = 0x02
+Q_AUTO_INCREMENT = 0x03
+Q_CHARSET_CODE = 0x04
+Q_TIME_ZONE_CODE = 0x05
+Q_CATALOG_NZ_CODE = 0x06
+Q_LC_TIME_NAMES_CODE = 0x07
+Q_CHARSET_DATABASE_CODE = 0x08
+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_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

pymysqlreplication/constants/__init__.py

@@ -2,3 +2,4 @@
 
 from .BINLOG import *
 from .FIELD_TYPE import *
+from .STATUS_VAR_KEY import *

pymysqlreplication/event.py

@@ -3,6 +3,7 @@
 import binascii
 import struct
 import datetime
+from pymysqlreplication.constants.STATUS_VAR_KEY import *
 
 
 class BinLogEvent(object):
@@ -167,7 +168,13 @@ def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs)
  self.status_vars_length = self.packet.read_uint16()
 
  # Payload
- self.status_vars = self.packet.read(self.status_vars_length)
+ status_vars_end_pos = self.packet.read_bytes + self.status_vars_length
+ while self.packet.read_bytes < status_vars_end_pos: # while 남은 data length가 얼마만큼? OR read_bytes
+ # read KEY for status variable
+ status_vars_key = self.packet.read_uint8()
+ # read VALUE for status variable
+ self._read_status_vars_value_for_key(status_vars_key)
+
  self.schema = self.packet.read(self.schema_length)
  self.packet.advance(1)
 
@@ -181,6 +188,80 @@ def _dump(self):
  print("Execution time: %d" % (self.execution_time))
  print("Query: %s" % (self.query))
 
+ 
+ # TODO: check if instance attribute with the same name already exists
+ # TODO: put all the instace attribute in separate class? called status_vars
+ # TODO: does length need to be remembered?
+ # TODO: ref(mysql doc. and mysql-server) for each hunk
+ def _read_status_vars_value_for_key(self, key):
+ """parse status variable VALUE for given KEY
+
+ A status variable in query events is a sequence of status KEY-VALUE pairs.
+ Parsing logic from mysql-server source code edited by dongwook-chan
+ https://github.com/mysql/mysql-server/blob/beb865a960b9a8a16cf999c323e46c5b0c67f21f/libbinlogevents/src/statement_events.cpp#L181-L336
+
+ Args:
+ key: key for status variable
+ """
+ if key == Q_FLAGS2_CODE: # 0x00
+ self.flags2 = self.packet.read_uint32()
+ elif key == Q_SQL_MODE_CODE: # 0x01
+ self.sql_mode = self.packet.read_uint64()
+ elif key == Q_CATALOG_CODE: # 0x02 for MySQL 5.0.x
+ pass
+ elif key == Q_AUTO_INCREMENT: # 0x03
+ self.auto_increment_increment = self.packet.read_uint16()
+ self.auto_increment_offset = self.packet.read_uint16()
+ elif key == Q_CHARSET_CODE: # 0x04
+ self.character_set_client = self.packet.read_uint16()
+ self.collation_connection = self.packet.read_uint16()
+ self.collation_server = self.packet.read_uint16()
+ elif key == Q_TIME_ZONE_CODE: # 0x05
+ time_zone_len = self.packet.read_uint8()
+ if time_zone_len:
+ self.time_zone = self.packet.read(time_zone_len) 
+ elif key == Q_CATALOG_NZ_CODE: # 0x06
+ catalog_len = self.packet.read_uint8()
+ if catalog_len:
+ self.catalog_nz_code = self.packet.read(catalog_len)
+ elif key == Q_LC_TIME_NAMES_CODE: # 0x07
+ self.lc_time_names_number = self.packet.read_uint16()
+ elif key == Q_CHARSET_DATABASE_CODE: # 0x08
+ self.charset_database_number = self.packet.read_uint16()
+ elif key == Q_TABLE_MAP_FOR_UPDATE_CODE: # 0x09
+ self.table_map_for_update = self.packet.read_uint64()
+ elif key == Q_MASTER_DATA_WRITTEN_CODE: # 0x0A
+ pass
+ elif key == Q_INVOKER: # 0x0B
+ user_len = self.packet.read_uint8()
+ if user_len:
+ self.user = self.packet.read(user_len)
+ host_len = self.packet.read_uint8()
+ if host_len:
+ self.host = self.packet.read(host_len)
+ elif key == Q_UPDATED_DB_NAMES: # 0x0C
+ mts_accessed_dbs = self.packet.read_uint8()
+ dbs = []
+ for i in range(mts_accessed_dbs):
+ db = self.packet.read_string()
+ dbs.append(db)
+ self.mts_accessed_db_names = dbs
+ elif key == Q_MICROSECONDS: # 0x0D
+ self.microseconds = self.packet.read_uint24()
+ elif key == Q_COMMIT_TS: # 0x0E
+ pass
+ elif key == Q_COMMIT_TS2: # 0x0F
+ pass
+ elif key == Q_EXPLICIT_DEFAULTS_FOR_TIMESTAMP:# 0x10
+ self.explicit_defaults_ts = self.packet.read_uint8()
+ elif key == Q_DDL_LOGGED_WITH_XID: # 0x11
+ self.ddl_xid = self.packet.read_uint64()
+ elif key == Q_DEFAULT_COLLATION_FOR_UTF8MB4: # 0x12
+ self.default_collation_for_utf8mb4_number = self.packet.read_uint16()
+ elif key == Q_SQL_REQUIRE_PRIMARY_KEY: # 0x13
+ self.sql_require_primary_key = self.packet.read_uint8()
+ elif key == Q_DEFAULT_TABLE_ENCRYPTION: # 0x14
+ self.default_table_encryption = self.packet.read_uint8()
 
 class BeginLoadQueryEvent(BinLogEvent):
  """

pymysqlreplication/packet.py

@@ -461,3 +461,20 @@ def _read(x):
  return self.read_binary_json_type(x[0], length)
 
  return [_read(x) for x in values_type_offset_inline]
+
+ def read_string(self):
+ """Read a 'Length Coded String' from the data buffer.
+
+ Read __data_buffer until NULL character (0 = \0 = \x00)
+
+ Returns:
+ Binary string parsed from __data_buffer
+ """
+ string = b''
+ while True:
+ char = self.read(1)
+ if char == b'\0':
+ break
+ string += char
+
+ return string

pymysqlreplication/tests/test_data_type.py

@@ -57,6 +57,26 @@ def create_and_insert_value(self, create_query, insert_query):
  self.assertIsInstance(event, WriteRowsEvent)
  return event
 
+ def create_table(self, create_query):
+ """Create table 
+
+ Create table in db and return query event.
+
+ Returns:
+ Query event
+ """
+
+ self.execute(create_query)
+
+ self.assertIsInstance(self.stream.fetchone(), RotateEvent)
+ self.assertIsInstance(self.stream.fetchone(), FormatDescriptionEvent)
+
+ event = self.stream.fetchone()
+
+ self.assertEqual(event.event_type, QUERY_EVENT)
+
+ return event
+
  def test_decimal(self):
  create_query = "CREATE TABLE test (test DECIMAL(2,1))"
  insert_query = "INSERT INTO test VALUES(4.2)"
@@ -641,5 +661,24 @@ def test_partition_id(self):
  self.assertEqual(event.extra_data_type, 1)
  self.assertEqual(event.partition_id, 3)
 
+ def test_status_vars(self):
+ """Test parse of status variables in query events
+
+ Majority of status variables available depends on the settings of db.
+ Therefore, this test only tests system variable values independent from settings of db.
+ Note that if you change default db name 'pymysqlreplication_test',
+ event.mts_accessed_db_names MUST be asserted against the changed db name.
+
+ Returns:
+ binary string parsed from __data_buffer
+
+ Raises:
+ AssertionError: if no 
+ """
+ create_query = "CREATE TABLE test (id INTEGER)"
+ event = self.create_table(create_query)
+ self.assertEqual(event.catalog_nz_code, b'std')
+ self.assertEqual(event.mts_accessed_db_names, [b'pymysqlreplication_test'])
+
 if __name__ == "__main__":
  unittest.main()