re-added version retrieval from session

Author: roy0424

pymysqlreplication/binlogstream.py

@@ -305,6 +305,7 @@ def __connect_to_ctl(self):
         self._ctl_connection_settings["cursorclass"] = DictCursor
         self._ctl_connection_settings["autocommit"] = True
         self._ctl_connection = self.pymysql_wrapper(**self._ctl_connection_settings)
+        self._ctl_connection._get_dbms = self.__get_dbms
         self.__connected_ctl = True
         self.__check_optional_meta_data()
 
@@ -744,6 +745,19 @@ def _allowed_event_list(
                 pass
         return frozenset(events)
 
+    def __get_dbms(self):
+        if not self.__connected_ctl:
+            self.__connect_to_ctl()
+
+        cur = self._ctl_connection.cursor()
+        cur.execute("SELECT VERSION();")
+
+        version_info = cur.fetchone().get("VERSION()", "")
+
+        if "MariaDB" in version_info:
+            return "mariadb"
+        return "mysql"
+
     def __log_valid_parameters(self):
         ignored = ["allowed_events", "table_map"]
         for parameter, value in self.__dict__.items():

pymysqlreplication/row_event.py

@@ -782,6 +782,7 @@ def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs)
         self.column_count = self.packet.read_length_coded_binary()
 
         self.columns = []
+        self.dbms = self._ctl_connection._get_dbms()
         # Read columns meta data
         column_types = bytearray(self.packet.read(self.column_count))
         self.packet.read_length_coded_binary()