cherry pick from column_schema delete

Author: sean-k1

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,
@@ -211,9 +210,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
@@ -250,7 +246,6 @@ 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 = optional_meta_data
@@ -619,7 +614,6 @@ 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,

pymysqlreplication/column.py

@@ -9,27 +9,17 @@ 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.character_set_name = None
 
         if self.type == FIELD_TYPE.VARCHAR:
             self.max_length = struct.unpack("<H", packet.read(2))[0]
@@ -43,13 +33,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 +50,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/event.py

@@ -24,7 +24,6 @@ def __init__(
         only_schemas=None,
         ignored_schemas=None,
         freeze_schema=False,
-        fail_on_table_metadata_unavailable=False,
         ignore_decode_errors=False,
         verify_checksum=False,
         optional_meta_data=False,
@@ -36,7 +35,6 @@ def __init__(
         self.event_size = event_size
         self._ctl_connection = ctl_connection
         self.mysql_version = mysql_version
-        self._fail_on_table_metadata_unavailable = fail_on_table_metadata_unavailable
         self._ignore_decode_errors = ignore_decode_errors
         self._verify_checksum = verify_checksum
         self._is_event_valid = None

pymysqlreplication/packet.py

@@ -104,7 +104,6 @@ def __init__(
         only_schemas,
         ignored_schemas,
         freeze_schema,
-        fail_on_table_metadata_unavailable,
         ignore_decode_errors,
         verify_checksum,
         optional_meta_data,
@@ -158,7 +157,6 @@ def __init__(
             only_schemas=only_schemas,
             ignored_schemas=ignored_schemas,
             freeze_schema=freeze_schema,
-            fail_on_table_metadata_unavailable=fail_on_table_metadata_unavailable,
             ignore_decode_errors=ignore_decode_errors,
             verify_checksum=verify_checksum,
             optional_meta_data=optional_meta_data,

pymysqlreplication/row_event.py

@@ -8,7 +8,6 @@
 from enum import Enum
 
 from .event import BinLogEvent
-from .exceptions import TableMetadataUnavailableError
 from .constants import FIELD_TYPE
 from .constants import BINLOG
 from .constants import CHARSET
@@ -92,14 +91,6 @@ def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs)
         # Body
         self.number_of_columns = self.packet.read_length_coded_binary()
         self.columns = self.table_map[self.table_id].columns
-        column_schemas = self.table_map[self.table_id].column_schemas
-
-        if (
-            len(column_schemas) == 0
-        ):  # could not read the table metadata, probably already dropped
-            self.complete = False
-            if self._fail_on_table_metadata_unavailable:
-                raise TableMetadataUnavailableError(self.table)
 
     @staticmethod
     def _is_null(null_bitmap, position):
@@ -124,19 +115,13 @@ def _read_column_data(self, cols_bitmap):
             column = self.columns[i]
             name = self.table_map[self.table_id].columns[i].name
             unsigned = self.table_map[self.table_id].columns[i].unsigned
-            zerofill = self.table_map[self.table_id].columns[i].zerofill
-            fixed_binary_length = (
-                self.table_map[self.table_id].columns[i].fixed_binary_length
-            )
 
             values[name] = self.__read_values_name(
                 column,
                 null_bitmap,
                 null_bitmap_index,
                 cols_bitmap,
                 unsigned,
-                zerofill,
-                fixed_binary_length,
                 i,
             )
 
@@ -146,15 +131,7 @@ def _read_column_data(self, cols_bitmap):
         return values
 
     def __read_values_name(
-        self,
-        column,
-        null_bitmap,
-        null_bitmap_index,
-        cols_bitmap,
-        unsigned,
-        zerofill,
-        fixed_binary_length,
-        i,
+        self, column, null_bitmap, null_bitmap_index, cols_bitmap, unsigned, i
     ):
         if BitGet(cols_bitmap, i) == 0:
             return None
@@ -165,32 +142,24 @@ def __read_values_name(
         if column.type == FIELD_TYPE.TINY:
             if unsigned:
                 ret = struct.unpack("<B", self.packet.read(1))[0]
-                if zerofill:
-                    ret = format(ret, "03d")
                 return ret
             else:
                 return struct.unpack("<b", self.packet.read(1))[0]
         elif column.type == FIELD_TYPE.SHORT:
             if unsigned:
                 ret = struct.unpack("<H", self.packet.read(2))[0]
-                if zerofill:
-                    ret = format(ret, "05d")
                 return ret
             else:
                 return struct.unpack("<h", self.packet.read(2))[0]
         elif column.type == FIELD_TYPE.LONG:
             if unsigned:
                 ret = struct.unpack("<I", self.packet.read(4))[0]
-                if zerofill:
-                    ret = format(ret, "010d")
                 return ret
             else:
                 return struct.unpack("<i", self.packet.read(4))[0]
         elif column.type == FIELD_TYPE.INT24:
             if unsigned:
                 ret = self.packet.read_uint24()
-                if zerofill:
-                    ret = format(ret, "08d")
                 return ret
             else:
                 return self.packet.read_int24()
@@ -205,12 +174,6 @@ def __read_values_name(
                 else self.__read_string(1, column)
             )
 
-            if fixed_binary_length and len(ret) < fixed_binary_length:
-                # Fixed-length binary fields are stored in the binlog
-                # without trailing zeros and must be padded with zeros up
-                # to the specified length at read time.
-                nr_pad = fixed_binary_length - len(ret)
-                ret += b"\x00" * nr_pad
             return ret
         elif column.type == FIELD_TYPE.NEWDECIMAL:
             return self.__read_new_decimal(column)
@@ -238,8 +201,6 @@ def __read_values_name(
         elif column.type == FIELD_TYPE.LONGLONG:
             if unsigned:
                 ret = self.packet.read_uint64()
-                if zerofill:
-                    ret = format(ret, "020d")
                 return ret
             else:
                 return self.packet.read_int64()
@@ -498,6 +459,10 @@ def _dump(self):
         print("Table: %s.%s" % (self.schema, self.table))
         print("Affected columns: %d" % self.number_of_columns)
         print("Changed rows: %d" % (len(self.rows)))
+        print(
+            "Column Name Information Flag: %s"
+            % self.table_map[self.table_id].column_name_flag
+        )
 
     def _fetch_rows(self):
         self.__rows = []
@@ -537,6 +502,7 @@ def _fetch_one_row(self):
     def _dump(self):
         super()._dump()
         print("Values:")
+        print(self.table.data)
         for row in self.rows:
             print("--")
             for key in row["values"]:
@@ -702,61 +668,20 @@ def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs)
         self.column_count = self.packet.read_length_coded_binary()
 
         self.columns = []
-
-        if self.table_id in table_map:
-            self.column_schemas = table_map[self.table_id].column_schemas
-        elif self.__optional_meta_data:
-            self.column_schemas = []
-        else:
-            self.column_schemas = self._ctl_connection._get_table_information(
-                self.schema, self.table
-            )
-
         self.dbms = self._ctl_connection._get_dbms()
-
-        ordinal_pos_loc = 0
-        if self.column_count != 0:
-            # Read columns meta data
-            column_types = bytearray(self.packet.read(self.column_count))
-            self.packet.read_length_coded_binary()
-            for i in range(0, len(column_types)):
-                column_type = column_types[i]
-                try:
-                    column_schema = self.column_schemas[ordinal_pos_loc]
-
-                    # only acknowledge the column definition if the iteration matches with ordinal position of
-                    # the column. this helps in maintaining support for restricted columnar access
-                    if i != (column_schema["ORDINAL_POSITION"] - 1):
-                        # raise IndexError to follow the workflow of dropping columns which are not matching the
-                        # underlying table schema
-                        raise IndexError
-
-                    ordinal_pos_loc += 1
-                except IndexError:
-                    # this is a dirty hack to prevent row events containing columns which have been dropped prior
-                    # to pymysqlreplication start, but replayed from binlog from blowing up the service.
-                    # TODO: this does not address the issue if the column other than the last one is dropped
-                    column_schema = {
-                        "COLUMN_NAME": "__dropped_col_{i}__".format(i=i),
-                        "COLLATION_NAME": None,
-                        "CHARACTER_SET_NAME": None,
-                        "CHARACTER_OCTET_LENGTH": None,
-                        "DATA_TYPE": "BLOB",
-                        "COLUMN_COMMENT": None,
-                        "COLUMN_TYPE": "BLOB",  # we don't know what it is, so let's not do anything with it.
-                        "COLUMN_KEY": "",
-                    }
-                col = Column(column_type, column_schema, from_packet)
-                self.columns.append(col)
-
-        self.table_obj = Table(
-            self.column_schemas, self.table_id, self.schema, self.table, self.columns
-        )
+        # Read columns meta data
+        column_types = bytearray(self.packet.read(self.column_count))
+        self.packet.read_length_coded_binary()
+        for i in range(0, len(column_types)):
+            column_type = column_types[i]
+            col = Column(column_type, from_packet)
+            self.columns.append(col)
 
         # ith column is nullable if (i - 1)th bit is set to True, not nullable otherwise
         ## Refer to definition of and call to row.event._is_null() to interpret bitmap corresponding to columns
         self.null_bitmask = self.packet.read((self.column_count + 7) / 8)
-        # optional meta Data
+        self.table_obj = Table(self.table_id, self.schema, self.table, self.columns)
+        table_map[self.table_id] = self.table_obj
         self.optional_metadata = self._get_optional_meta_data()
 
         # We exclude 'CHAR' and 'INTERVAL' as they map to 'TINY' and 'ENUM' respectively

pymysqlreplication/table.py

@@ -2,9 +2,8 @@
 
 
 class Table(object):
-    def __init__(
-        self, table_id, schema, table, columns, primary_key=None, column_name_flag=False
-    ):
+    def __init__(self, table_id, schema, table, columns, primary_key=None):
+        self.column_name_flag = False
         if primary_key is None:
             primary_key = [c.data["name"] for c in columns if c.data["is_primary"]]
             if len(primary_key) == 0:
@@ -16,7 +15,6 @@ def __init__(
 
         self.__dict__.update(
             {
-                "column_schemas": column_schemas,
                 "table_id": table_id,
                 "schema": schema,
                 "table": table,