#!/usr/bin/env python3 """ Copyright (c) 2017-present, Facebook, Inc. All rights reserved. This source code is licensed under the BSD-style license found in the LICENSE file in the root directory of this source tree. """ import glob import logging import os import re import time from copy import deepcopy from threading import Timer from typing import List, Optional, Set import MySQLdb from .. import constant, sql, util from ..error import OSCError from ..hook import wrap_hook from ..sqlparse import ( is_equal, need_default_ts_bootstrap, parse_create, ParseError, SchemaDiff, ) from .base import Payload from .cleanup import CleanupPayload log: logging.Logger = logging.getLogger(__name__) class CopyPayload(Payload): """ This payload implements the actual OSC logic. Basically it'll create a new physical table and then load data into it while it keeps the original table serving read/write requests. Later it will replay the changes captured by trigger onto the new table. Finally, a table name flip will be issued to make the new schema serve requests Properties in this class have consistant name convention. A property name will look like: [old/new]_[pk/non_pk]_column_list with: - old/new representing which schema these columns are from, old or new - pk/non_pk representing whether these columns are a part of primary key """ IDCOLNAME = "_osc_ID_" DMLCOLNAME = "_osc_dml_type_" DML_TYPE_INSERT = 1 DML_TYPE_DELETE = 2 DML_TYPE_UPDATE = 3 def __init__(self, *args, **kwargs): super(CopyPayload, self).__init__(*args, **kwargs) self._current_db = None self._pk_for_filter = [] self._idx_name_for_filter = "PRIMARY" self._new_table = None self._old_table = None self._replayed_chg_ids = util.RangeChain() self.select_chunk_size = 0 self.bypass_replay_timeout = False self.is_ttl_disabled_by_me = False self.stop_before_swap = False self.outfile_suffix_end = 0 self.last_replayed_id = 0 self.last_checksumed_id = 0 self.table_size = 0 self.session_overrides = [] self._cleanup_payload = CleanupPayload(*args, **kwargs) self.stats = {} self.partitions = {} self.eta_chunks = 1 self._last_kill_timer = None self.table_swapped = False self.repl_status = kwargs.get("repl_status", "") self.outfile_dir = kwargs.get("outfile_dir", "") # By specify this option we are allowed to open a long transaction # during full table dump and full table checksum self.allow_new_pk = kwargs.get("allow_new_pk", False) self.allow_drop_column = kwargs.get("allow_drop_column", False) self.detailed_mismatch_info = kwargs.get("detailed_mismatch_info", False) self.dump_after_checksum = kwargs.get("dump_after_checksum", False) self.eliminate_dups = kwargs.get("eliminate_dups", False) self.rm_partition = kwargs.get("rm_partition", False) self.force_cleanup = kwargs.get("force_cleanup", False) self.skip_cleanup_after_kill = kwargs.get("skip_cleanup_after_kill", False) self.pre_load_statement = kwargs.get("pre_load_statement", "") self.post_load_statement = kwargs.get("post_load_statement", "") self.replay_max_attempt = kwargs.get( "replay_max_attempt", constant.DEFAULT_REPLAY_ATTEMPT ) self.replay_timeout = kwargs.get( "replay_timeout", constant.REPLAY_DEFAULT_TIMEOUT ) self.replay_batch_size = kwargs.get( "replay_batch_size", constant.DEFAULT_BATCH_SIZE ) self.replay_group_size = kwargs.get( "replay_group_size", constant.DEFAULT_REPLAY_GROUP_SIZE ) self.skip_pk_coverage_check = kwargs.get("skip_pk_coverage_check", False) self.pk_coverage_size_threshold = kwargs.get( "pk_coverage_size_threshold", constant.PK_COVERAGE_SIZE_THRESHOLD ) self.skip_long_trx_check = kwargs.get("skip_long_trx_check", False) self.ddl_file_list = kwargs.get("ddl_file_list", "") self.free_space_reserved_percent = kwargs.get( "free_space_reserved_percent", constant.DEFAULT_RESERVED_SPACE_PERCENT ) self.long_trx_time = kwargs.get("long_trx_time", constant.LONG_TRX_TIME) self.max_running_before_ddl = kwargs.get( "max_running_before_ddl", constant.MAX_RUNNING_BEFORE_DDL ) self.ddl_guard_attempts = kwargs.get( "ddl_guard_attempts", constant.DDL_GUARD_ATTEMPTS ) self.lock_max_attempts = kwargs.get( "lock_max_attempts", constant.LOCK_MAX_ATTEMPTS ) self.lock_max_wait_before_kill_seconds = kwargs.get( "lock_max_wait_before_kill_seconds", constant.LOCK_MAX_WAIT_BEFORE_KILL_SECONDS, ) self.session_timeout = kwargs.get( "mysql_session_timeout", constant.SESSION_TIMEOUT ) self.idx_recreation = kwargs.get("idx_recreation", False) self.rocksdb_bulk_load_allow_sk = kwargs.get( "rocksdb_bulk_load_allow_sk", False ) self.unblock_table_creation_without_pk = kwargs.get( "unblock_table_creation_without_pk", False ) self.rebuild = kwargs.get("rebuild", False) self.keep_tmp_table = kwargs.get("keep_tmp_table_after_exception", False) self.skip_checksum = kwargs.get("skip_checksum", False) self.skip_checksum_for_modified = kwargs.get( "skip_checksum_for_modified", False ) self.skip_delta_checksum = kwargs.get("skip_delta_checksum", False) self.skip_named_lock = kwargs.get("skip_named_lock", False) self.skip_affected_rows_check = kwargs.get("skip_affected_rows_check", False) self.where = kwargs.get("where", None) self.session_overrides_str = kwargs.get("session_overrides", "") self.fail_for_implicit_conv = kwargs.get("fail_for_implicit_conv", False) self.max_wait_for_slow_query = kwargs.get( "max_wait_for_slow_query", constant.MAX_WAIT_FOR_SLOW_QUERY ) self.max_replay_batch_size = kwargs.get( "max_replay_batch_size", constant.MAX_REPLAY_BATCH_SIZE ) self.allow_unsafe_ts_bootstrap = kwargs.get("allow_unsafe_ts_bootstrap", False) self.is_full_table_dump = False self.replay_max_changes = kwargs.get( "replay_max_changes", constant.MAX_REPLAY_CHANGES ) self.use_sql_wsenv = kwargs.get("use_sql_wsenv", False) if self.use_sql_wsenv: # by default, wsenv requires to use big chunk self.chunk_size = kwargs.get("chunk_size", constant.WSENV_CHUNK_BYTES) # by default, wsenv doesn't use local disk self.skip_disk_space_check = kwargs.get("skip_disk_space_check", True) # skip local disk space check when using wsenv if not self.skip_disk_space_check: raise OSCError("SKIP_DISK_SPACE_CHECK_VALUE_INCOMPATIBLE_WSENV") # require outfile_dir not empty if not self.outfile_dir: raise OSCError("OUTFILE_DIR_NOT_SPECIFIED_WSENV") else: self.chunk_size = kwargs.get("chunk_size", constant.CHUNK_BYTES) self.skip_disk_space_check = kwargs.get("skip_disk_space_check", False) @property def current_db(self): """ The database name this payload currently working on """ return self._current_db @property def old_pk_list(self): """ List of column names representing the primary key in the old schema. If will be used to check whether the old schema has a primary key by comparing the length to zero. Also will be used in construct the condition part of the replay query """ return [col.name for col in self._old_table.primary_key.column_list] @property def dropped_column_name_list(self): """ list of column names which exists only in old schema """ column_list = [] new_tbl_columns = [col.name for col in self._new_table.column_list] for col in self._old_table.column_list: if col.name not in new_tbl_columns: column_list.append(col.name) return column_list @property def old_column_list(self): """ list of column names for all the columns in the old schema except the ones are being dropped in the new schema. It will be used in query construction for checksum """ return [ col.name for col in self._old_table.column_list if col.name not in self.dropped_column_name_list ] @property def old_non_pk_column_list(self): """ A list of column name for all non-pk columns in the old schema. It will be used in query construction for replay """ return [ col.name for col in self._old_table.column_list if col.name not in self._pk_for_filter and col.name not in self.dropped_column_name_list ] @property def checksum_column_list(self): """ A list of non-pk column name suitable for comparing checksum """ column_list = [] old_pk_name_list = [c.name for c in self._old_table.primary_key.column_list] for col in self._old_table.column_list: if col.name in old_pk_name_list: continue if col.name in self.dropped_column_name_list: continue new_columns = {col.name: col for col in self._new_table.column_list} if col != new_columns[col.name]: if self.skip_checksum_for_modified: continue column_list.append(col.name) return column_list @property def delta_table_name(self): """ Name of the physical intermediate table for data loading. Used almost everywhere """ if len(self._old_table.name) < constant.MAX_TABLE_LENGTH - 10: return constant.DELTA_TABLE_PREFIX + self._old_table.name elif ( len(self._old_table.name) >= constant.MAX_TABLE_LENGTH - 10 and len(self._old_table.name) < constant.MAX_TABLE_LENGTH - 2 ): return constant.SHORT_DELTA_TABLE_PREFIX + self._old_table.name else: return constant.DELTA_TABLE_PREFIX + constant.GENERIC_TABLE_NAME @property def table_name(self): """ Name of the original table. Because we don't support table name change in OSC, name of the existing table should be the exactly the same as the one in the sql file. We are using 'self._new_table.name' here instead of _old_table, because _new_table will be instantiated before _old_table at early stage. It will be used by some sanity checks before we fetching data from information_schema """ return self._new_table.name @property def new_table_name(self): """ Name of the physical temporary table for loading data during OSC """ if len(self._old_table.name) < constant.MAX_TABLE_LENGTH - 10: return constant.NEW_TABLE_PREFIX + self.table_name elif ( len(self._old_table.name) >= constant.MAX_TABLE_LENGTH - 10 and len(self._old_table.name) < constant.MAX_TABLE_LENGTH - 2 ): return constant.SHORT_NEW_TABLE_PREFIX + self.table_name else: return constant.NEW_TABLE_PREFIX + constant.GENERIC_TABLE_NAME @property def renamed_table_name(self): """ Name of the old table after swap. """ if len(self._old_table.name) < constant.MAX_TABLE_LENGTH - 10: return constant.RENAMED_TABLE_PREFIX + self._old_table.name elif ( len(self._old_table.name) >= constant.MAX_TABLE_LENGTH - 10 and len(self._old_table.name) < constant.MAX_TABLE_LENGTH - 2 ): return constant.SHORT_RENAMED_TABLE_PREFIX + self._old_table.name else: return constant.RENAMED_TABLE_PREFIX + constant.GENERIC_TABLE_NAME @property def insert_trigger_name(self): """ Name of the "AFTER INSERT" trigger on the old table to capture changes during data dump/load """ if len(self._old_table.name) < constant.MAX_TABLE_LENGTH - 10: return constant.INSERT_TRIGGER_PREFIX + self._old_table.name elif ( len(self._old_table.name) >= constant.MAX_TABLE_LENGTH - 10 and len(self._old_table.name) < constant.MAX_TABLE_LENGTH - 2 ): return constant.SHORT_INSERT_TRIGGER_PREFIX + self._old_table.name else: return constant.INSERT_TRIGGER_PREFIX + constant.GENERIC_TABLE_NAME @property def update_trigger_name(self): """ Name of the "AFTER UPDATE" trigger on the old table to capture changes during data dump/load """ if len(self._old_table.name) < constant.MAX_TABLE_LENGTH - 10: return constant.UPDATE_TRIGGER_PREFIX + self._old_table.name elif ( len(self._old_table.name) >= constant.MAX_TABLE_LENGTH - 10 and len(self._old_table.name) < constant.MAX_TABLE_LENGTH - 2 ): return constant.SHORT_UPDATE_TRIGGER_PREFIX + self._old_table.name else: return constant.UPDATE_TRIGGER_PREFIX + constant.GENERIC_TABLE_NAME @property def delete_trigger_name(self): """ Name of the "AFTER DELETE" trigger on the old table to capture changes during data dump/load """ if len(self._old_table.name) < constant.MAX_TABLE_LENGTH - 10: return constant.DELETE_TRIGGER_PREFIX + self._old_table.name elif ( len(self._old_table.name) >= constant.MAX_TABLE_LENGTH - 10 and len(self._old_table.name) < constant.MAX_TABLE_LENGTH - 2 ): return constant.SHORT_DELETE_TRIGGER_PREFIX + self._old_table.name else: return constant.DELETE_TRIGGER_PREFIX + constant.GENERIC_TABLE_NAME @property def outfile(self): """ Full file path of the outfile for data dumping/loading. It's the prefix of outfile chunks. A single outfile chunk will look like '@datadir/__osc_tbl_@TABLE_NAME.1' """ return os.path.join(self.outfile_dir, constant.OUTFILE_TABLE + self.table_name) @property def tmp_table_exclude_id(self): """ Name of the temporary table which contains the value of IDCOLNAME in self.delta_table_name which we've already replayed """ return "__osc_temp_ids_to_exclude" @property def tmp_table_include_id(self): """ Name of the temporary table which contains the value of IDCOLNAME in self.delta_table_name which we will be replaying for a single self.replay_changes() call """ return "__osc_temp_ids_to_include" @property def outfile_exclude_id(self): """ Name of the outfile which contains the data which will be loaded to self.tmp_table_exclude_id soon. We cannot use insert into select from, because that will hold gap lock inside transaction. The whole select into outfile/load data infile logic is a work around for this. """ return os.path.join( self.outfile_dir, constant.OUTFILE_EXCLUDE_ID + self.table_name ) @property def outfile_include_id(self): """ Name of the outfile which contains the data which will be loaded to self.tmp_table_include_id soon. See docs in self.outfile_exclude_id for more """ return os.path.join( self.outfile_dir, constant.OUTFILE_INCLUDE_ID + self.table_name ) @property def droppable_indexes(self): """ A list of lib.sqlparse.models objects representing the indexes which can be dropped before loading data into self.new_table_name to speed up data loading """ # If we don't specified index recreation then just return a empty list # which stands for no index is suitable of dropping if not self.idx_recreation: return [] # We need to keep unique index, if we need to use it to eliminate # duplicates during data loading return self._new_table.droppable_indexes(keep_unique_key=self.eliminate_dups) def set_tx_isolation(self): """ Setting the session isolation level to RR for OSC """ # https://dev.mysql.com/worklog/task/?id=9636 # MYSQL_5_TO_8_MIGRATION if self.mysql_version.is_mysql8: self.execute_sql( sql.set_session_variable("transaction_isolation"), ("REPEATABLE-READ",) ) else: self.execute_sql( sql.set_session_variable("tx_isolation"), ("REPEATABLE-READ",) ) def set_sql_mode(self): """ Setting the sql_mode to STRICT for the connection we will using for OSC """ self.execute_sql( sql.set_session_variable("sql_mode"), ("STRICT_ALL_TABLES,NO_AUTO_VALUE_ON_ZERO",), ) def parse_session_overrides_str(self, overrides_str): """ Given a session overrides string, break it down to a list of overrides @param overrides_str: A plain string that contains the overrides @type overrides_str: string @return : A list of [var, value] """ overrides = [] if overrides_str is None or overrides_str == "": return [] for section in overrides_str.split(";"): splitted_array = section.split("=") if ( len(splitted_array) != 2 or splitted_array[0] == "" or splitted_array[1] == "" ): raise OSCError("INCORRECT_SESSION_OVERRIDE", {"section": section}) overrides.append(splitted_array) return overrides def override_session_vars(self): """ Override session variable if there's any """ self.session_overrides = self.parse_session_overrides_str( self.session_overrides_str ) for var_name, var_value in self.session_overrides: log.info( "Override session variable {} with value: {}".format( var_name, var_value ) ) self.execute_sql(sql.set_session_variable(var_name), (var_value,)) def is_var_enabled(self, var_name): if var_name not in self.mysql_vars: return False if self.mysql_vars[var_name] == "OFF": return False if self.mysql_vars[var_name] == "0": return False return True @property def is_trigger_rbr_safe(self): """ Only fb-mysql is safe for RBR if we create trigger on master alone Otherwise slave will hit _chg table not exists error """ # We only need to check this if RBR is enabled if self.mysql_vars["binlog_format"] == "ROW": if self.mysql_version.is_fb: if not self.is_var_enabled("sql_log_bin_triggers"): return True else: return False else: return False else: return True @property def is_myrocks_table(self): if not self._new_table.engine: return False return self._new_table.engine.upper() == "ROCKSDB" @property def is_myrocks_ttl_table(self): return self._new_table.is_myrocks_ttl_table def sanity_checks(self): """ Check MySQL setting for requirements that we don't necessarily need to hold a name lock for """ if not self.is_trigger_rbr_safe: raise OSCError("NOT_RBR_SAFE") def skip_cache_fill_for_myrocks(self): """ Skip block cache fill for dumps and scans to avoid cache polution """ if "rocksdb_skip_fill_cache" in self.mysql_vars: self.execute_sql(sql.set_session_variable("rocksdb_skip_fill_cache"), (1,)) @wrap_hook def init_connection(self, db): """ Initiate a connection for OSC, set session variables and get OSC lock This connection will be the only connection for the whole OSC operation It also maintain some internal state using MySQL temporary table. So an interrupted connection means a failure for the whole OSC attempt. """ log.info("== Stage 1: Init ==") self.use_db(db) self.set_no_binlog() self.get_mysql_settings() self.init_mysql_version() self.sanity_checks() self.set_tx_isolation() self.set_sql_mode() self.enable_priority_ddl() self.skip_cache_fill_for_myrocks() self.enable_sql_wsenv() self.override_session_vars() self.get_osc_lock() def table_exists(self, table_name): """ Given a table_name check whether this table already exist under current working database @param table_name: Name of the table to check existence @type table_name: string """ table_exists = self.query( sql.table_existence, ( table_name, self._current_db, ), ) return bool(table_exists) def fetch_table_schema(self, table_name): """ Use lib.sqlparse.parse_create to turn a CREATE TABLE syntax into a TABLE object, so that we can then do stuffs in a phythonic way later """ ddl = self.query(sql.show_create_table(table_name)) if ddl: try: return parse_create(ddl[0]["Create Table"]) except ParseError as e: raise OSCError( "TABLE_PARSING_ERROR", {"db": self._current_db, "table": self.table_name, "msg": str(e)}, ) def fetch_partitions(self, table_name): """ Fetching partition names from information_schema. This will be used when dropping table. If a table has a partition schema, then its partition will be dropped one by one before the table get dropped. This way we will bring less pressure to the MySQL server """ partition_result = self.query( sql.fetch_partition, ( self._current_db, table_name, ), ) # If a table doesn't have partition schema the "PARTITION_NAME" # will be string "None" instead of something considered as false # in python return [ partition_entry["PARTITION_NAME"] for partition_entry in partition_result if partition_entry["PARTITION_NAME"] != "None" ] @wrap_hook def init_table_obj(self): """ Instantiate self._old_table by parsing the output of SHOW CREATE TABLE from MySQL instance. Because we need to parse out the table name we'll act on, this should be the first step before we start to doing anything """ # Check the existence of original table if not self.table_exists(self.table_name): raise OSCError( "TABLE_NOT_EXIST", {"db": self._current_db, "table": self.table_name} ) self._old_table = self.fetch_table_schema(self.table_name) self.partitions[self.table_name] = self.fetch_partitions(self.table_name) # The table after swap will have the same partition layout as current # table self.partitions[self.renamed_table_name] = self.partitions[self.table_name] # Preserve the auto_inc value from old table, so that we don't revert # back to a smaller value after OSC if self._old_table.auto_increment: self._new_table.auto_increment = self._old_table.auto_increment # Populate both old and new tables with explicit charset/collate self.populate_charset_collation(self._old_table) self.populate_charset_collation(self._new_table) def cleanup_with_force(self): """ Loop through all the tables we will touch during OSC, and clean them up if force_cleanup is specified """ log.info( "--force-cleanup specified, cleaning up things that may left " "behind by last run" ) cleanup_payload = CleanupPayload(charset=self.charset, sudo=self.sudo) # cleanup outfiles for include_id and exclude_id for filepath in (self.outfile_exclude_id, self.outfile_include_id): cleanup_payload.add_file_entry(filepath) # cleanup outfiles for detailed checksum for suffix in ["old", "new"]: cleanup_payload.add_file_entry("{}.{}".format(self.outfile, suffix)) # cleanup outfiles for table dump file_prefixes = [ self.outfile, "{}.old".format(self.outfile), "{}.new".format(self.outfile), ] for file_prefix in file_prefixes: log.debug("globbing {}".format(file_prefix)) for outfile in glob.glob("{}.[0-9]*".format(file_prefix)): cleanup_payload.add_file_entry(outfile) for trigger in ( self.delete_trigger_name, self.update_trigger_name, self.insert_trigger_name, ): cleanup_payload.add_drop_trigger_entry(self._current_db, trigger) for tbl in ( self.new_table_name, self.delta_table_name, self.renamed_table_name, ): partitions = self.fetch_partitions(tbl) cleanup_payload.add_drop_table_entry(self._current_db, tbl, partitions) cleanup_payload.mysql_user = self.mysql_user cleanup_payload.mysql_pass = self.mysql_pass cleanup_payload.socket = self.socket cleanup_payload.get_conn_func = self.get_conn_func cleanup_payload.cleanup(self._current_db) cleanup_payload.close_conn() @wrap_hook def determine_outfile_dir(self): """ Determine the output directory we will use to store dump file """ if self.outfile_dir: return # if --tmpdir is not specified on command line for outfiles # use @@secure_file_priv for var_name in ("@@secure_file_priv", "@@datadir"): result = self.query(sql.select_as(var_name, "folder")) if not result: raise Exception("Failed to get {} system variable".format(var_name)) if result[0]["folder"]: if var_name == "@@secure_file_priv": self.outfile_dir = result[0]["folder"] else: self.outfile_dir = os.path.join( result[0]["folder"], self._current_db_dir ) log.info("Will use {} storing dump outfile".format(self.outfile_dir)) return raise Exception("Cannot determine output dir for dump") def trigger_check(self): """ Check whether there's any trigger already exist on the table we're about to touch """ triggers = self.query( sql.trigger_existence, (self.table_name, self._current_db), ) if triggers: trigger_desc = [] for trigger in triggers: trigger_desc.append( "Trigger name: {}, Action: {} {}".format( trigger["TRIGGER_NAME"], trigger["ACTION_TIMING"], trigger["EVENT_MANIPULATION"], ) ) raise OSCError( "TRIGGER_ALREADY_EXIST", {"triggers": "\n".join(trigger_desc)} ) def foreign_key_check(self): """ Check whether the table has been referred to any existing foreign definition """ # MyRocks doesn't support foreign key if self.is_myrocks_table: log.info( "SKip foreign key check because MyRocks doesn't support " "this yet" ) return True foreign_keys = self.query( sql.foreign_key_cnt, ( self.table_name, self._current_db, self.table_name, self._current_db, ), ) if foreign_keys: fk = "CONSTRAINT `{}` FOREIGN KEY (`{}`) REFERENCES `{}` (`{}`)".format( foreign_keys[0]["constraint_name"], foreign_keys[0]["col_name"], foreign_keys[0]["ref_tab"], foreign_keys[0]["ref_col_name"], ) raise OSCError( "FOREIGN_KEY_FOUND", {"db": self._current_db, "table": self.table_name, "fk": fk}, ) def get_table_size_from_IS(self, table_name): """ Given a table_name return its current size in Bytes from information_schema @param table_name: Name of the table to fetch size @type table_name: string """ result = self.query(sql.show_table_stats(self._current_db), (self.table_name,)) if result: return result[0]["Data_length"] + result[0]["Index_length"] return 0 def get_table_size_for_myrocks(self, table_name): """ Given a table_name return its raw data size before compression. MyRocks is very good at compression, the on disk dump size is much bigger than the actual MyRocks table size, hence we will use raw size for the esitmation of the maximum disk usage @param table_name: Name of the table to fetch size @type table_name: string """ result = self.query( sql.get_myrocks_table_size(), ( self._current_db, self.table_name, ), ) if result: return result[0]["raw_size"] or 0 return 0 def get_table_size(self, table_name): """ Given a table_name return its current size in Bytes @param table_name: Name of the table to fetch size @type table_name: string """ if self.is_myrocks_table: return self.get_table_size_for_myrocks(table_name) else: return self.get_table_size_from_IS(table_name) def check_disk_size(self): """ Check if we have enough disk space to execute the DDL """ if self.skip_disk_space_check: return True self.table_size = int(self.get_table_size(self.table_name)) disk_space = int(util.disk_partition_free(self.outfile_dir)) # With allow_new_pk, we will create one giant outfile, and so at # some point will have the entire new table and the entire outfile # both existing simultaneously. if self.allow_new_pk and not self._old_table.primary_key.column_list: required_size = self.table_size * 2 else: required_size = self.table_size * 1.1 log.info( "Disk space required: {}, available: {}".format( util.readable_size(required_size), util.readable_size(disk_space) ) ) if required_size > disk_space: raise OSCError( "NOT_ENOUGH_SPACE", { "need": util.readable_size(required_size), "avail": util.readable_size(disk_space), }, ) def check_disk_free_space_reserved(self): """ Check if we have enough free space left during dump data """ if self.skip_disk_space_check: return True disk_partition_size = util.disk_partition_size(self.outfile_dir) free_disk_space = util.disk_partition_free(self.outfile_dir) free_space_factor = self.free_space_reserved_percent / 100 free_space_reserved = disk_partition_size * free_space_factor if free_disk_space < free_space_reserved: raise OSCError( "NOT_ENOUGH_SPACE", { "need": util.readable_size(free_space_reserved), "avail": util.readable_size(free_disk_space), }, ) def validate_post_alter_pk(self): """ As we force (primary) when replaying changes, we have to make sure rows in new table schema can be accessed using old PK combination. The logic here is to make sure the old table's primary key list equals to the set which one of the new table's index prefix can form. Otherwise there'll be a performance issue when replaying changes based on old primary key combination. Note that if old PK is (a, b), new PK is (b, a, c) is acceptable, because for each combination of (a, b), it still can utilize the new PK for row searching. Same for old PK being (a, b, c), new PK is (a, b) because new PK is more strict, so it will always return at most one row when using old PK columns as WHERE condition. However if the old PK is (a, b, c), new PK is (b, c, d). Then there's a chance the changes may not be able to be replay efficiently. Because using only column (b, c) for row searching may result in a huge number of matched rows """ idx_on_new_table = [self._new_table.primary_key] + self._new_table.indexes old_pk_len = len(self._pk_for_filter) for idx in idx_on_new_table: log.debug("Checking prefix for {}".format(idx.name)) idx_prefix = idx.column_list[:old_pk_len] idx_name_set = {col.name for col in idx_prefix} # Identical set and covered set are considered as covering if set(self._pk_for_filter) == idx_name_set: log.info("PK prefix on new table can cover PK from old table") return True if idx.is_unique and set(self._pk_for_filter) > idx_name_set: log.info("old PK can uniquely identify rows from new schema") return True return False def find_coverage_index(self): """ Find an unique index which can perfectly cover old pri-key search in order to calculate checksum for new table. We will use this index name as force index in checksum query See validate_post_alter_pk for more detail about pri-key coverage """ idx_on_new_table = [self._new_table.primary_key] + self._new_table.indexes old_pk_len = len(self._pk_for_filter) for idx in idx_on_new_table: # list[:idx] where idx > len(list) yields full list idx_prefix = idx.column_list[:old_pk_len] idx_name_list = [col.name for col in idx_prefix] if self._pk_for_filter == idx_name_list: if idx.is_unique: return idx.name return None def init_range_variables(self): """ Initial array and string which contains the same number of session variables as the columns of primary key. This will be used as chunk boundary when dumping and checksuming """ self.range_start_vars_array = [] self.range_end_vars_array = [] for idx in range(len(self._pk_for_filter)): self.range_start_vars_array.append("@range_start_{}".format(idx)) self.range_end_vars_array.append("@range_end_{}".format(idx)) self.range_start_vars = ",".join(self.range_start_vars_array) self.range_end_vars = ",".join(self.range_end_vars_array) def make_chunk_size_odd(self): """ Ensure select_chunk_size is an odd number. Because we use this number as chunk size for checksum as well. If a column has exact the same value for all its rows, then return value from BIT_XOR(CRC32(`col`)) will be zero for even number of rows, no matter what value it has. """ if self.select_chunk_size % 2 == 0: self.select_chunk_size = self.select_chunk_size + 1 def get_table_chunk_size(self): """ Calculate the number of rows for each table dump query table based on average row length and the chunks size we've specified """ result = self.query( sql.table_avg_row_len, ( self._current_db, self.table_name, ), ) if result: tbl_avg_length = result[0]["AVG_ROW_LENGTH"] # avoid huge chunk row count if tbl_avg_length < 20: tbl_avg_length = 20 self.select_chunk_size = self.chunk_size // tbl_avg_length # This means either the avg row size is huge, or user specified # a tiny select_chunk_size on CLI. Let's make it one row per # outfile to avoid zero division if not self.select_chunk_size: self.select_chunk_size = 1 log.info( "TABLE contains {} rows, table_avg_row_len: {} bytes," "chunk_size: {} bytes, ".format( result[0]["TABLE_ROWS"], tbl_avg_length, self.chunk_size ) ) log.info("Outfile will contain {} rows each".format(self.select_chunk_size)) self.eta_chunks = max( int(result[0]["TABLE_ROWS"] / self.select_chunk_size), 1 ) else: raise OSCError("FAIL_TO_GUESS_CHUNK_SIZE") def has_desired_schema(self): """ Check whether the existing table already has the desired schema. """ if self._new_table == self._old_table: if not self.rebuild: log.info("Table already has the desired schema. ") return True else: log.info( "Table already has the desired schema. However " "--rebuild is specified, doing a rebuild instead" ) return False return False def decide_pk_for_filter(self): # If we are adding a PK, then we should use all the columns in # old table to identify an unique row if not all( (self._old_table.primary_key, self._old_table.primary_key.column_list) ): # Let's try to get an UK if possible for idx in self._old_table.indexes: if idx.is_unique: log.info( "Old table doesn't have a PK but has an UK: {}".format(idx.name) ) self._pk_for_filter = [col.name for col in idx.column_list] self._pk_for_filter_def = idx.column_list.copy() self._idx_name_for_filter = idx.name break else: # There's no UK either if self.allow_new_pk: self._pk_for_filter = [ col.name for col in self._old_table.column_list ] self._pk_for_filter_def = self._old_table.column_list.copy() self.is_full_table_dump = True else: raise OSCError("NEW_PK") # If we have PK in existing schema, then we use current PK as an unique # row finder else: # if any of the columns of the primary key is prefixed, we want to # use full_table_dump, instead of chunking, so that it doesn't fill # up the disk # e.g. name below is a prefixed col in the PK (assume varchar(99)) # since we dont use full col in PK - `PRIMARY KEY(id, name(10))` for col in self._old_table.primary_key.column_list: if col.length: log.info( "Found prefixed column/s as part of the PK. " "Will do full table dump (no chunking)." ) self._pk_for_filter = [c.name for c in self._old_table.column_list] self._pk_for_filter_def = self._old_table.column_list.copy() self.is_full_table_dump = True break else: self._pk_for_filter = [ col.name for col in self._old_table.primary_key.column_list ] self._pk_for_filter_def = self._old_table.primary_key.column_list.copy() def ts_bootstrap_check(self): """ Check when going from old schema to new, whether bootstraping column using CURRENT_TIMESTAMP is involved. This is a dangerous thing to do out of replication and is disallowed by default """ if not need_default_ts_bootstrap(self._old_table, self._new_table): return if self.allow_unsafe_ts_bootstrap: log.warning( "Bootstraping timestamp column using current time is required. " "Bypassing the safety check as requested" ) return raise OSCError("UNSAFE_TS_BOOTSTRAP") @wrap_hook def pre_osc_check(self): """ Pre-OSC sanity check. Make sure all temporary table which will be used during data copy stage doesn't exist before we actually creating one. Also doing some index sanity check. """ # Make sure temporary table we will use during copy doesn't exist tables_to_check = ( self.new_table_name, self.delta_table_name, self.renamed_table_name, ) for table_name in tables_to_check: if self.table_exists(table_name): raise OSCError( "TABLE_ALREADY_EXIST", {"db": self._current_db, "table": table_name} ) # Make sure new table schema has primary key if not all( (self._new_table.primary_key, self._new_table.primary_key.column_list) ): raise OSCError( "NO_PK_EXIST", {"db": self._current_db, "table": self.table_name} ) self.decide_pk_for_filter() # Check if we can have indexes in new table to efficiently look up # current old pk combinations if not self.validate_post_alter_pk(): self.table_size = self.get_table_size(self.table_name) if self.skip_pk_coverage_check: log.warning( "Indexes on new table cannot cover current PK of " "the old schema, which will make binary logs replay " "in an inefficient way." ) elif self.table_size < self.pk_coverage_size_threshold: log.warning( "No index on new table can cover old pk. Since this is " "a small table: {}, we fallback to a full table dump".format( self.table_size ) ) # All columns will be chosen if we are dumping table without # chunking, this means all columns will be used as a part of # the WHERE condition when replaying self.is_full_table_dump = True self._pk_for_filter = [col.name for col in self._old_table.column_list] self._pk_for_filter_def = self._old_table.column_list.copy() elif self.is_full_table_dump: log.warning( "Skipping coverage index test, since we are doing " "full table dump" ) else: old_pk_names = ", ".join( "`{}`".format(col.name) for col in self._old_table.primary_key.column_list ) raise OSCError("NO_INDEX_COVERAGE", {"pk_names": old_pk_names}) log.info( "PK filter for replaying changes later: {}".format(self._pk_for_filter) ) self.foreign_key_check() self.trigger_check() self.init_range_variables() self.get_table_chunk_size() self.make_chunk_size_odd() self.check_disk_size() self.ts_bootstrap_check() self.drop_columns_check() def drop_columns_check(self): # We only allow dropping columns with the flag --alow-drop-column. if self.dropped_column_name_list: if self.allow_drop_column: for diff_column in self.dropped_column_name_list: log.warning( "Column `{}` is missing in the new schema, " "but --alow-drop-column is specified. Will " "drop this column.".format(diff_column) ) else: missing_columns = ", ".join(self.dropped_column_name_list) raise OSCError("MISSING_COLUMN", {"column": missing_columns}) # We don't allow dropping columns from current primary key for col in self._pk_for_filter: if col in self.dropped_column_name_list: raise OSCError("PRI_COL_DROPPED", {"pri_col": col}) def add_drop_table_entry(self, table_name): """ A wrapper for adding drop table request to CleanupPayload. The database name will always be the one we are currently working on. Also partition name list will be included as fetched from information schema before DDL """ self._cleanup_payload.add_drop_table_entry( self._current_db, table_name, self.partitions.get(table_name, []) ) def get_collations(self): """ Get a list of supported collations with their corresponding charsets """ collations = self.query(sql.all_collation) collation_charsets = {} for r in collations: collation_charsets[r["COLLATION_NAME"]] = r["CHARACTER_SET_NAME"] return collation_charsets def get_default_collations(self): """ Get a list of supported character set and their corresponding default collations """ collations = self.query(sql.default_collation) charset_collations = {} for r in collations: charset_collations[r["CHARACTER_SET_NAME"]] = r["COLLATION_NAME"] # Populate utf8mb4 override utf8_override = self.query( sql.get_global_variable("default_collation_for_utf8mb4") ) if utf8_override and "utf8mb4" in charset_collations: charset_collations["utf8mb4"] = utf8_override[0]["Value"] return charset_collations def populate_charset_collation(self, schema_obj): default_collations = self.get_default_collations() collation_charsets = self.get_collations() if schema_obj.charset is not None and schema_obj.collate is None: schema_obj.collate = default_collations.get(schema_obj.charset, None) if schema_obj.charset is None and schema_obj.collate is not None: # Shouldn't reach here, since every schema should have default charset, # otherwise linting will error out. Leave the logic here just in case. # In this case, we would not populate the charset because we actually # want the user to explicit write the charset in the desired schema. # In db, charset is always populated(explicit) by default. schema_obj.charset = None # make column charset & collate explicit # follow https://dev.mysql.com/doc/refman/8.0/en/charset-column.html text_types = {"CHAR", "VARCHAR", "TEXT", "MEDIUMTEXT", "LONGTEXT", "ENUM"} for column in schema_obj.column_list: if column.column_type in text_types: # Check collate first to guarantee the column uses table collate # if column charset is absent. If checking charset first and column # collate is absent, it will use table charset and get default # collate from the database, which does not work for tables with # non default collate settings if column.collate is None: if column.charset and default_collations.get(column.charset, None): column.collate = default_collations[column.charset] else: column.collate = schema_obj.collate if column.charset is None: if column.collate and collation_charsets.get(column.collate, None): column.charset = collation_charsets[column.collate] else: # shouldn't reach here, unless charset_to_collate # or collate_to_charset doesn't have the mapped value column.charset = schema_obj.charset return schema_obj def remove_using_hash_for_80(self): """ Remove `USING HASH` for indexes that explicitly have it, because that's the 8.0 behavior """ for index in self._new_table.indexes: if index.using == "HASH": index.using = None @wrap_hook def create_copy_table(self): """ Create the physical temporary table using new schema """ tmp_sql_obj = deepcopy(self._new_table) tmp_sql_obj.name = self.new_table_name if self.rm_partition: tmp_sql_obj.partition = self._old_table.partition tmp_sql_obj.partition_config = self._old_table.partition_config tmp_table_ddl = tmp_sql_obj.to_sql() log.info("Creating copy table using: {}".format(tmp_table_ddl)) self.execute_sql(tmp_table_ddl) self.partitions[self.new_table_name] = self.fetch_partitions( self.new_table_name ) self.add_drop_table_entry(self.new_table_name) # Check whether the schema is consistent after execution to avoid # any implicit conversion if self.fail_for_implicit_conv: obj_after = self.fetch_table_schema(self.new_table_name) obj_after.engine = self._new_table.engine obj_after.name = self._new_table.name # Ignore partition difference, since there will be no implicit # conversion here obj_after.partition = self._new_table.partition obj_after.partition_config = self._new_table.partition_config self.populate_charset_collation(obj_after) if self.mysql_version.is_mysql8: # Remove 'USING HASH' in keys on 8.0, when present in 5.6, as 8.0 # removes it by default self.remove_using_hash_for_80() if obj_after != self._new_table: raise OSCError( "IMPLICIT_CONVERSION_DETECTED", {"diff": str(SchemaDiff(self._new_table, obj_after))}, ) @wrap_hook def create_delta_table(self): """ Create the table which will store changes made to existing table during OSC. This can be considered as table level binlog """ self.execute_sql( sql.create_delta_table( self.delta_table_name, self.IDCOLNAME, self.DMLCOLNAME, self._old_table.engine, self.old_column_list, self._old_table.name, ) ) self.add_drop_table_entry(self.delta_table_name) # We will break table into chunks when calculate checksums using # old primary key. We need this index to skip verify the same row # for multiple time if it has been changed a lot if self._pk_for_filter_def and not self.is_full_table_dump: self.execute_sql( sql.create_idx_on_delta_table( self.delta_table_name, [col.to_sql() for col in self._pk_for_filter_def], ) ) def create_insert_trigger(self): self.execute_sql( sql.create_insert_trigger( self.insert_trigger_name, self.table_name, self.delta_table_name, self.DMLCOLNAME, self.old_column_list, self.DML_TYPE_INSERT, ) ) self._cleanup_payload.add_drop_trigger_entry( self._current_db, self.insert_trigger_name ) @wrap_hook def create_delete_trigger(self): self.execute_sql( sql.create_delete_trigger( self.delete_trigger_name, self.table_name, self.delta_table_name, self.DMLCOLNAME, self.old_column_list, self.DML_TYPE_DELETE, ) ) self._cleanup_payload.add_drop_trigger_entry( self._current_db, self.delete_trigger_name ) def create_update_trigger(self): self.execute_sql( sql.create_update_trigger( self.update_trigger_name, self.table_name, self.delta_table_name, self.DMLCOLNAME, self.old_column_list, self.DML_TYPE_UPDATE, self.DML_TYPE_DELETE, self.DML_TYPE_INSERT, self._pk_for_filter, ) ) self._cleanup_payload.add_drop_trigger_entry( self._current_db, self.update_trigger_name ) def get_long_trx(self): """ Return a long running transaction agaisnt the table we'll touch, if there's one. This is mainly for safety as long running transaction may block DDL, thus blocks more other requests """ if self.skip_long_trx_check: return False processes = self.query(sql.show_processlist) for proc in processes: if not proc["Info"]: sql_statement = "" else: if isinstance(proc["Info"], bytes): sql_statement = proc["Info"].decode("utf-8", "replace") else: sql_statement = proc["Info"] proc["Info"] = sql_statement # Time can be None if the connection is in "Connect" state if ( (proc.get("Time") or 0) > self.long_trx_time and proc.get("db", "") == self._current_db and self.table_name in "--" + sql_statement and not proc.get("Command", "") == "Sleep" ): return proc def wait_until_slow_query_finish(self): for _ in range(self.max_wait_for_slow_query): slow_query = self.get_long_trx() if slow_query: log.info( "Slow query pid={} is still running".format(slow_query.get("Id", 0)) ) time.sleep(5) else: return True else: raise OSCError( "LONG_RUNNING_TRX", { "pid": slow_query.get("Id", 0), "user": slow_query.get("User", ""), "host": slow_query.get("Host", ""), "time": slow_query.get("Time", ""), "command": slow_query.get("Command", ""), "info": slow_query.get("Info", b"") .encode("utf-8") .decode("utf-8", "replace"), }, ) def kill_selects(self, table_names, conn=None): """ Kill current running SELECTs against the specified tables in the working database so that they won't block the DDL statement we're about to execute. The conn parameter allows to use a different connection. A different connection is necessary when it is needed to kill queries that may be blocking the current connection """ conn = conn or self.conn table_names = [tbl.lower() for tbl in table_names] # We use regex matching to find running queries on top of the tables # Better options (as in more precise) would be: # 1. List the current held metadata locks, but this is not possible # without the performance schema # 2. Actually parse the SQL of the running queries, but this can be # quite expensive keyword_pattern = ( r"(\s|^)" # whitespace or start r"({})" # keyword(s) r"(\s|$)" # whitespace or end ) table_pattern = ( r"(\s|`)" # whitespace or backtick r"({})" # table(s) r"(\s|`|$)" # whitespace, backtick or end ) alter_or_select_pattern = re.compile(keyword_pattern.format("select|alter")) information_schema_pattern = re.compile( keyword_pattern.format("information_schema") ) any_tables_pattern = re.compile(table_pattern.format("|".join(table_names))) processlist = conn.get_running_queries() for proc in processlist: sql_statement = proc.get("Info") or "".encode("utf-8") sql_statement = sql_statement.decode("utf-8", "replace").lower() if ( proc["db"] == self._current_db and sql_statement and not information_schema_pattern.search(sql_statement) and any_tables_pattern.search(sql_statement) and alter_or_select_pattern.search(sql_statement) ): try: conn.kill_query_by_id(int(proc["Id"])) except MySQLdb.MySQLError as e: errcode, errmsg = e.args # 1094: Unknown thread id # This means the query we were trying to kill has finished # before we run kill %d if errcode == 1094: log.info( "Trying to kill query id: {}, but it has " "already finished".format(proc["Id"]) ) else: raise def start_transaction(self): """ Start a transaction. """ self.execute_sql(sql.start_transaction) def commit(self): """ Commit and close the transaction """ self.execute_sql(sql.commit) def ddl_guard(self): """ If there're already too many concurrent queries running, it's probably a bad idea to run DDL. Wait for some time until they finished or we timed out """ for _ in range(self.ddl_guard_attempts): result = self.query(sql.show_status, ("Threads_running",)) if result: threads_running = int(result[0]["Value"]) if threads_running > self.max_running_before_ddl: log.warning( "Threads running: {}, bigger than allowed: {}. " "Sleep 1 second before check again.".format( threads_running, self.max_running_before_ddl ) ) time.sleep(1) else: log.debug( "Threads running: {}, less than: {}. We are good " "to go".format(threads_running, self.max_running_before_ddl) ) return log.error( "Hit max attempts: {}, but the threads running still don't drop" "below: {}.".format(self.ddl_guard_attempts, self.max_running_before_ddl) ) raise OSCError("DDL_GUARD_FAILED") @wrap_hook def lock_tables(self, tables): for _ in range(self.lock_max_attempts): # We use a threading.Timer with a second connection in order to # kill any selects on top of the tables being altered if we could # not lock the tables in time another_conn = self.get_conn(self._current_db) kill_timer = Timer( self.lock_max_wait_before_kill_seconds, self.kill_selects, args=(tables, another_conn), ) # keeping a reference to kill timer helps on tests self._last_kill_timer = kill_timer kill_timer.start() try: self.execute_sql(sql.lock_tables(tables)) # It is best to cancel the timer as soon as possible kill_timer.cancel() log.info( "Successfully lock table(s) for write: {}".format(", ".join(tables)) ) break except MySQLdb.MySQLError as e: errcode, errmsg = e.args # 1205 is timeout and 1213 is deadlock if errcode in (1205, 1213): log.warning("Retry locking because of error: {}".format(e)) else: raise finally: # guarantee that we dont leave a stray kill timer running # or any open resources kill_timer.cancel() kill_timer.join() another_conn.close() else: # Cannot lock write after max lock attempts raise OSCError("FAILED_TO_LOCK_TABLE", {"tables": ", ".join(tables)}) def unlock_tables(self): self.execute_sql(sql.unlock_tables) log.info("Table(s) unlocked") @wrap_hook def create_triggers(self): self.wait_until_slow_query_finish() self.stop_slave_sql() self.ddl_guard() log.debug("Locking table: {} before creating trigger".format(self.table_name)) if not self.is_high_pri_ddl_supported: self.lock_tables(tables=[self.table_name]) try: log.info("Creating triggers") # Because we've already hold the WRITE LOCK on the table, it's now safe # to deal with operations that require metadata lock self.create_insert_trigger() self.create_delete_trigger() self.create_update_trigger() except Exception as e: if not self.is_high_pri_ddl_supported: self.unlock_tables() self.start_slave_sql() log.error("Failed to execute sql for creating triggers") raise OSCError("CREATE_TRIGGER_ERROR", {"msg": str(e)}) if not self.is_high_pri_ddl_supported: self.unlock_tables() self.start_slave_sql() def disable_ttl_for_myrocks(self): if self.mysql_vars.get("rocksdb_enable_ttl", "OFF") == "ON": self.execute_sql(sql.set_global_variable("rocksdb_enable_ttl"), ("OFF",)) self.is_ttl_disabled_by_me = True else: log.debug("TTL not enabled for MyRocks, skip") def enable_ttl_for_myrocks(self): if self.is_ttl_disabled_by_me: self.execute_sql(sql.set_global_variable("rocksdb_enable_ttl"), ("ON",)) else: log.debug("TTL not enabled for MyRocks before schema change, skip") @wrap_hook def start_snapshot(self): # We need to disable TTL feature in MyRocks. Otherwise rows will # possibly be purged during dump/load, and cause checksum mismatch if self.is_myrocks_table and self.is_myrocks_ttl_table: log.debug("It's schema change for MyRocks table which is using TTL") self.disable_ttl_for_myrocks() self.execute_sql(sql.start_transaction_with_snapshot) current_max = self.get_max_delta_id() log.info( "Changes with id <= {} committed before dump snapshot, " "and should be ignored.".format(current_max) ) # Only replay changes in this range (last_replayed_id, max_id_now] new_changes = self.query( sql.get_replay_row_ids( self.IDCOLNAME, self.DMLCOLNAME, self.delta_table_name, None, self.mysql_version.is_mysql8, ), ( self.last_replayed_id, current_max, ), ) self._replayed_chg_ids.extend([r[self.IDCOLNAME] for r in new_changes]) self.last_replayed_id = current_max def affected_rows(self): return self._conn.conn.affected_rows() def refresh_range_start(self): self.execute_sql(sql.select_into(self.range_end_vars, self.range_start_vars)) def select_full_table_into_outfile(self): stage_start_time = time.time() try: outfile = "{}.1".format(self.outfile) sql_string = sql.select_full_table_into_file( self._pk_for_filter + self.old_non_pk_column_list, self.table_name, self.where, ) affected_rows = self.execute_sql(sql_string, (outfile,)) self.outfile_suffix_end = 1 self.stats["outfile_lines"] = affected_rows self._cleanup_payload.add_file_entry(outfile) self.commit() except MySQLdb.OperationalError as e: errnum, errmsg = e.args # 1086: File exists if errnum == 1086: raise OSCError("FILE_ALREADY_EXIST", {"file": outfile}) else: raise self.stats["time_in_dump"] = time.time() - stage_start_time @wrap_hook def select_chunk_into_outfile(self, outfile, use_where): try: sql_string = sql.select_full_table_into_file_by_chunk( self.table_name, self.range_start_vars_array, self.range_end_vars_array, self._pk_for_filter, self.old_non_pk_column_list, self.select_chunk_size, use_where, self.where, self._idx_name_for_filter, ) affected_rows = self.execute_sql(sql_string, (outfile,)) except MySQLdb.OperationalError as e: errnum, errmsg = e.args # 1086: File exists if errnum == 1086: raise OSCError("FILE_ALREADY_EXIST", {"file": outfile}) else: raise log.debug("{} affected".format(affected_rows)) self.stats["outfile_lines"] = affected_rows + self.stats.setdefault( "outfile_lines", 0 ) self.stats["outfile_cnt"] = 1 + self.stats.setdefault("outfile_cnt", 0) self._cleanup_payload.add_file_entry( "{}.{}".format(self.outfile, self.outfile_suffix_end) ) return affected_rows @wrap_hook def select_table_into_outfile(self): log.info("== Stage 2: Dump ==") stage_start_time = time.time() # We can not break the table into chunks when there's no existing pk # We'll have to use one big file for copy data if self.is_full_table_dump: log.info("Dumping full table in one go.") return self.select_full_table_into_outfile() outfile_suffix = 1 # To let the loop run at least once affected_rows = 1 use_where = False printed_chunk = 0 while affected_rows: self.outfile_suffix_end = outfile_suffix outfile = "{}.{}".format(self.outfile, outfile_suffix) affected_rows = self.select_chunk_into_outfile(outfile, use_where) # Refresh where condition range for next select if affected_rows: self.refresh_range_start() use_where = True outfile_suffix += 1 self.check_disk_free_space_reserved() progress_pct = int((float(outfile_suffix) / self.eta_chunks) * 100) progress_chunk = int(progress_pct / 10) if progress_chunk > printed_chunk and self.eta_chunks > 10: log.info( "Dump progress: {}/{}(ETA) chunks".format( outfile_suffix, self.eta_chunks ) ) printed_chunk = progress_chunk self.commit() log.info("Dump finished") self.stats["time_in_dump"] = time.time() - stage_start_time @wrap_hook def drop_non_unique_indexes(self): """ Drop non-unique indexes from the new table to speed up the load process """ for idx in self.droppable_indexes: log.info("Dropping index '{}' on intermediate table".format(idx.name)) self.ddl_guard() self.execute_sql(sql.drop_index(idx.name, self.new_table_name)) @wrap_hook def load_chunk(self, column_list, chunk_id): sql_string = sql.load_data_infile( self.new_table_name, column_list, ignore=self.eliminate_dups ) log.debug(sql_string) filepath = "{}.{}".format(self.outfile, chunk_id) self.execute_sql(sql_string, (filepath,)) # Delete the outfile once we have the data in new table to free # up space as soon as possible if not self.use_sql_wsenv and self.rm_file(filepath): util.sync_dir(self.outfile_dir) self._cleanup_payload.remove_file_entry(filepath) def change_explicit_commit(self, enable=True): """ Turn on/off rocksdb_commit_in_the_middle to avoid commit stall for large data infiles """ v = 1 if enable else 0 try: self.execute_sql( sql.set_session_variable("rocksdb_commit_in_the_middle"), (v,) ) except MySQLdb.OperationalError as e: errnum, errmsg = e.args # 1193: unknown variable if errnum == 1193: log.warning( "Failed to set rocksdb_commit_in_the_middle: {}".format(errmsg) ) else: raise def change_rocksdb_bulk_load(self, enable=True): # rocksdb_bulk_load relies on data being dumping in the same sequence # as new pk. If we are changing pk, then we cannot ensure that if self._old_table.primary_key != self._new_table.primary_key: log.warning("Skip rocksdb_bulk_load, because we are changing PK") return v = 1 if enable else 0 # rocksdb_bulk_load and rocksdb_bulk_load_allow_sk have the # following sequence requirement so setting values accordingly. # SET SESSION rocksdb_bulk_load_allow_sk=1; # SET SESSION rocksdb_bulk_load=1; # ... (bulk loading) # SET SESSION rocksdb_bulk_load=0; # SET SESSION rocksdb_bulk_load_allow_sk=0; try: if self.rocksdb_bulk_load_allow_sk and enable: self.execute_sql( sql.set_session_variable("rocksdb_bulk_load_allow_sk"), (v,) ) self.execute_sql(sql.set_session_variable("rocksdb_bulk_load"), (v,)) if self.rocksdb_bulk_load_allow_sk and not enable: self.execute_sql( sql.set_session_variable("rocksdb_bulk_load_allow_sk"), (v,) ) except MySQLdb.OperationalError as e: errnum, errmsg = e.args # 1193: unknown variable if errnum == 1193: log.warning("Failed to set rocksdb_bulk_load: {}".format(errmsg)) else: raise @wrap_hook def load_data(self): stage_start_time = time.time() log.info("== Stage 3: Load data ==") # Generate the column name list string for load data infile # The column sequence is not exact the same as the original table. # It's pk_col_names + non_pk_col_name instead if self._pk_for_filter: if self.old_non_pk_column_list: column_list = self._pk_for_filter + self.old_non_pk_column_list else: column_list = self._pk_for_filter elif self.old_non_pk_column_list: column_list = self.old_non_pk_column_list else: # It's impossible to reach here, otherwise it means there's zero # column in old table which MySQL doesn't support. Something is # totally wrong if we get to this point raise OSCError( "OSC_INTERNAL_ERROR", { "msg": "Unexpected scenario. Both _pk_for_filter " "and old_non_pk_column_list are empty" }, ) if self.is_myrocks_table: # Enable rocksdb bulk load before loading data self.change_rocksdb_bulk_load(enable=True) # Enable rocksdb explicit commit before loading data self.change_explicit_commit(enable=True) for suffix in range(1, self.outfile_suffix_end + 1): self.load_chunk(column_list, suffix) # Print out information after every 10% chunks have been loaded # We won't show progress if the number of chunks is less than 50 if suffix % max(5, int(self.outfile_suffix_end / 10)) == 0: log.info( "Load progress: {}/{} chunks".format( suffix, self.outfile_suffix_end ) ) if self.is_myrocks_table: # Disable rocksdb bulk load after loading data self.change_rocksdb_bulk_load(enable=False) # Disable rocksdb explicit commit after loading data self.change_explicit_commit(enable=False) self.stats["time_in_load"] = time.time() - stage_start_time def check_max_statement_time_exists(self): """ Check whether current MySQL instance support MAX_STATEMENT_TIME which is only supported by WebScaleSQL """ if self.mysql_version.is_mysql8: return self.is_var_enabled("max_execution_time") else: # the max_statement_time is count in miliseconds try: self.query(sql.select_max_statement_time) return True except Exception: # if any excpetion raised here, we'll treat it as # MAX_STATEMENT_TIME is not supported log.warning("MAX_STATEMENT_TIME doesn't support in this MySQL") return False def append_to_exclude_id(self): """ Add all replayed IDs into tmp_table_exclude_id so that we won't replay again later """ self.execute_sql( sql.insert_into_select_from( into_table=self.tmp_table_exclude_id, into_col_list=(self.IDCOLNAME, self.DMLCOLNAME), from_table=self.tmp_table_include_id, from_col_list=(self.IDCOLNAME, self.DMLCOLNAME), ) ) def get_max_delta_id(self): """ Get current maximum delta table ID. """ result = self.query(sql.get_max_id_from(self.IDCOLNAME, self.delta_table_name)) # If no events has been replayed, max would return a string 'None' # instead of a pythonic None. So we should treat 'None' as 0 here if result[0]["max_id"] == "None": return 0 return result[0]["max_id"] @wrap_hook def replay_delete_row(self, sql, *ids): """ Replay delete type change @param sql: SQL statement to replay the changes stored in chg table @type sql: string @param ids: values of ID column from self.delta_table_name @type ids: list """ affected_row = self.execute_sql(sql, ids) if ( not self.eliminate_dups and not self.where and not self.skip_affected_rows_check ): if not affected_row != 0: raise OSCError("REPLAY_WRONG_AFFECTED", {"num": affected_row}) @wrap_hook def replay_insert_row(self, sql, *ids): """ Replay insert type change @param sql: SQL statement to replay the changes stored in chg table @type sql: string @param ids: values of ID column from self.delta_table_name @type ids: list """ affected_row = self.execute_sql(sql, ids) if ( not self.eliminate_dups and not self.where and not self.skip_affected_rows_check ): if not affected_row != 0: raise OSCError("REPLAY_WRONG_AFFECTED", {"num": affected_row}) @wrap_hook def replay_update_row(self, sql, *ids): """ Replay update type change @param sql: SQL statement to replay the changes stored in chg table @type sql: string @param row: single row of delta information from self.delta_table_name @type row: list """ self.execute_sql(sql, ids) def get_gap_changes(self): # See if there're some gaps we need to cover. Because there're some # transactions that may started before last replay snapshot but # committed afterwards, which will cause __OSC_ID_ smaller than # self.last_replayed_id delta = [] log.info( "Checking {} gap ids".format(len(self._replayed_chg_ids.missing_points())) ) for chg_id in self._replayed_chg_ids.missing_points(): row = self.query( sql.get_chg_row(self.IDCOLNAME, self.DMLCOLNAME, self.delta_table_name), (chg_id,), ) if bool(row): log.debug("Change {} appears now!".format(chg_id)) delta.append(row[0]) for row in delta: self._replayed_chg_ids.fill(row[self.IDCOLNAME]) log.info( "{} changes before last checkpoint ready for replay".format(len(delta)) ) return delta def divide_changes_to_group(self, chg_rows): """ Put consecutive changes with the same type into a group so that we can execute them in a single query to speed up replay @param chg_rows: list of rows returned from _chg select query @type chg_rows: list[dict] """ id_group = [] type_now = None for idx, chg in enumerate(chg_rows): # Start of the current group if type_now is None: type_now = chg[self.DMLCOLNAME] id_group.append(chg[self.IDCOLNAME]) # Dump when we are at the end of the changes if idx == len(chg_rows) - 1: yield type_now, id_group return # update type cannot be grouped elif type_now == self.DML_TYPE_UPDATE: yield type_now, id_group type_now = None id_group = [] # The next change is a different type, dump what we have now elif chg_rows[idx + 1][self.DMLCOLNAME] != type_now: yield type_now, id_group type_now = None id_group = [] # Reach the max group size, let's submit the query for now elif len(id_group) >= self.replay_group_size: yield type_now, id_group type_now = None id_group = [] # The next element will be the same as what we are now else: continue def replay_changes( self, single_trx=False, holding_locks=False, delta_id_limit=None ): """ Loop through all the existing events in __osc_chg table and replay the change @param single_trx: Replay all the changes in single transaction or not @type single_trx: bool """ stage_start_time = time.time() log.debug("Timeout for replay changes: {}".format(self.replay_timeout)) time_start = time.time() deleted, inserted, updated = 0, 0, 0 # all the changes to be replayed in this round will be stored in # tmp_table_include_id. Though change events may keep being generated, # we'll only replay till the end of temporary table if ( single_trx and not self.bypass_replay_timeout and self.check_max_statement_time_exists() ): replay_ms = self.replay_timeout * 1000 else: replay_ms = None if delta_id_limit: max_id_now = delta_id_limit else: max_id_now = self.get_max_delta_id() log.debug("max_id_now is %r / %r", max_id_now, self.replay_max_changes) if max_id_now > self.replay_max_changes: raise OSCError( "REPLAY_TOO_MANY_DELTAS", {"deltas": max_id_now, "max_deltas": self.replay_max_changes}, ) if self.detailed_mismatch_info or self.dump_after_checksum: # We need this information for better understanding of the checksum # mismatch issue log.info( "Replaying changes happened before change ID: {}".format(max_id_now) ) delta = self.get_gap_changes() # Only replay changes in this range (last_replayed_id, max_id_now] new_changes = self.query( sql.get_replay_row_ids( self.IDCOLNAME, self.DMLCOLNAME, self.delta_table_name, replay_ms, self.mysql_version.is_mysql8, ), ( self.last_replayed_id, max_id_now, ), ) self._replayed_chg_ids.extend([r[self.IDCOLNAME] for r in new_changes]) delta.extend(new_changes) log.info("Total {} changes to replay".format(len(delta))) # Generate all three possible replay SQL here, so that we don't waste # CPU time regenerating them for each replay event delete_sql = sql.replay_delete_row( self.new_table_name, self.delta_table_name, self.IDCOLNAME, self._pk_for_filter, ) update_sql = sql.replay_update_row( self.old_non_pk_column_list, self.new_table_name, self.delta_table_name, self.eliminate_dups, self.IDCOLNAME, self._pk_for_filter, ) insert_sql = sql.replay_insert_row( self.old_column_list, self.new_table_name, self.delta_table_name, self.IDCOLNAME, self.eliminate_dups, ) replayed = 0 replayed_total = 0 showed_pct = 0 for chg_type, ids in self.divide_changes_to_group(delta): # We only care about replay time when we are holding a write lock if ( holding_locks and not self.bypass_replay_timeout and time.time() - time_start > self.replay_timeout ): raise OSCError("REPLAY_TIMEOUT") replayed_total += len(ids) # Commit transaction after every replay_batch_szie number of # changes have been replayed if not single_trx and replayed > self.replay_batch_size: self.commit() self.start_transaction() replayed = 0 else: replayed += len(ids) # Use corresponding SQL to replay each type of changes if chg_type == self.DML_TYPE_DELETE: self.replay_delete_row(delete_sql, ids) deleted += len(ids) elif chg_type == self.DML_TYPE_UPDATE: self.replay_update_row(update_sql, ids) updated += len(ids) elif chg_type == self.DML_TYPE_INSERT: self.replay_insert_row(insert_sql, ids) inserted += len(ids) else: # We are not supposed to reach here, unless someone explicitly # insert a row with unknown type into _chg table during OSC raise OSCError("UNKOWN_REPLAY_TYPE", {"type_value": chg_type}) # Print progress information after every 10% changes have been # replayed. If there're no more than 100 changes to replay then # there'll be no such progress information progress_pct = int(replayed_total / len(delta) * 100) if progress_pct > showed_pct: log.info( "Replay progress: {}/{} changes".format( replayed_total + 1, len(delta) ) ) showed_pct += 10 # Commit for last batch if not single_trx: self.commit() self.last_replayed_id = max_id_now time_spent = time.time() - stage_start_time self.stats["time_in_replay"] = ( self.stats.setdefault("time_in_replay", 0) + time_spent ) log.info( "Replayed {} INSERT, {} DELETE, {} UPDATE in {:.2f} Seconds".format( inserted, deleted, updated, time_spent ) ) def set_innodb_tmpdir(self, innodb_tmpdir): try: self.execute_sql( sql.set_session_variable("innodb_tmpdir"), (innodb_tmpdir,) ) except MySQLdb.OperationalError as e: errnum, errmsg = e.args # data_dir cannot always be set to innodb_tmpdir due to # priviledge issue. Falling back to tmpdir if it happens # 1193: unknown variable # 1231: Failed to set because of priviledge error if errnum in (1231, 1193): log.warning( "Failed to set innodb_tmpdir, falling back to tmpdir: {}".format( errmsg ) ) else: raise @wrap_hook def recreate_non_unique_indexes(self): """ Re-create non-unique indexes onto the new table """ # Skip replaying changes for now, if don't have to recreate index if not self.droppable_indexes: return self.set_innodb_tmpdir(self.outfile_dir) # Execute alter table only if we have index to create if self.droppable_indexes: self.ddl_guard() log.info( "Recreating indexes: {}".format( ", ".join(col.name for col in self.droppable_indexes) ) ) self.execute_sql(sql.add_index(self.new_table_name, self.droppable_indexes)) @wrap_hook def analyze_table(self): """ Force to update internal optimizer statistics. So that we are less likely to hit bad execution plan because too many changes have been made """ # Analyze table has a query result, we have to use query here. # Otherwise we'll get a out of sync error self.query(sql.analyze_table(self.new_table_name)) self.query(sql.analyze_table(self.delta_table_name)) def compare_checksum(self, old_table_checksum, new_table_checksum): """ Given two list of checksum result generated by checksum_by_chunk, compare whether there's any difference between them @param old_table_checksum: checksum from old table @type old_table_checksum: list of list @param new_table_checksum: checksum from new table @type new_table_checksum: list of list """ if len(old_table_checksum) != len(new_table_checksum): log.error( "The total number of checksum chunks mismatch " "OLD={}, NEW={}".format( len(old_table_checksum), len(new_table_checksum) ) ) raise OSCError("CHECKSUM_MISMATCH") log.info("{} checksum chunks in total".format(len(old_table_checksum))) for idx, checksum_entry in enumerate(old_table_checksum): for col in checksum_entry: if not old_table_checksum[idx][col] == new_table_checksum[idx][col]: log.error( "checksum/count mismatch for chunk {} " "column `{}`: OLD={}, NEW={}".format( idx, col, old_table_checksum[idx][col], new_table_checksum[idx][col], ) ) log.error( "Number of rows for the chunk that cause the " "mismatch: OLD={}, NEW={}".format( old_table_checksum[idx]["cnt"], new_table_checksum[idx]["cnt"], ) ) log.error( "Current replayed max(__OSC_ID) of chg table {}".format( self.last_replayed_id ) ) raise OSCError("CHECKSUM_MISMATCH") def checksum_full_table(self): """ Running checksum in single query, this will be used only for tables which don't have primary in the old schema. See checksum_by_chunk for more detail """ # Calculate checksum for old table old_checksum = self.query( sql.checksum_full_table(self.table_name, self._old_table.column_list) ) # Calculate checksum for new table new_checksum = self.query( sql.checksum_full_table(self.new_table_name, self._old_table.column_list) ) self.commit() # Compare checksum if old_checksum and new_checksum: self.compare_checksum(old_checksum, new_checksum) def checksum_for_single_chunk(self, table_name, use_where, idx_for_checksum): """ Using the same set of session variable as chunk start point and calculate checksum for old table/new table. If assign is provided, current right boundry will be passed into range_start_vars as the start of next chunk """ return self.query( sql.checksum_by_chunk_with_assign( table_name, self.checksum_column_list, self._pk_for_filter, self.range_start_vars_array, self.range_end_vars_array, self.select_chunk_size, use_where, idx_for_checksum, ) )[0] def dump_current_chunk(self, use_where): """ Use select into outfile to dump the data in the previous chunk that caused checksum mismatch @param use_where: whether we should use session variable as selection boundary in where condition @type use_where: bool """ log.info("Dumping raw data onto local disk for further investigation") log.info("Columns will be dumped in following order: ") log.info(", ".join(self._pk_for_filter + self.checksum_column_list)) for table_name in [self.table_name, self.new_table_name]: if table_name == self.new_table_name: # index for new scehma can be any indexes that provides # uniqueness and covering old PK lookup idx_for_checksum = self.find_coverage_index() outfile = "{}.new".format(self.outfile) else: # index for old schema should always be PK idx_for_checksum = "PRIMARY" outfile = "{}.old".format(self.outfile) log.info("Dump offending chunk from {} into {}".format(table_name, outfile)) self.execute_sql( sql.dump_current_chunk( table_name, self.checksum_column_list, self._pk_for_filter, self.range_start_vars_array, self.select_chunk_size, idx_for_checksum, use_where, ), (outfile,), ) @wrap_hook def detailed_checksum(self): """ Yet another way of calculating checksum but it opens a longer trx than the default approach. By doing this we will able to print out the exact chunk of data that caused a checksum mismatch """ affected_rows = 1 use_where = False new_idx_for_checksum = self.find_coverage_index() old_idx_for_checksum = "PRIMARY" chunk_id = 0 while affected_rows: chunk_id += 1 old_checksum = self.checksum_for_single_chunk( self.table_name, use_where, old_idx_for_checksum ) new_checksum = self.checksum_for_single_chunk( self.new_table_name, use_where, new_idx_for_checksum ) affected_rows = old_checksum["_osc_chunk_cnt"] # Need to convert to List here because dict_values type will always # claim two sides as different if list(old_checksum.values()) != list(new_checksum.values()): log.info("Checksum mismatch detected for chunk {}: ".format(chunk_id)) log.info("OLD: {}".format(str(old_checksum))) log.info("NEW: {}".format(str(new_checksum))) self.dump_current_chunk(use_where) raise OSCError("CHECKSUM_MISMATCH") # Refresh where condition range for next select if affected_rows: self.refresh_range_start() use_where = True @wrap_hook def checksum_by_chunk(self, table_name, dump_after_checksum=False): """ Running checksum for all the existing data in new table. This is to make sure there's no data corruption after load and first round of replay """ checksum_result = [] # Checksum by chunk. This is pretty much the same logic as we've used # in select_table_into_outfile affected_rows = 1 use_where = False outfile_id = 0 if table_name == self.new_table_name: idx_for_checksum = self.find_coverage_index() outfile_prefix = "{}.new".format(self.outfile) else: idx_for_checksum = self._idx_name_for_filter outfile_prefix = "{}.old".format(self.outfile) while affected_rows: checksum = self.query( sql.checksum_by_chunk( table_name, self.checksum_column_list, self._pk_for_filter, self.range_start_vars_array, self.range_end_vars_array, self.select_chunk_size, use_where, idx_for_checksum, ) ) # Dump the data onto local disk for further investigation # This will be very helpful when there's a reproducable checksum # mismatch issue if dump_after_checksum: self.execute_sql( sql.dump_current_chunk( table_name, self.checksum_column_list, self._pk_for_filter, self.range_start_vars_array, self.select_chunk_size, idx_for_checksum, use_where, ), ("{}.{}".format(outfile_prefix, str(outfile_id)),), ) outfile_id += 1 # Refresh where condition range for next select if checksum: self.refresh_range_start() affected_rows = checksum[0]["cnt"] checksum_result.append(checksum[0]) use_where = True return checksum_result def need_checksum(self): """ Check whether we should checksum or not """ if self.skip_checksum: log.warning("Skip checksum because --skip-checksum is specified") return False # There's no point running a checksum compare for selective dump if self.where: log.warning("Skip checksum because --where is given") return False # If the collation of primary key column has been changed, then # it's high possible that the checksum will mis-match, because # the returning sequence after order by primary key may be vary # for different collations for pri_column in self._pk_for_filter: old_column_tmp = [ col for col in self._old_table.column_list if col.name == pri_column ] if old_column_tmp: old_column = old_column_tmp[0] new_column_tmp = [ col for col in self._new_table.column_list if col.name == pri_column ] if new_column_tmp: new_column = new_column_tmp[0] if old_column and new_column: if not is_equal(old_column.collate, new_column.collate): log.warning( "Collation of primary key column {} has been " "changed. Skip checksum ".format(old_column.name) ) return False # There's no way we can run checksum by chunk if the primary key cannot # be covered by any index of the new schema if not self.validate_post_alter_pk(): if self.skip_pk_coverage_check: log.warning( "Skipping checksuming because there's no unique index " "in new table schema can perfectly cover old primary key " "combination for search".format(old_column.name) ) return False else: # Though we have enough coverage for primary key doesn't # necessarily mean we can use it for checksum, it has to be an # unique index as well. Skip checksum if there's no such index if not self.find_coverage_index(): log.warning( "Skipping checksuming because there's no unique index " "in new table schema can perfectly cover old primary key " "combination for search".format(old_column.name) ) return False return True def need_checksum_for_changes(self): """ Check whether we should checksum for changes or not """ # We don't need to run checksum for changes, if we don't want checksum # at all if not self.need_checksum(): return False if self.is_full_table_dump: log.warning( "We're adding new primary key to the table. Skip running " "checksum for changes, because that's inefficient" ) return False return True @wrap_hook def checksum(self): """ Run checksum for all existing data in new table. We will do another around of checksum, but only for changes happened in between """ log.info("== Stage 4: Checksum ==") if not self.need_checksum(): return stage_start_time = time.time() if self.eliminate_dups: log.warning("Skip checksum, because --eliminate-duplicate " "specified") return # Replay outside of transaction so that we won't hit max allowed # transaction time, log.info("= Stage 4.1: Catch up before generating checksum =") self.replay_till_good2go(checksum=False) log.info("= Stage 4.2: Comparing checksum =") self.start_transaction() # To fill the gap between old and new table since last replay log.info("Replay changes to bring two tables to a comparable state") self.replay_changes(single_trx=True) # if we don't have a PK on old schema, then we are not able to checksum # by chunk. We'll do a full table scan for checksum instead if self.is_full_table_dump: return self.checksum_full_table() if not self.detailed_mismatch_info: log.info("Checksuming data from old table") old_table_checksum = self.checksum_by_chunk( self.table_name, dump_after_checksum=self.dump_after_checksum ) # We can calculate the checksum for new table outside the # transaction, because the data in new table is static without # replaying chagnes self.commit() log.info("Checksuming data from new table") new_table_checksum = self.checksum_by_chunk( self.new_table_name, dump_after_checksum=self.dump_after_checksum ) log.info("Compare checksum") self.compare_checksum(old_table_checksum, new_table_checksum) else: self.detailed_checksum() self.last_checksumed_id = self.last_replayed_id log.info("Checksum match between new and old table") self.stats["time_in_table_checksum"] = time.time() - stage_start_time @wrap_hook def replay_till_good2go(self, checksum): """ Keep replaying changes until the time spent in replay is below self.replay_timeout For table which has huge numbers of writes during OSC, we'll probably hit replay timeout if we call swap_tables directly after checksum. We will do several round iteration here in order to bring the number of un-played changes down to a proper level a proper level @param checksum: Run checksum for replayed changes or not @type checksum: bool """ log.info( "Replay at most {} more round(s) until we can finish in {} " "seconds".format(self.replay_max_attempt, self.replay_timeout) ) # Temporarily enable slow query log for slow replay statements self.execute_sql(sql.set_session_variable("long_query_time"), (1,)) for i in range(self.replay_max_attempt): log.info("Catchup Attempt: {}".format(i + 1)) start_time = time.time() # If checksum is required, then we need to make sure total time # spent in replay+checksum is below replay_timeout. if checksum and self.need_checksum(): self.start_transaction() log.info( "Catch up in order to compare checksum for the " "rows that have been changed" ) self.replay_changes(single_trx=True) self.checksum_for_changes(single_trx=False) else: # Break replay into smaller chunks if it's too big max_id_now = self.get_max_delta_id() while max_id_now - self.last_replayed_id > self.max_replay_batch_size: delta_id_limit = self.last_replayed_id + self.max_replay_batch_size log.info("Replay up to {}".format(delta_id_limit)) self.replay_changes(single_trx=False, delta_id_limit=delta_id_limit) self.replay_changes(single_trx=False, delta_id_limit=max_id_now) time_in_replay = time.time() - start_time if time_in_replay < self.replay_timeout: log.info( "Time spent in last round of replay is {:.2f}, which " "is less than replay_timeout: {} for final replay. " "We are good to proceed".format(time_in_replay, self.replay_timeout) ) break else: # We are not able to bring the replay time down to replay_timeout if not self.bypass_replay_timeout: raise OSCError("MAX_ATTEMPT_EXCEEDED", {"timeout": self.replay_timeout}) else: log.warning( "Proceed after max replay attempts exceeded. " "Because --bypass-replay-timeout is specified" ) @wrap_hook def checksum_by_replay_chunk(self, table_name): """ Run checksum for rows which have been touched by changes made after last round of checksum. """ # Generate a column string which contains all non-changed columns # wrapped with checksum function. checksum_result = [] id_limit = self.last_checksumed_id # Using the same batch size for checksum as we used for replaying while id_limit < self.last_replayed_id: result = self.query( sql.checksum_by_replay_chunk( table_name, self.delta_table_name, self.old_column_list, self._pk_for_filter, self.IDCOLNAME, id_limit, self.last_replayed_id, self.replay_batch_size, ) ) checksum_result.append(result[0]) id_limit += self.replay_batch_size return checksum_result @wrap_hook def checksum_for_changes(self, single_trx=False): """ This checksum will only run against changes made between last full table checksum and before swap table We assume A transaction has been opened before calling this function, and changes has been replayed @param single_trx: whether skip the commit call after checksum old table. This can prevent opening a transaction for too long when we don't actually need it @type single_trx: bool """ if self.eliminate_dups: log.warning("Skip checksum, because --elimiate-duplicate " "specified") return elif not self.need_checksum_for_changes(): return # Because chunk checksum use old pk combination for searching row # If we don't have a pk/uk on old table then it'll be very slow, so we # have to skip here elif self.is_full_table_dump: return else: log.info( "Running checksum for rows have been changed since " "last checksum from change ID: {}".format(self.last_checksumed_id) ) start_time = time.time() old_table_checksum = self.checksum_by_replay_chunk(self.table_name) # Checksum for the __new table should be issued inside the transcation # too. Otherwise those invisible gaps in the __chg table will show # up when calculating checksums new_table_checksum = self.checksum_by_replay_chunk(self.new_table_name) # After calculation checksums from both tables, we now can close the # transcation, if we want if not single_trx: self.commit() self.compare_checksum(old_table_checksum, new_table_checksum) self.last_checksumed_id = self.last_replayed_id self.stats["time_in_delta_checksum"] = self.stats.setdefault( "time_in_delta_checksum", 0 ) + (time.time() - start_time) @wrap_hook def apply_partition_differences( self, parts_to_drop: Optional[Set[str]], parts_to_add: Optional[Set[str]] ) -> None: # we can just drop partitions by name (ie, p[0-9]+), but to add # partitions we need the range value for each - get this from orig # table if parts_to_add: add_parts = [] for part_name in parts_to_add: part_value = self.partition_value_for_name(self.table_name, part_name) add_parts.append( "PARTITION {} VALUES LESS THAN ({})".format(part_name, part_value) ) add_parts_str = ", ".join(add_parts) add_sql = "ALTER TABLE `{}` ADD PARTITION ({})".format( self.new_table_name, add_parts_str ) log.info(add_sql) self.execute_sql(add_sql) if parts_to_drop: drop_parts_str = ", ".join(parts_to_drop) drop_sql = "ALTER TABLE `{}` DROP PARTITION {}".format( self.new_table_name, drop_parts_str ) log.info(drop_sql) self.execute_sql(drop_sql) @wrap_hook def partition_value_for_name(self, table_name: str, part_name: str) -> str: result = self.query( sql.fetch_partition_value, ( self._current_db, table_name, part_name, ), ) for r in result: return r["PARTITION_DESCRIPTION"] raise RuntimeError(f"No partition value found for {table_name} {part_name}") @wrap_hook def list_partition_names(self, table_name: str) -> List[str]: tbl_parts = [] result = self.query(sql.fetch_partition, (self._current_db, table_name)) for r in result: tbl_parts.append(r["PARTITION_NAME"]) if not tbl_parts: raise RuntimeError(f"No partition values found for {table_name}") return tbl_parts @wrap_hook def sync_table_partitions(self) -> None: """ If table partitions have changed on the original table, apply the same changes before swapping table, or we will likely break replication if using row-based. """ log.info("== Stage 5.1: Check table partitions are up-to-date ==") # we're using partitions in the ddl file, skip syncing anything if not self.rm_partition: return # not a partitioned table, nothing to do if not self.partitions: return # only apply this logic to RANGE partitioning, as other types # are usually static partition_method = self.get_partition_method( self._current_db, self.new_table_name ) if partition_method != "RANGE": return try: new_tbl_parts = self.list_partition_names(self.new_table_name) orig_tbl_parts = self.list_partition_names(self.table_name) parts_to_drop = set(new_tbl_parts) - set(orig_tbl_parts) parts_to_add = set(orig_tbl_parts) - set(new_tbl_parts) # information schema literally has the string None for # non-partitioned tables. Previous checks *should* prevent us # from hitting this. if "None" in parts_to_add or "None" in parts_to_drop: log.warning( "MySQL claims either %s or %s are not partitioned", self.new_table_name, self.table_name, ) return if parts_to_drop: log.info( "Partitions missing from source table " "to drop from new table %s: %s", self.new_table_name, ", ".join(parts_to_drop), ) if parts_to_add: log.info( "Partitions in source table to add to new table %s: %s", self.new_table_name, ", ".join(parts_to_add), ) self.apply_partition_differences(parts_to_drop, parts_to_add) except Exception: log.exception( "Unable to sync new table %s with orig table %s partitions", self.new_table_name, self.table_name, ) @wrap_hook def swap_tables(self): """ Flip the table name while holding the write lock. All operations during this stage will be executed inside a single transaction. """ if self.stop_before_swap: return True log.info("== Stage 6: Swap table ==") self.stop_slave_sql() self.execute_sql(sql.set_session_variable("autocommit"), (0,)) self.start_transaction() stage_start_time = time.time() self.lock_tables((self.new_table_name, self.table_name, self.delta_table_name)) log.info("Final round of replay before swap table") self.replay_changes(single_trx=True, holding_locks=True) # We will not run delta checksum here, because there will be an error # like this, if we run a nested query using `NOT EXISTS`: # SQL execution error: [1100] Table 't' was not locked with LOCK TABLES self.execute_sql(sql.rename_table(self.table_name, self.renamed_table_name)) self.table_swapped = True self.add_drop_table_entry(self.renamed_table_name) self.execute_sql(sql.rename_table(self.new_table_name, self.table_name)) log.info("Table has successfully swapped, new schema takes effect now") self._cleanup_payload.remove_drop_table_entry( self._current_db, self.new_table_name ) self.commit() self.unlock_tables() self.stats["time_in_lock"] = self.stats.setdefault("time_in_lock", 0) + ( time.time() - stage_start_time ) self.execute_sql(sql.set_session_variable("autocommit"), (1,)) self.start_slave_sql() def rename_back(self): """ If the orignal table was successfully renamed to _old but the second rename operation failed, rollback the first renaming """ if ( self.table_swapped and self.table_exists(self.renamed_table_name) and not self.table_exists(self.table_name) ): self.unlock_tables() self.execute_sql(sql.rename_table(self.renamed_table_name, self.table_name)) @wrap_hook def cleanup(self): """ Cleanup all the temporary thing we've created so far """ log.info("== Stage 7: Cleanup ==") # Close current connection to free up all the temporary resource # and locks try: self.rename_back() self.start_slave_sql() if self.is_myrocks_table and self.is_myrocks_ttl_table: self.enable_ttl_for_myrocks() self.release_osc_lock() self.close_conn() except Exception: log.exception( "Ignore following exception, because we want to try our " "best to cleanup, and free disk space:" ) self._cleanup_payload.mysql_user = self.mysql_user self._cleanup_payload.mysql_pass = self.mysql_pass self._cleanup_payload.socket = self.socket self._cleanup_payload.get_conn_func = self.get_conn_func self._cleanup_payload.cleanup(self._current_db) def print_stats(self): log.info("Time in dump: {:.3f}s".format(self.stats.get("time_in_dump", 0))) log.info("Time in load: {:.3f}s".format(self.stats.get("time_in_load", 0))) log.info("Time in replay: {:.3f}s".format(self.stats.get("time_in_replay", 0))) log.info( "Time in table checksum: {:.3f}s".format( self.stats.get("time_in_table_checksum", 0) ) ) log.info( "Time in delta checksum: {:.3f}s".format( self.stats.get("time_in_delta_checksum", 0) ) ) log.info( "Time holding locks: {:.3f}s".format(self.stats.get("time_in_lock", 0)) ) @wrap_hook def run_ddl(self, db, sql): try: time_started = time.time() self._new_table = parse_create(sql) self._current_db = db self._current_db_dir = util.dirname_for_db(db) self.init_connection(db) self.init_table_obj() self.determine_outfile_dir() if self.force_cleanup: self.cleanup_with_force() if self.has_desired_schema(): self.release_osc_lock() return self.unblock_no_pk_creation() self.pre_osc_check() self.create_delta_table() self.create_copy_table() self.create_triggers() self.start_snapshot() self.select_table_into_outfile() self.drop_non_unique_indexes() self.load_data() self.recreate_non_unique_indexes() self.analyze_table() self.checksum() log.info("== Stage 5: Catch up to reduce time for holding lock ==") self.replay_till_good2go(checksum=self.skip_delta_checksum) self.sync_table_partitions() self.swap_tables() self.reset_no_pk_creation() self.cleanup() self.print_stats() self.stats["wall_time"] = time.time() - time_started except ( MySQLdb.OperationalError, MySQLdb.ProgrammingError, MySQLdb.IntegrityError, ) as e: errnum, errmsg = e.args log.error( "SQL execution error: [{}] {}\n" "When executing: {}\n" "With args: {}".format( errnum, errmsg, self._sql_now, self._sql_args_now ) ) # 2013 stands for lost connection to MySQL # 2006 stands for MySQL has gone away # Both means we have been killed if errnum in (2006, 2013) and self.skip_cleanup_after_kill: # We can skip dropping table, and removing files. # However leaving trigger around may break # replication which is really bad. So trigger is the only # thing we need to clean up in this case self._cleanup_payload.remove_drop_table_entry( self._current_db, self.new_table_name ) self._cleanup_payload.remove_drop_table_entry( self._current_db, self.delta_table_name ) self._cleanup_payload.remove_all_file_entries() if not self.keep_tmp_table: self.cleanup() raise OSCError( "GENERIC_MYSQL_ERROR", { "stage": "running DDL on db '{}'".format(db), "errnum": errnum, "errmsg": errmsg, }, mysql_err_code=errnum, ) except Exception as e: log.exception( "{0} Exception raised, start to cleanup before exit {0}".format( "-" * 10 ) ) # We want keep the temporary table for further investigation if not self.keep_tmp_table: self.cleanup() if not isinstance(e, OSCError): # It's a python exception raise OSCError("OSC_INTERNAL_ERROR", {"msg": str(e)}) else: raise