setting BinlogStream class parameter optional_meta_data

Author: sean-k1

pymysqlreplication/binlogstream.py

@@ -141,7 +141,8 @@ def __init__(self, connection_settings, server_id,
                  slave_heartbeat=None,
                  is_mariadb=False,
                  annotate_rows_event=False,
-                 ignore_decode_errors=False):
+                 ignore_decode_errors=False,
+                 optional_meta_data=False):
         """
         Attributes:
             ctl_connection_settings: Connection settings for cluster holding
@@ -205,7 +206,7 @@ def __init__(self, connection_settings, server_id,
             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.__optional_meta_data = optional_meta_data
         # We can't filter on packet level TABLE_MAP and rotate event because
         # we need them for handling other operations
         self.__allowed_events_in_packet = frozenset(
@@ -535,7 +536,8 @@ def fetchone(self):
                                                self.__ignored_schemas,
                                                self.__freeze_schema,
                                                self.__fail_on_table_metadata_unavailable,
-                                               self.__ignore_decode_errors)
+                                               self.__ignore_decode_errors,
+                                               self.__optional_meta_data)
 
             if binlog_event.event_type == ROTATE_EVENT:
                 self.log_pos = binlog_event.event.position

pymysqlreplication/event.py

@@ -16,7 +16,8 @@ def __init__(self, from_packet, event_size, table_map, ctl_connection,
                  ignored_schemas=None,
                  freeze_schema=False,
                  fail_on_table_metadata_unavailable=False,
-                 ignore_decode_errors=False):
+                 ignore_decode_errors=False,
+                 optional_meta_data=False):
         self.packet = from_packet
         self.table_map = table_map
         self.event_type = self.packet.event_type

pymysqlreplication/packet.py

@@ -104,7 +104,8 @@ def __init__(self, from_packet, table_map,
                  ignored_schemas,
                  freeze_schema,
                  fail_on_table_metadata_unavailable,
-                 ignore_decode_errors):
+                 ignore_decode_errors,
+                 optional_meta_data):
         # -1 because we ignore the ok byte
         self.read_bytes = 0
         # Used when we want to override a value in the data buffer
@@ -150,7 +151,8 @@ def __init__(self, from_packet, table_map,
                                  ignored_schemas=ignored_schemas,
                                  freeze_schema=freeze_schema,
                                  fail_on_table_metadata_unavailable=fail_on_table_metadata_unavailable,
-                                 ignore_decode_errors=ignore_decode_errors)
+                                 ignore_decode_errors=ignore_decode_errors,
+                                 optional_meta_data=optional_meta_data)
         if self.event._processed == False:
             self.event = None
 

pymysqlreplication/row_event.py

@@ -623,7 +623,7 @@ def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs)
         self.__only_schemas = kwargs["only_schemas"]
         self.__ignored_schemas = kwargs["ignored_schemas"]
         self.__freeze_schema = kwargs["freeze_schema"]
-
+        self.__optional_meta_data = kwargs["optional_meta_data"]
         # Post-Header
         self.table_id = self._read_table_id()
 
@@ -661,6 +661,8 @@ def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs)
 
         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)
 
@@ -800,10 +802,8 @@ def _sync_column_info(self):
         column_schemas = []
         if len(self.optional_metadata.column_name_list) == 0:
             return
-        if len(self.column_schemas) == self.column_count:
-            # If the column schema length matches the number of columns,
-            # updating column schema information from optional metadata is not advisable.
-            # The reason is that the information obtained from optional metadata is not sufficient.
+        if not self.__optional_meta_data:
+            # If optional_meta_data is False Do not sync Event Time Column Schemas
             return
 
         charset_pos = 0