Mysql 8.0.14 version support Table map Event optional metaData (#446)

* feat: optional meta data

* feat: optional meta data implement

* feat: numeric_count , count parsing add

* refactor : python code lint

* Fix optional meta data over-read

Previously, `get_optional_meta_data` attempt to read trailing 4 bytes
reserved for Common-Footer.

Changes:
- Added `bytes_to_read` to indicate remaining bytes (including 4 bytes)
- Updated `get_optional_meta_data` to read everything but 4 bytes

* feat: fix bug signedness order

* refactor : name changed

* feat: fix bug if column length > =8

* feat : dump log add

* refactor : read tlv format

* feat: add _read_primary_keys_with_prefix & updates

* refactor: field_type -> MetadataFieldType

* refactor: fix to store optional_metadata as a variable

* Test: Add test_set_str_value, test_enum_str_value

* fix: fix dump location of optional_metadata

* Test: Add primary keys simple, prefix

* Test: Add signedness

* fix: fix enum_and_set_default_column_charset_list from dict to list

* Test: add percona:8.0.14 for optional metadata

* fix: percona version change to 8.0

* test: add isMySQL80AndMore in optional metadata test

* test: add isMySQL8014AndMore

* fix: fix binlog_row_metadata setting location

* test: add column_charset

* Test: add default_charset

* test: add mysql8 for optional metadata

* Test: add column_name and refactor other tests

* fix: modify isMySQL8014AndMore

* refactoring , parsing column

* fix: test pass

* fix: json name error

* feat : sync column

* feat: sync column data first version

* feat: add _get_field_type_key

* feat : add charset

* feat: get column_schema info from column info

* schema default value change

* refactor : code convention

* restore column schema from  optional meta data

* delete print

* enum refactoring

* refactor delete unused code

* fix: reverse_field_type & remove None from dump

* feat: add mariadb charset

* fix : decode error ignore unknown decode type

* only optional metadata info

* Revert "only optional metadata info "

This reverts commit 59402f9.

* ignore decode error

* force encode utf-8 type

* error packet goes on

* remove: delete mysql8-related settings

* test: add geometry, enum_and_set, visibility testcases

* fix : Adding the missing column.

* fix: charset error

* refactor: variable name and protect None 

-refactor variable name 
-bug fix None lower()

* test: test add when table dropped

* fix : restore column schema from  optional meta data

* fix: protect when database column schema length and column_count are same

* refactor: add charset_list.csv for adding charsets

* fix: modify .extract_charset_list.sh typo

* test add : when alter drop column case

* test add column comment dropped

* fix: test case drop table example

* fix: modify process of reading charset_list.csv

* setting BinlogStream class parameter optional_meta_data

* Revert "setting BinlogStream class parameter optional_meta_data "

This reverts commit b882122.

* Revert "Revert "setting BinlogStream class parameter optional_meta_data "" resolve conflict

This reverts commit ee27a5f.

* resolve  conflict test case error

* resolve conflict missing

resolve conflict missing

* black lint

* remove testcase

* remove set, enum

* fix: testcase

fix : testcase

fix testcase

fix: testcase

* table init changed

table init changed

* remove ununsed variable

* check possible optional metadata version And delete get Table information

* cherry pick from column_schema delete

* Column Values add

* sync column

* TestCase resolved

test case resolved

Test Case resolved

* delete print debug and print optional meta data conditional

* docker test 8.0

* visibility None case

* for visibility test

* Column read String Mysql 5version

* Revert "docker test 8.0"

This reverts commit b9a2b66.

* fix : testcase 8 version

fix testcase

fix test case

testcase fix

* Add when column name list length 0

It measns  Now BINLOG_ROW_METADATA = FULL
but still remain  BINLOG_ROW_METADATA Mode = MINIMAL in Binlog

* mysql 8.0.23 env Test

* Revert "mysql 8.0.23 env Test"

This reverts commit 30b5fe2.

* enum catch out of index

enum when Mysql 5.7 case error

enum and set BINLOG_IMAGE = MINIMAL

erase print

---------

Co-authored-by: dongwook-chan <[email protected]>
Co-authored-by: mjs <[email protected]>
Co-authored-by: mikaniz <[email protected]>
Co-authored-by: starcat37 <[email protected]>
Co-authored-by: heehehe <[email protected]>

Author: 6 people

pymysqlreplication/binlogstream.py

@@ -176,7 +176,6 @@ def __init__(
         report_slave=None,
         slave_uuid=None,
         pymysql_wrapper=None,
-        fail_on_table_metadata_unavailable=False,
         slave_heartbeat=None,
         is_mariadb=False,
         annotate_rows_event=False,
@@ -210,9 +209,6 @@ def __init__(
             report_slave: Report slave in SHOW SLAVE HOSTS.
             slave_uuid: Report slave_uuid or replica_uuid in SHOW SLAVE HOSTS(MySQL 8.0.21-) or
                         SHOW REPLICAS(MySQL 8.0.22+) depends on your MySQL version.
-            fail_on_table_metadata_unavailable: Should raise exception if we
-                                                can't get table information on
-                                                row_events
             slave_heartbeat: (seconds) Should master actively send heartbeat on
                              connection. This also reduces traffic in GTID
                              replication on replication resumption (in case
@@ -249,9 +245,9 @@ def __init__(
         self.__allowed_events = self._allowed_event_list(
             only_events, ignored_events, filter_non_implemented_events
         )
-        self.__fail_on_table_metadata_unavailable = fail_on_table_metadata_unavailable
         self.__ignore_decode_errors = ignore_decode_errors
         self.__verify_checksum = verify_checksum
+        self.__optional_meta_data = False
 
         # We can't filter on packet level TABLE_MAP and rotate event because
         # we need them for handling other operations
@@ -295,7 +291,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
 
@@ -306,9 +301,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"""
@@ -553,6 +548,30 @@ 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:  # BinLog Variable Not exist It means Not Supported Version
+            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 """,
+            )
+        else:
+            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. 
+                        """,
+                )
+            else:
+                self.__optional_meta_data = True
+
     def fetchone(self):
         while True:
             if self.end_log_pos and self.is_past_end_log_pos:
@@ -596,9 +615,9 @@ def fetchone(self):
                 self.__only_schemas,
                 self.__ignored_schemas,
                 self.__freeze_schema,
-                self.__fail_on_table_metadata_unavailable,
                 self.__ignore_decode_errors,
                 self.__verify_checksum,
+                self.__optional_meta_data,
             )
 
             if binlog_event.event_type == ROTATE_EVENT:
@@ -715,44 +734,13 @@ 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_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:

pymysqlreplication/column.py

@@ -9,27 +9,22 @@ class Column(object):
     """Definition of a column"""
 
     def __init__(self, *args, **kwargs):
-        if len(args) == 3:
+        if len(args) == 2:
             self.__parse_column_definition(*args)
         else:
             self.__dict__.update(kwargs)
 
-    def __parse_column_definition(self, column_type, column_schema, packet):
+    def __parse_column_definition(self, column_type, packet):
         self.type = column_type
-        self.name = column_schema["COLUMN_NAME"]
-        self.collation_name = column_schema["COLLATION_NAME"]
-        self.character_set_name = column_schema["CHARACTER_SET_NAME"]
-        self.comment = column_schema["COLUMN_COMMENT"]
-        self.unsigned = column_schema["COLUMN_TYPE"].find("unsigned") != -1
-        self.zerofill = column_schema["COLUMN_TYPE"].find("zerofill") != -1
-        self.type_is_bool = False
-        self.is_primary = column_schema["COLUMN_KEY"] == "PRI"
-
-        # Check for fixed-length binary type. When that's the case then we need
-        # to zero-pad the values to full length at read time.
-        self.fixed_binary_length = None
-        if column_schema["DATA_TYPE"] == "binary":
-            self.fixed_binary_length = column_schema["CHARACTER_OCTET_LENGTH"]
+        self.name = None
+        self.unsigned = False
+        self.is_primary = False
+        self.charset_id = None
+        self.character_set_name = None
+        self.collation_name = None
+        self.enum_values = None
+        self.set_values = None
+        self.visibility = False
 
         if self.type == FIELD_TYPE.VARCHAR:
             self.max_length = struct.unpack("<H", packet.read(2))[0]
@@ -43,13 +38,8 @@ def __parse_column_definition(self, column_type, column_schema, packet):
             self.fsp = packet.read_uint8()
         elif self.type == FIELD_TYPE.TIME2:
             self.fsp = packet.read_uint8()
-        elif (
-            self.type == FIELD_TYPE.TINY
-            and column_schema["COLUMN_TYPE"] == "tinyint(1)"
-        ):
-            self.type_is_bool = True
         elif self.type == FIELD_TYPE.VAR_STRING or self.type == FIELD_TYPE.STRING:
-            self.__read_string_metadata(packet, column_schema)
+            self.__read_string_metadata(packet)
         elif self.type == FIELD_TYPE.BLOB:
             self.length_size = packet.read_uint8()
         elif self.type == FIELD_TYPE.GEOMETRY:
@@ -65,27 +55,15 @@ def __parse_column_definition(self, column_type, column_schema, packet):
             self.bits = (bytes * 8) + bits
             self.bytes = int((self.bits + 7) / 8)
 
-    def __read_string_metadata(self, packet, column_schema):
+    def __read_string_metadata(self, packet):
         metadata = (packet.read_uint8() << 8) + packet.read_uint8()
         real_type = metadata >> 8
         if real_type == FIELD_TYPE.SET or real_type == FIELD_TYPE.ENUM:
             self.type = real_type
             self.size = metadata & 0x00FF
-            self.__read_enum_metadata(column_schema)
         else:
             self.max_length = (((metadata >> 4) & 0x300) ^ 0x300) + (metadata & 0x00FF)
 
-    def __read_enum_metadata(self, column_schema):
-        enums = column_schema["COLUMN_TYPE"]
-        if self.type == FIELD_TYPE.ENUM:
-            self.enum_values = [""] + enums.replace("enum(", "").replace(
-                ")", ""
-            ).replace("'", "").split(",")
-        else:
-            self.set_values = (
-                enums.replace("set(", "").replace(")", "").replace("'", "").split(",")
-            )
-
     def __eq__(self, other):
         return self.data == other.data
 

pymysqlreplication/constants/.extract_charset_list.sh

@@ -0,0 +1,24 @@
+#!/usr/bin/env bash
+
+usage(){
+    echo "Usage: bash .extract_charset_list.sh (mysql|mariadb) >> charset_list.csv"
+}
+
+dbms=$1
+if [ -z "$dbms" ]; then
+    usage
+    exit 1
+fi
+
+SQL_QUERY="SELECT id, character_set_name, collation_name, is_default
+FROM information_schema.collations ORDER BY id;"
+
+mysql -N -s -e "$SQL_QUERY" | python3 -c "import sys
+dbms = sys.argv[1]
+for line in sys.stdin:
+    _id, name, collation, is_default = line.split(chr(9))
+    if _id == 'NULL':
+        continue
+    is_default = True if is_default.strip() == 'Yes' else False
+    print(f'{_id},{name},{collation},{is_default},{dbms}')
+" "$dbms"

pymysqlreplication/constants/BINLOG.py

@@ -44,11 +44,13 @@
 INTVAR_INSERT_ID_EVENT = 0x02
 
 # MariaDB events
+
 MARIADB_ANNOTATE_ROWS_EVENT = 0xA0
 MARIADB_BINLOG_CHECKPOINT_EVENT = 0xA1
 MARIADB_GTID_EVENT = 0xA2
 MARIADB_GTID_GTID_LIST_EVENT = 0xA3
 MARIADB_START_ENCRYPTION_EVENT = 0xA4
 
+
 # Common-Footer
 BINLOG_CHECKSUM_LEN = 4

pymysqlreplication/constants/CHARSET.py

@@ -0,0 +1,67 @@
+from collections import defaultdict
+import os
+
+
+class Charset:
+    def __init__(self, id, name, collation, is_default=False, dbms="mysql"):
+        self.id, self.name, self.collation = id, name, collation
+        self.is_default = is_default
+        self.dbms = dbms
+
+    def __repr__(self):
+        return (
+            f"Charset(id={self.id}, name={self.name!r}, collation={self.collation!r})"
+        )
+
+    @property
+    def encoding(self):
+        name = self.name
+        if name in ("utf8mb4", "utf8mb3"):
+            return "utf8"
+        if name == "latin1":
+            return "cp1252"
+        if name == "koi8r":
+            return "koi8_r"
+        if name == "koi8u":
+            return "koi8_u"
+        return name
+
+    @property
+    def is_binary(self):
+        return self.id == 63
+
+
+class Charsets:
+    def __init__(self):
+        self._by_id = defaultdict(dict)  # key: mysql / mariadb
+        self._by_name = defaultdict(dict)  # key: mysql / mariadb
+
+    def add(self, _charset):
+        self._by_id[_charset.dbms][_charset.id] = _charset
+        if _charset.is_default:
+            self._by_name[_charset.dbms][_charset.name] = _charset
+
+    def by_id(self, id, dbms="mysql"):
+        return self._by_id.get(dbms, {}).get(id)
+
+    def by_name(self, name, dbms="mysql"):
+        if name == "utf8":
+            name = "utf8mb4"
+        return self._by_name.get(dbms, {}).get(name.lower())
+
+
+charsets = Charsets()
+charset_by_name = charsets.by_name
+charset_by_id = charsets.by_id
+
+with open(
+    os.path.join(os.path.dirname(os.path.abspath(__file__)), "charset_list.csv"), "r"
+) as f:
+    f.readline()  # pass header
+    for line in f:
+        lines = line.rstrip("\n").split(",")
+        if len(lines) != 5:
+            continue
+
+        _id, _name, _collation, _is_default, _dbms = lines
+        charsets.add(Charset(_id, _name, _collation, _is_default, _dbms))