数据库前进回退方案

一：基础要求

1. DDL 分两部分，第一部分为开发过程中新增的字段和表（不允许有删除和修改操作）；第二部分为需要删除的字段和表（不允许有新增和修改操作）；如果有涉及修改的DDL，需要分开为新增和删除；
2. DML 不允许DROP 和 TRUNCATE; 如有需要，请使用DELETE；

二：整体流程

1. 预先准备：centos 7机器， my2sql，2台mysql数据库均需要root权限， mysqlbench；
   暂时无法在飞书文档外展示此内容
2. 数据库执行DDL第一部分；前进数据库结构；
3. 使用mysqlbench 数据库比对，获取DDL第一部分后退脚本保存；
   1. 在mysql model下，选择database， sync with any source
   2. (可选)没有mysql model 窗口，打开随便一个数据库，选择备份数据库
4. 记录当前时间，前进DML；
5. 在centos 7 机器上，使用my2sql生成回退脚本；命令参考如下; 各参数意义参考https://github.com/liuhr/my2sql
   ./my2sql -user root -password xxxxx -host 192.168.3.25 -port 30306 -mode file -local-binlog-file ./test-begin/binlog.000001 -work-type rollback -start-file ./test-begin/binlog.000001 -start-datetime "2024-12-19 10:34:00" -stop-datetime "2028-07-16 11:00:00" -output-dir ./tmpdir -add-extraInfo true
6. 测试验证
   1. 若有问题：
   2. 执行DML回退脚本回退数据，或者根据第四步时间重新生成DML回退脚本（测试有数据污染情况下），回退数据
   3. 执行DDL第一部分回退脚本
   4. 若无问题：
   5. 第二台mysql执行第一部分DDL前进脚本
   6. 已前进的mysql执行DDL第二部分前进脚本
   7. 使用mysql-workbench生成第二部分DDL回退脚本
   8. 测试验证，如果有问题使用DDL第二部分回退脚本回退至上一步

tip：

验证mysql回退后数据是否完全一致，先导出原始和回退后的mysql数据为csv格式，然后使用下面python脚本验证。
from csv_diff import load_csv, compare
import os

def get_all_files(directory):
    file_names = []
    for root, dirs, files in os.walk(directory):
        for file in files:
            # file_names.append(os.path.join(root, file))
            file_names.append(file)
    return file_names

def comp(names):
    for i in names:
        difference = compare(
            load_csv(open(".\\new\\"+i, 'rt', newline='', encoding='utf-8', errors='ignore')),
            load_csv(open(".\\old\\"+i, 'rt', newline='', encoding='utf-8', errors='ignore'))
        )
        print(difference)

def comp2(names):
    for i in names:

        with open(".\\new\\"+i, 'rt', newline='', encoding='utf-8', errors='ignore') as f1, open(".\\old\\"+i, 'rt', newline='', encoding='utf-8', errors='ignore') as f2:
            for line1, line2 in zip(f1, f2):
                if line1 != line2:
                    print(line1)

def auto_create_path(FilePath):
    if os.path.exists(FilePath):  ##目录存在，返回为真
        print('dir exists')
    else:
        print('dir not exists')
        os.makedirs(FilePath)

auto_create_path(".\\new")
auto_create_path(".\\old2")
directory = '.\\new' # 替换为你的目录路径
file_names = get_all_files(directory)
comp2(file_names)

print(file_names)