check possible optional metadata version And delete get Table information

Author: sean-k1

pymysqlreplication/binlogstream.py

@@ -290,7 +290,6 @@ def close(self):
         if self.__connected_ctl:
             # break reference cycle between stream reader and underlying
             # mysql connection object
-            self._ctl_connection._get_table_information = None
             self._ctl_connection.close()
             self.__connected_ctl = False
 
@@ -301,9 +300,9 @@ 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_table_information = self.__get_table_information
         self._ctl_connection._get_dbms = self.__get_dbms
         self.__connected_ctl = True
+        self.__check_optional_meta_data()
 
     def __checksum_enabled(self):
         """Return True if binlog-checksum = CRC32. Only for MySQL > 5.6"""
@@ -548,6 +547,21 @@ def __set_mariadb_settings(self):
 
         return prelude
 
+    def __check_optional_meta_data(self):
+        cur = self._ctl_connection.cursor()
+        cur.execute("SHOW VARIABLES LIKE 'BINLOG_ROW_METADATA';")
+        value = cur.fetchone()
+        if value is None:
+            logging.log(logging.WARN, """
+                    Before using MARIADB 10.5.0 and MYSQL 8.0.14 versions,
+                    use python-mysql-replication version Before 1.0 version """)
+        value = value.get("Value", "")
+        if value.upper() != "FULL":
+            logging.log(logging.WARN, """
+                   Setting The Variable Value BINLOG_ROW_METADATA = FULL 
+                   By Applying this, provide properly mapped column information on UPDATE,DELETE,INSERT. 
+                    """)
+
     def fetchone(self):
         while True:
             if self.end_log_pos and self.is_past_end_log_pos:
@@ -709,37 +723,37 @@ def _allowed_event_list(
                 pass
         return frozenset(events)
 
-    def __get_table_information(self, schema, table):
-        for i in range(1, 3):
-            try:
-                if not self.__connected_ctl:
-                    self.__connect_to_ctl()
-
-                cur = self._ctl_connection.cursor()
-                cur.execute(
-                    """
-                    SELECT
-                        COLUMN_NAME, COLLATION_NAME, CHARACTER_SET_NAME,
-                        COLUMN_COMMENT, COLUMN_TYPE, COLUMN_KEY, ORDINAL_POSITION,
-                        DATA_TYPE, CHARACTER_OCTET_LENGTH
-                    FROM
-                        information_schema.columns
-                    WHERE
-                        table_schema = %s AND table_name = %s
-                    """,
-                    (schema, table),
-                )
-                result = sorted(cur.fetchall(), key=lambda x: x["ORDINAL_POSITION"])
-                cur.close()
-
-                return result
-            except pymysql.OperationalError as error:
-                code, message = error.args
-                if code in MYSQL_EXPECTED_ERROR_CODES:
-                    self.__connected_ctl = False
-                    continue
-                else:
-                    raise error
+    # def __get_table_information(self, schema, table):
+    #     for i in range(1, 3):
+    #         try:
+    #             if not self.__connected_ctl:
+    #                 self.__connect_to_ctl()
+    #
+    #             cur = self._ctl_connection.cursor()
+    #             cur.execute(
+    #                 """
+    #                 SELECT
+    #                     COLUMN_NAME, COLLATION_NAME, CHARACTER_SET_NAME,
+    #                     COLUMN_COMMENT, COLUMN_TYPE, COLUMN_KEY, ORDINAL_POSITION,
+    #                     DATA_TYPE, CHARACTER_OCTET_LENGTH
+    #                 FROM
+    #                     information_schema.columns
+    #                 WHERE
+    #                     table_schema = %s AND table_name = %s
+    #                 """,
+    #                 (schema, table),
+    #             )
+    #             result = sorted(cur.fetchall(), key=lambda x: x["ORDINAL_POSITION"])
+    #             cur.close()
+    #
+    #             return result
+    #         except pymysql.OperationalError as error:
+    #             code, message = error.args
+    #             if code in MYSQL_EXPECTED_ERROR_CODES:
+    #                 self.__connected_ctl = False
+    #                 continue
+    #             else:
+    #                 raise error
 
     def __get_dbms(self):
         if not self.__connected_ctl: