""" 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. """ from __future__ import absolute_import, division, print_function, unicode_literals import copy import logging from enum import Enum from osc.lib.sqlparse import CreateParser from .models import ( EnumColumn, escape, is_equal, PartitionConfig, SetColumn, TimestampColumn, ) log: logging.Logger = logging.getLogger(__name__) class BaseAlterType(Enum): pass # Side note the algorithm that could be chosen for DDL. See https://fburl.com/bxftsout class ColAlterType(BaseAlterType): CHANGE_COL_DEFAULT_VAL = "change_col_default_val" # instant REORDER_COL = "reorder_col" # rebuild ADD_COL = "add_col" # instant ADD_AUTO_INC_COL = "add_auto_inc_col" # inplace DROP_COL = "drop_col" # rebuild CHANGE_COL_DATA_TYPE = "change_col_data_type" # copy CHANGE_NULL = "change_null" # rebuild CHANGE_ENUM = "change_enum" # instant/copy CHANGE_SET = "change_set" # instant/copy CHANGE_COL_CHARSET = "change_col_charset" CHANGE_COL_COLLATE = "change_col_collate" CHANGE_COL_COMMENT = "change_col_comment" class IndexAlterType(BaseAlterType): CHANGE_INDEX_TYPE = "change_index_type" # instant. index type is hash/btree CHANGE_UNIQUE_CONSTRAINT = "change_unique_constraint" CHANGE_INDEX_KEY_BLOCK_SIZE = "change_index_key_block_size" CHANGE_KEY_TYPE = "change_key_type" # key type is FULLTEXT/SPATIAL CHANGE_INDEX_COMMENT = "change_index_comment" ADD_INDEX = "add_index" # inplace DROP_INDEX = "drop_index" # inplace, metadata only CHANGE_PK = "change_pk" # copy BECOME_UNIQUE_INDEX = "become_unique_index" class TableAlterType(BaseAlterType): CHANGE_ROW_FORMAT = "change_row_format" CHANGE_TABLE_KEY_BLOCK_SIZE = "change_table_key_block_size" CHANGE_TABLE_CHARSET = "change_table_charset" CHANGE_TABLE_COLLATE = "change_table_collate" CHANGE_TABLE_COMMENT = "change_table_comment" CHANGE_ENGINE = "change_engine" CHANGE_AUTO_INC_VAL = "change_auto_inc_val" # inplace class PartitionAlterType(BaseAlterType): ADD_PARTITION = "add_partition" DROP_PARTITION = "drop_partition" REORGANIZE_PARTITION = "reorganize_partition" ADD_PARTITIONING = "add_partitioning" REMOVE_PARTITIONING = "remove_partitioning" CHANGE_PARTITIONING = "change_partitioning" INVALID_PARTITIONING = "invalid_partitioning" class NewMysql80FeatureAlterType(BaseAlterType): JSON = "json" DESC_INDEX = "desc_index" INSTANT_DDLS = { ColAlterType.CHANGE_COL_DEFAULT_VAL, ColAlterType.ADD_COL, IndexAlterType.CHANGE_INDEX_TYPE, } class TableOptionDiff(object): def __init__(self, option_name, value): self.option_name = option_name self.value = value def to_sql(self): return "{}={}".format(self.option_name, self.value) class SchemaDiff(object): """ Representing the difference between two Table object """ def __init__(self, left, right, ignore_partition=False): self.left = left self.right = right self.attrs_to_check = [ "charset", "collate", "comment", "engine", "key_block_size", "name", "row_format", ] self.ignore_partition = ignore_partition if not ignore_partition: self.attrs_to_check.append("partition") self._alter_types = set() def get_dropped_cols(self): new_column_names = [col.name for col in self.right.column_list] return [ col for col in self.left.column_list if col.name not in new_column_names ] def get_added_cols(self): old_column_names = [col.name for col in self.left.column_list] return [ col for col in self.right.column_list if col.name not in old_column_names ] def _calculate_diff(self): diffs = { "removed": [], "added": [], # Customized messages "msgs": [], # Any attributes that were modified "attrs_modified": [], } # We are copying here since we want to change the col list. # Shallow copy should be enough here col_left_copy = copy.copy(self.left.column_list) col_right_copy = copy.copy(self.right.column_list) col_left_dict = {col.name: col for col in col_left_copy} col_right_dict = {col.name: col for col in col_right_copy} for col in self.get_dropped_cols(): diffs["removed"].append(col) col_left_copy.remove(col) for col in self.get_added_cols(): diffs["added"].append(col) col_right_copy.remove(col) column_changes = [] for col in set(col_left_dict.keys()) & set(col_right_dict.keys()): if col_left_dict[col] != col_right_dict[col]: column_changes.append( f"Previous column: {col_left_dict[col]}, " f"desired column: {col_right_dict[col]}" ) if column_changes: diffs["msgs"].append("Column attrs changes detected:") diffs["msgs"].append("\n".join(column_changes)) # Two tables have different col order if sorted(col_left_copy, key=lambda col: col.name) == sorted( col_right_copy, key=lambda col: col.name ): old_order = [] new_order = [] for col1, col2 in zip(col_left_copy, col_right_copy): if col1 != col2: old_order.append(col1.name) new_order.append(col2.name) if old_order: diffs["msgs"].append("Column order mismatch was detected:") diffs["msgs"].append("- " + ", ".join(old_order)) diffs["msgs"].append("+ " + ", ".join(new_order)) for idx in self.left.indexes: if idx not in self.right.indexes: diffs["removed"].append(idx) for idx in self.right.indexes: if idx not in self.left.indexes: diffs["added"].append(idx) if self.left.primary_key != self.right.primary_key: if self.left.primary_key.column_list: diffs["removed"].append(self.left.primary_key) if self.right.primary_key.column_list: diffs["added"].append(self.right.primary_key) for attr in self.attrs_to_check: tbl_option_old = getattr(self.left, attr) tbl_option_new = getattr(self.right, attr) if not is_equal(tbl_option_old, tbl_option_new): diffs["removed"].append(TableOptionDiff(attr, tbl_option_old)) diffs["added"].append(TableOptionDiff(attr, tbl_option_new)) diffs["attrs_modified"].append(attr) return diffs def __str__(self): if self.left == self.right: return "No difference" else: diff_strs = [] diffs = self._calculate_diff() for diff in diffs["removed"]: diff_strs.append("- " + diff.to_sql()) for diff in diffs["added"]: diff_strs.append("+ " + diff.to_sql()) for diff in diffs["msgs"]: diff_strs.append(diff) for attr in diffs["attrs_modified"]: diff_strs.append(f"attrs_modified: {attr}") diff_str = "\n".join(diff_strs) return diff_str def diffs(self): return self._calculate_diff() @property def alter_types(self): if not self._alter_types: self.to_sql() return self._alter_types def add_alter_type(self, ddl_alter_type): self._alter_types.add(ddl_alter_type) def _gen_col_sql(self): """ Generate the column section for ALTER TABLE statement """ segments = [] old_columns = {col.name: col for col in self.left.column_list} new_columns = {col.name: col for col in self.right.column_list} old_column_names = [col.name for col in self.left.column_list] new_column_names = [col.name for col in self.right.column_list] # Drop columns for col in self.left.column_list: if col.name not in new_columns.keys(): segments.append("DROP `{}`".format(escape(col.name))) old_column_names.remove(col.name) self.add_alter_type(ColAlterType.DROP_COL) # Add columns # If the added column is not at the end, recognize that as reordering columns handled_cols = [] for idx, col in enumerate(self.right.column_list): if col.name not in old_columns.keys(): if idx == 0: position = "FIRST" if ( old_column_names and ColAlterType.DROP_COL not in self._alter_types ): self.add_alter_type(ColAlterType.REORDER_COL) old_column_names = [col.name] + old_column_names else: position = "AFTER `{}`".format( escape(self.right.column_list[idx - 1].name) ) new_idx = ( old_column_names.index(self.right.column_list[idx - 1].name) + 1 ) if ( new_idx != len(old_column_names) and ColAlterType.DROP_COL not in self._alter_types ): self.add_alter_type(ColAlterType.REORDER_COL) old_column_names = ( old_column_names[:new_idx] + [col.name] + old_column_names[new_idx:] ) handled_cols.append(col.name) self.add_alter_type(ColAlterType.ADD_COL) if col.column_type == "JSON": self.add_alter_type(NewMysql80FeatureAlterType.JSON) if col.auto_increment: self.add_alter_type(ColAlterType.ADD_AUTO_INC_COL) segments.append("ADD {} {}".format(col.to_sql(), position)) # Adjust position # The idea here is to compare column ancestor if they are the same between # old and new column list, this means the position of this particular # column hasn't been changed. Otherwise add a MODIFY clause to change the # position for idx, col_name in enumerate(new_column_names): # If the column is recently added, then skip because it's already # in the DDL if col_name in handled_cols: continue # Get column definition col = new_columns[col_name] old_pos = old_column_names.index(col_name) # If the first column is diferent, we need to adjust the sequence if idx == 0: if old_pos == 0: continue segments.append("MODIFY {} FIRST".format(col.to_sql())) handled_cols.append(col_name) self.add_alter_type(ColAlterType.REORDER_COL) continue # If this column has the same ancestor then it means there's no sequence # adjustment needed if new_column_names[idx - 1] == old_column_names[old_pos - 1]: continue segments.append( "MODIFY {} AFTER `{}`".format( col.to_sql(), escape(new_column_names[idx - 1]) ) ) handled_cols.append(col_name) self.add_alter_type(ColAlterType.REORDER_COL) # Modify columns for col in self.right.column_list: if col.name in old_columns and col != old_columns[col.name]: # If the column has been taken care of because of sequence change # previously we can skip the work here if col.name in handled_cols: continue self._update_col_attrs_changes(col, old_columns[col.name]) segments.append("MODIFY {}".format(col.to_sql())) return segments def _is_null_change(self, old_col, new_col): if isinstance(old_col, TimestampColumn): old_col.explicit_ts_default() if isinstance(new_col, TimestampColumn): new_col.explicit_ts_default() return old_col.nullable != new_col.nullable def _is_col_default_change(self, old_col, new_col): if isinstance(old_col, TimestampColumn): old_col.explicit_ts_default() if isinstance(new_col, TimestampColumn): new_col.explicit_ts_default() return not old_col.has_same_default(new_col) def _update_col_attrs_changes(self, new_col, old_col): if ( new_col.column_type != old_col.column_type or new_col.length != old_col.length ): self.add_alter_type(ColAlterType.CHANGE_COL_DATA_TYPE) if ( self._is_col_default_change(old_col, new_col) and ColAlterType.CHANGE_COL_DATA_TYPE not in self._alter_types ): self.add_alter_type(ColAlterType.CHANGE_COL_DEFAULT_VAL) if ( self._is_null_change(old_col, new_col) and ColAlterType.CHANGE_COL_DATA_TYPE not in self._alter_types ): self.add_alter_type(ColAlterType.CHANGE_NULL) if ( isinstance(new_col, EnumColumn) and isinstance(old_col, EnumColumn) and new_col.enum_list != old_col.enum_list ): self.add_alter_type(ColAlterType.CHANGE_ENUM) if ( isinstance(new_col, SetColumn) and isinstance(old_col, SetColumn) and new_col.set_list != old_col.set_list ): self.add_alter_type(ColAlterType.CHANGE_SET) if new_col.charset != old_col.charset: self.add_alter_type(ColAlterType.CHANGE_COL_CHARSET) if new_col.collate != old_col.collate: self.add_alter_type(ColAlterType.CHANGE_COL_COLLATE) if new_col.comment != old_col.comment: self.add_alter_type(ColAlterType.CHANGE_COL_COMMENT) def _gen_idx_sql(self): """ Generate the index section for ALTER TABLE statement """ segments = [] # Drop index for idx in self.left.indexes: if idx not in self.right.indexes: segments.append("DROP KEY `{}`".format(escape(idx.name))) self.add_alter_type(IndexAlterType.DROP_INDEX) # Add index for idx in self.right.indexes: if idx not in self.left.indexes: segments.append("ADD {}".format(idx.to_sql())) self.add_alter_type(IndexAlterType.ADD_INDEX) self._update_index_attrs_changes(idx.name) self._update_desc_index_type(idx.column_list) if self.left.primary_key and not self.right.primary_key: segments.append("DROP PRIMARY KEY") self.add_alter_type(IndexAlterType.CHANGE_PK) elif ( not self.left.primary_key.column_list and self.right.primary_key.column_list ): segments.append("ADD {}".format(self.right.primary_key.to_sql())) self.add_alter_type(IndexAlterType.CHANGE_PK) self._update_desc_index_type(self.right.primary_key.column_list) elif self.left.primary_key != self.right.primary_key: segments.append("DROP PRIMARY KEY") segments.append("ADD {}".format(self.right.primary_key.to_sql())) self.add_alter_type(IndexAlterType.CHANGE_PK) self._update_desc_index_type(self.right.primary_key.column_list) return segments def _update_desc_index_type(self, cols) -> None: for col in cols: if col.order == "DESC": self.add_alter_type(NewMysql80FeatureAlterType.DESC_INDEX) def _update_index_attrs_changes(self, idx_name): old_indexes = {idx.name: idx for idx in self.left.indexes} new_indexes = {idx.name: idx for idx in self.right.indexes} if ( idx_name not in old_indexes and idx_name in new_indexes and new_indexes[idx_name].is_unique ): self.add_alter_type(IndexAlterType.BECOME_UNIQUE_INDEX) if not (idx_name in old_indexes and idx_name in new_indexes): return attrs = ["key_block_size", "comment", "is_unique", "key_type", "using"] for attr in attrs: if not is_equal( getattr(old_indexes[idx_name], attr), getattr(new_indexes[idx_name], attr), ): if attr == "key_block_size": self.add_alter_type(IndexAlterType.CHANGE_INDEX_KEY_BLOCK_SIZE) elif attr == "comment": self.add_alter_type(IndexAlterType.CHANGE_INDEX_COMMENT) elif attr == "is_unique": if ( new_indexes[idx_name].is_unique and not old_indexes[idx_name].is_unique ): self.add_alter_type(IndexAlterType.BECOME_UNIQUE_INDEX) self.add_alter_type(IndexAlterType.CHANGE_UNIQUE_CONSTRAINT) elif attr == "key_type": self.add_alter_type(IndexAlterType.CHANGE_KEY_TYPE) elif attr == "using": self.add_alter_type(IndexAlterType.CHANGE_INDEX_TYPE) def _gen_tbl_attr_sql(self): """ Generate the table attribute section for ALTER TABLE statement """ segments = [] for attr in self.attrs_to_check: tbl_option_old = getattr(self.left, attr) tbl_option_new = getattr(self.right, attr) if not is_equal(tbl_option_old, tbl_option_new): # when tbl_option_new is None, do "alter table xxx attr=None" won't work if attr == "comment" and tbl_option_new is None: segments.append("{}={}".format(attr, "''")) elif attr == "row_format" and tbl_option_new is None: segments.append("{}={}".format(attr, "default")) elif attr == "partition" and self.ignore_partition is False: self.generate_table_partition_operations(tbl_option_new, segments) else: segments.append("{}={}".format(attr, tbl_option_new)) # populate alter types data if attr == "row_format": self.add_alter_type(TableAlterType.CHANGE_ROW_FORMAT) elif attr == "key_block_size": self.add_alter_type(TableAlterType.CHANGE_TABLE_KEY_BLOCK_SIZE) elif attr == "charset": self.add_alter_type(TableAlterType.CHANGE_TABLE_CHARSET) elif attr == "collate": self.add_alter_type(TableAlterType.CHANGE_TABLE_COLLATE) elif attr == "comment": self.add_alter_type(TableAlterType.CHANGE_TABLE_COMMENT) elif attr == "engine": self.add_alter_type(TableAlterType.CHANGE_ENGINE) # we don't want to alter auto_increment value in db, just record the alter type if not is_equal(self.left.auto_increment, self.right.auto_increment): self.add_alter_type(TableAlterType.CHANGE_AUTO_INC_VAL) return segments def generate_table_partition_operations( self, partition_attr_value, segments ) -> None: """ Check difference between partition configurations """ try: old_tbl_parts = PartitionConfig() new_tbl_parts = PartitionConfig() if self.left.partition: old_tbl_parts = CreateParser.partition_to_model( CreateParser.parse_partitions(self.left.partition) ) if self.right.partition: new_tbl_parts = CreateParser.partition_to_model( CreateParser.parse_partitions(self.right.partition) ) # Check for partitioning type change # The case of partition to no partition is # not handled here. if ( partition_attr_value and old_tbl_parts.get_type() != new_tbl_parts.get_type() ): if not old_tbl_parts.get_type(): self.add_alter_type(PartitionAlterType.ADD_PARTITIONING) else: self.add_alter_type(PartitionAlterType.CHANGE_PARTITIONING) segments.append( "{}".format(partition_attr_value.lstrip(" ").rstrip(" ")) ) return pdef_type = old_tbl_parts.get_type() if ( pdef_type == PartitionConfig.PTYPE_HASH or pdef_type == PartitionConfig.PTYPE_KEY ): self.populate_alter_substatement_for_add_or_drop_table_for_hash_or_key( old_tbl_parts.num_partitions, new_tbl_parts.num_partitions, partition_attr_value, segments, ) return parts_to_drop = [] parts_to_add = [] [ is_valid_alter, is_reorganization, ] = self.could_generate_sql_substatement_for_partition_reorganization( old_tbl_parts, new_tbl_parts, pdef_type, parts_to_drop, parts_to_add, segments, ) if not is_valid_alter: self.add_alter_type(PartitionAlterType.INVALID_PARTITIONING) log.warning("job failed - alter partition operation sanity check") return if not is_reorganization: self.populate_alter_substatement_for_add_or_drop_table_for_range_or_list( parts_to_drop, parts_to_add, partition_attr_value, segments ) except Exception: self.add_alter_type(PartitionAlterType.INVALID_PARTITIONING) log.exception("Unable to sync new table with orig table") def could_generate_sql_substatement_for_partition_reorganization( self, old_tbl_parts, new_tbl_parts, pdef_type, parts_to_drop, parts_to_add, segments, ) -> [bool, bool]: old_pname_to_pdefs_dict = {} old_pvalues_to_pdefs_dict = {} old_pnames = [] for pdef in old_tbl_parts.part_defs: old_pnames.append(pdef.pdef_name) old_pname_to_pdefs_dict[pdef.pdef_name] = pdef value_list = self.get_value_list(pdef.pdef_value_list) old_pvalues_to_pdefs_dict[value_list] = pdef new_pname_to_pdefs_dict = {} new_pvalues_to_pdefs_dict = {} new_pnames = [] for pdef in new_tbl_parts.part_defs: new_pnames.append(pdef.pdef_name) new_pname_to_pdefs_dict[pdef.pdef_name] = pdef value_list = self.get_value_list(pdef.pdef_value_list) new_pvalues_to_pdefs_dict[value_list] = pdef old_tbl_part_defs = old_tbl_parts.part_defs new_tbl_part_defs = new_tbl_parts.part_defs for k, v in old_pvalues_to_pdefs_dict.items(): if ( k not in new_pvalues_to_pdefs_dict.keys() or v.pdef_name != new_pvalues_to_pdefs_dict[k].pdef_name ): parts_to_drop.append(v) max_value_added = False # Ensuring "MAXVALUE" to be the last of the list, if present since sorting can alter the order for k, v in sorted(new_pvalues_to_pdefs_dict.items(), key=lambda item: item[0]): if ( k not in old_pvalues_to_pdefs_dict.keys() or v.pdef_name != old_pvalues_to_pdefs_dict[k].pdef_name ): if k != "MAXVALUE": parts_to_add.append(v) else: max_value_added = True max_value_entry = v if max_value_added: parts_to_add.append(max_value_entry) if parts_to_drop and parts_to_add: # check for partition reorganization since we don't allow add/drop # in one alter statement return self.got_reorganize_subsql_after_add_drop_partition_detection( pdef_type, old_tbl_part_defs, new_tbl_part_defs, segments ) elif parts_to_drop or parts_to_add: if parts_to_drop: self.add_alter_type(PartitionAlterType.DROP_PARTITION) if parts_to_add: self.add_alter_type(PartitionAlterType.ADD_PARTITION) return [True, False] return [False, False] def got_reorganize_subsql_after_add_drop_partition_detection( self, pdef_type, old_tbl_part_defs, new_tbl_part_defs, segments, ) -> [bool, bool]: old_tbl_end_index = len(old_tbl_part_defs) - 1 new_tbl_end_index = len(new_tbl_part_defs) - 1 # this verification takes place only after the checks for add/drop tables # if this fails, it means the user is implicitly trying to perform multiple # alter operations in a single statement which is not supported. # Any reorganization should satisfy the min max constraints as well as # not drop individual elements. if pdef_type == PartitionConfig.PTYPE_RANGE: # the new range max should be higher or equal to the old range max # dateime/integers should be handled # MAXVALUE is extracted as a string in PartitionModel, hence need to handle # it separately if not isinstance( new_tbl_part_defs[new_tbl_end_index].pdef_value_list, str ) and ( isinstance(old_tbl_part_defs[old_tbl_end_index].pdef_value_list, str) or ( self.old_range_is_higher_than_new_range( old_tbl_part_defs[old_tbl_end_index].pdef_value_list[0], new_tbl_part_defs[new_tbl_end_index].pdef_value_list[0], ) ) ): return [False, False] # The preliminiary check for split/merge is done at this point # For list we are good to go and form a reorganize statement # but range needs to satisfy the inner strict range limits old_tbl_start_index = 0 new_tbl_start_index = 0 j = 0 for i in range(0, old_tbl_end_index + 1): if j <= new_tbl_end_index and ( old_tbl_part_defs[i].pdef_name != new_tbl_part_defs[j].pdef_name or old_tbl_part_defs[i].pdef_value_list != new_tbl_part_defs[j].pdef_value_list ): old_tbl_start_index = i new_tbl_start_index = j break j = j + 1 rj = new_tbl_end_index for ri in reversed(range(old_tbl_end_index + 1)): if rj >= 0 and ( old_tbl_part_defs[ri].pdef_name != new_tbl_part_defs[rj].pdef_name or old_tbl_part_defs[ri].pdef_value_list != new_tbl_part_defs[rj].pdef_value_list ): old_tbl_end_index = ri new_tbl_end_index = rj break rj = rj - 1 # the range value before the end should match exactly if pdef_type == PartitionConfig.PTYPE_RANGE: if ri >= 0 and ri < len(old_tbl_part_defs) - 1: if ( old_tbl_part_defs[ri].pdef_value_list != new_tbl_part_defs[rj].pdef_value_list ): return [False, False] source_partitions_reorganized = [] for ni in range(old_tbl_start_index, old_tbl_end_index + 1): source_partitions_reorganized.append(old_tbl_part_defs[ni].pdef_name) source_partitions_reorganized_sql = (", ").join(source_partitions_reorganized) target_partitions_reorganized = [] for ni in range(new_tbl_start_index, new_tbl_end_index + 1): partition = new_tbl_part_defs[ni] value_list = self.get_value_list(partition.pdef_value_list) target_partitions_reorganized.append( "PARTITION {} VALUES {} ({})".format( partition.pdef_name, self.partition_definition_type(partition.pdef_type), value_list, ) ) target_partitions_reorganized_sql = (", ").join(target_partitions_reorganized) sub_alter_sql = "REORGANIZE PARTITION {} INTO ({})".format( source_partitions_reorganized_sql, target_partitions_reorganized_sql ) self.add_alter_type(PartitionAlterType.REORGANIZE_PARTITION) segments.append(sub_alter_sql) return [True, True] def get_value_list(self, pdef_value_list): if isinstance(pdef_value_list, str): value_list = pdef_value_list else: value_list = ", ".join(pdef_value_list) return value_list def old_range_is_higher_than_new_range(self, left_range, right_range) -> bool: return left_range > right_range def populate_alter_substatement_for_add_or_drop_table_for_range_or_list( self, parts_to_drop: PartitionConfig, parts_to_add: PartitionConfig, partition_attr_value, segments, ) -> None: if parts_to_add and parts_to_drop: return if parts_to_add: add_parts = [] for partition in parts_to_add: value_list = self.get_value_list(partition.pdef_value_list) add_parts.append( "PARTITION {} VALUES {} ({})".format( partition.pdef_name, self.partition_definition_type(partition.pdef_type), value_list, ) ) segments.append("ADD PARTITION (" + ", ".join(add_parts) + ")") elif parts_to_drop: # When we reach here, we cannot have a partition clause mentioned # for list/range without associated partitions as the partition model # should have flagged earlier. if not partition_attr_value: self.add_alter_type(PartitionAlterType.REMOVE_PARTITIONING) segments.append("REMOVE PARTITIONING") else: dropped_parts = [] for partition in parts_to_drop: dropped_parts.append("{}".format(partition.pdef_name)) segments.append("DROP PARTITION " + ", ".join(dropped_parts)) def populate_alter_substatement_for_add_or_drop_table_for_hash_or_key( self, old_partition_count, new_partition_count, partition_attr_value, segments, ) -> None: if old_partition_count == new_partition_count: return if old_partition_count < new_partition_count: self.add_alter_type(PartitionAlterType.ADD_PARTITION) segments.append( "ADD PARTITION PARTITIONS {}".format( new_partition_count - old_partition_count ) ) return if not partition_attr_value: self.add_alter_type(PartitionAlterType.REMOVE_PARTITIONING) segments.append("REMOVE PARTITIONING") return if new_partition_count == 0: # Cannot remove all partitions, use DROP TABLE instead self.add_alter_type(PartitionAlterType.INVALID_PARTITIONING) return self.add_alter_type(PartitionAlterType.DROP_PARTITION) segments.append( "COALESCE PARTITION {}".format(old_partition_count - new_partition_count) ) def partition_definition_type(self, def_type): return PartitionConfig.TYPE_MAP[def_type] def valid_partitioning_alter(self, type): return ( isinstance(type, PartitionAlterType) and type != PartitionAlterType.INVALID_PARTITIONING ) def to_sql(self): """ Generate an ALTER TABLE statement that can bring the schema from left to right """ segments = [] segments.extend(self._gen_col_sql()) segments.extend(self._gen_idx_sql()) segments.extend(self._gen_tbl_attr_sql()) if segments: return "ALTER TABLE `{}` {}".format( escape(self.right.name), ", ".join(segments) ) def get_type_conv_columns(old_obj, new_obj): """ Return a list of columns that involve type conversion when transit from left to right """ type_conv_cols = [] current_cols = {c.name: c for c in old_obj.column_list} new_cols = {c.name: c for c in new_obj.column_list} # find columns that will involve type conversions for name, old_col in current_cols.items(): new_col = new_cols.get(name) # this column isn't in the new schema, so it # doesn't matter if new_col is None: continue # Type changes are considered as type conversion if new_col.column_type != old_col.column_type: type_conv_cols.append(old_col) else: # Length change also considered as type conversion if new_col.length != old_col.length: type_conv_cols.append(old_col) return type_conv_cols def need_default_ts_bootstrap(old_obj, new_obj): """ Check when going from old schema to new, whether bootstraping column using CURRENT_TIMESTAMP is involved. This is normally dangerous thing to do out of replication and will be disallowed by default from OSC perspective """ current_cols = {c.name: c for c in old_obj.column_list} new_cols = {c.name: c for c in new_obj.column_list} # find columns that will involve type conversions for name, new_col in new_cols.items(): old_col = current_cols.get(name) # This check only applies to column types that support default ts value if new_col.column_type not in ["TIMESTAMP", "DATE", "DATETIME"]: continue if new_col.column_type == "TIMESTAMP": new_col.explicit_ts_default() # Nothing to worry if a vulnerable column type doesn't use current time # as default if str(new_col.column_type) == "TIMESTAMP": # Cases for TIMESTAMP type if ( str(new_col.default).upper() != "CURRENT_TIMESTAMP" and str(new_col.on_update_current_timestamp).upper() != "CURRENT_TIMESTAMP" ): continue else: # Cases for DATE and DATETIME type if str(new_col.default).upper() != "CURRENT_TIMESTAMP": continue # Adding timestamp column with defaults is considered unsafe # out of replication bootstraping if not old_col: return True # At this point we know this column in new schema need default value setting # to curernt ts. We will need to further confirm if old schema does the same # or not. If not, this will be consider as dangerous for replication if ( str(new_col.default).upper() == "CURRENT_TIMESTAMP" and str(old_col.default).upper() != "CURRENT_TIMESTAMP" ): return True if ( str(new_col.on_update_current_timestamp).upper() == "CURRENT_TIMESTAMP" and str(old_col.on_update_current_timestamp).upper() != "CURRENT_TIMESTAMP" ): return True return False