database-migrations 插件实战：PostgreSQL/MySQL 零停机 SQL 迁移的完整实施指南-编程实验室

database-migrations 插件实战：PostgreSQL/MySQL 零停机 SQL 迁移的完整实施指南

【免费下载链接】agentsMulti-harness agentic plugin marketplace for Claude Code, Codex, Cursor, OpenCode, GitHub Copilot, and Google Antigravity项目地址: https://gitcode.com/GitHub_Trending/agents24/agents

在生产环境做数据库结构变更，最大的风险从来不是 SQL 写错，而是迁移期间服务不可用、数据不一致、以及无法回滚。本指南以 agents24 仓库中database-migrations插件的 sql-migrations 命令为骨架，系统讲解面向 PostgreSQL、MySQL、SQL Server 的零停机迁移策略（Expand-Contract、Blue-Green、Online Schema Change）、Flyway/Alembic 迁移脚本写法、前后置数据完整性校验、自动化回滚流程与大表性能优化方案。读完你可以直接在 Claude Code / Codex / Cursor 等 harness 中调用/database-migrations:sql-migrations命令，让 Agent 按企业级标准为你生成一整套「分析报告 + 迁移脚本 + 校验套件 + 回滚脚本」的交付物。

命令定位：在哪个 harness 下、如何调用

该命令是database-migrations插件（仓库中对数据库迁移自动化的插件封装，见 docs/plugins.md 的 Database 分类）提供的一个 slash command。按仓库的插件结构约定（docs/usage.md），插件的命令采用命名空间调用方式：

/plugin install database-migrations # 先安装插件 /database-migrations:sql-migrations "为 users 表增加 email_verified 字段并做零停机迁移" /database-migrations:migration-observability "对迁移过程接入 Prometheus 监控"

命令格式为/plugin-name:command-name [arguments]，$ARGUMENTS占位符接收你传入的自然语言需求，描述要交付的内容（见 sql-migrations.md 的Requirements段）；
命令声明了tool_access: [Read, Write, Edit, Bash, Grep, Glob]，即 Agent 可以读写迁移脚本文件、在数据库上执行 SQL、搜索工程结构；
安装插件后只会把该插件自身的 agents、commands、skills 加载进上下文，不会加载整个市场（见 docs/usage.md 的安装说明）。

命令约定最终输出 7 项交付物：迁移分析报告、零停机实施计划、版本化迁移脚本、前后置校验套件、自动化与手动回滚脚本、批处理/并行性能优化、监控集成。下文逐一展开其技术内核。

零停机迁移的三种核心策略

Expand-Contract（扩展-收缩）模式

Expand-Contract 是关系型数据库做线上 Schema 变更最经典的「兼容两代」方案，核心思想是任何一步都不破坏当前正在运行的旧代码，三个阶段完整代码如下（摘自 sql-migrations.md）：

-- Phase 1: EXPAND（向后兼容的扩展） ALTER TABLE users ADD COLUMN email_verified BOOLEAN DEFAULT FALSE; CREATE INDEX CONCURRENTLY idx_users_email_verified ON users(email_verified); -- Phase 2: MIGRATE DATA（分批回填） DO $$ DECLARE batch_size INT := 10000; rows_updated INT; BEGIN LOOP UPDATE users SET email_verified = (email_confirmation_token IS NOT NULL) WHERE id IN ( SELECT id FROM users WHERE email_verified IS NULL LIMIT batch_size ); GET DIAGNOSTICS rows_updated = ROW_COUNT; EXIT WHEN rows_updated = 0; COMMIT; PERFORM pg_sleep(0.1); END LOOP; END $$; -- Phase 3: CONTRACT（代码上线后收缩） ALTER TABLE users DROP COLUMN email_confirmation_token;

关键要点：

Phase 1 只做加法：新增列、新增索引都是向后兼容的，旧版本代码读不到新列也无所谓；CREATE INDEX CONCURRENTLY避免长时间持锁阻塞写入；
Phase 2 分批回填：用LIMIT batch_size游标式推进、每次COMMIT释放锁与事务日志压力，pg_sleep(0.1)给数据库喘息空间，避免单事务更新百万行导致 WAL 暴涨与复制延迟；
Phase 3 才做减法：DROP COLUMN必须等新代码全部上线、不再引用旧列之后执行。收缩阶段一旦出错，回滚只需恢复旧列，数据早已在 Phase 2 回填完成。

Blue-Green Schema Migration（蓝绿 Schema 迁移）

当涉及表结构大改（如字段重命名、类型变更、状态机迁移）时，Expand-Contract 不够用，此时采用新表 + 双写 + 回填的蓝绿模式：

-- Step 1: 创建新版本表 v2_orders CREATE TABLE v2_orders ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), customer_id UUID NOT NULL, total_amount DECIMAL(12,2) NOT NULL, status VARCHAR(50) NOT NULL, metadata JSONB DEFAULT '{}', created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP, CONSTRAINT fk_v2_orders_customer FOREIGN KEY (customer_id) REFERENCES customers(id), CONSTRAINT chk_v2_orders_amount CHECK (total_amount >= 0) ); CREATE INDEX idx_v2_orders_customer ON v2_orders(customer_id); CREATE INDEX idx_v2_orders_status ON v2_orders(status); -- Step 2: 双写同步（触发器将 orders 的写入实时同步到 v2_orders） CREATE OR REPLACE FUNCTION sync_orders_to_v2() RETURNS TRIGGER AS $$ BEGIN INSERT INTO v2_orders (id, customer_id, total_amount, status) VALUES (NEW.id, NEW.customer_id, NEW.amount, NEW.state) ON CONFLICT (id) DO UPDATE SET total_amount = EXCLUDED.total_amount, status = EXCLUDED.status; RETURN NEW; END; $$ LANGUAGE plpgsql; CREATE TRIGGER sync_orders_trigger AFTER INSERT OR UPDATE ON orders FOR EACH ROW EXECUTE FUNCTION sync_orders_to_v2(); -- Step 3: 历史数据分批回填（基于主键游标，ON CONFLICT 幂等） DO $$ DECLARE batch_size INT := 10000; last_id UUID := NULL; BEGIN LOOP INSERT INTO v2_orders (id, customer_id, total_amount, status) SELECT id, customer_id, amount, state FROM orders WHERE (last_id IS NULL OR id > last_id) ORDER BY id LIMIT batch_size ON CONFLICT (id) DO NOTHING; SELECT id INTO last_id FROM orders WHERE (last_id IS NULL OR id > last_id) ORDER BY id LIMIT 1 OFFSET (batch_size - 1); EXIT WHEN last_id IS NULL; COMMIT; END LOOP; END $$;

这个模式的工程要点：

双写是蓝绿切换成功的关键：触发器（Trigger）方案简单可靠，但要评估写入放大的代价；高并发场景更推荐改用 CDC（Change Data Capture）工具异步同步，避免触发器拖慢主库写路径——这正是本插件配套的 migration-observability 命令中 Debezium + Kafka 管道的用武之地；
回填必须幂等：ON CONFLICT (id) DO NOTHING保证双写期间重放历史数据不会产生重复行，也允许回填任务失败后从断点续跑；
切换与收敛：回填完成并校验行数一致后，应用层切换到v2_orders，再逐步停用双写触发器并归档旧表。

Online Schema Change：大表安全加 NOT NULL 约束

给一张大表直接ADD COLUMN ... NOT NULL DEFAULT在多数数据库上会引发全表重写或长锁。安全做法是「先可空、再回填、最后用 NOT VALID 约束校验」（PostgreSQL 12+ 支持）：

-- Step 1: 以可空方式加列 ALTER TABLE large_table ADD COLUMN new_field VARCHAR(100); -- Step 2: 分批回填数据 UPDATE large_table SET new_field = 'default_value' WHERE new_field IS NULL; -- Step 3: 加约束但不立即全表扫描（NOT VALID） ALTER TABLE large_table ADD CONSTRAINT chk_new_field_not_null CHECK (new_field IS NOT NULL) NOT VALID; -- Step 4: 后台校验约束（VALIDATE 阶段不加写锁阻塞业务） ALTER TABLE large_table VALIDATE CONSTRAINT chk_new_field_not_null;

NOT VALID+VALIDATE两步走的价值在于：约束添加瞬间只做元数据操作，真正的全表扫描在VALIDATE阶段于后台完成，期间数据库仍可正常读写，这正是零停机 Schema 变更的核心诉求。

版本化迁移脚本：Flyway 与 Alembic

Flyway（SQL 原生）

Flyway 按V<版本号>__<描述>.sql约定管理脚本，事务包裹保证「要么全部成功、要么全部回滚」：

-- V001__add_user_preferences.sql BEGIN; CREATE TABLE IF NOT EXISTS user_preferences ( user_id UUID PRIMARY KEY, theme VARCHAR(20) DEFAULT 'light' NOT NULL, language VARCHAR(10) DEFAULT 'en' NOT NULL, timezone VARCHAR(50) DEFAULT 'UTC' NOT NULL, notifications JSONB DEFAULT '{}' NOT NULL, created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP, CONSTRAINT fk_user_preferences_user FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE ); CREATE INDEX idx_user_preferences_language ON user_preferences(language); -- 为存量用户播种默认偏好 INSERT INTO user_preferences (user_id) SELECT id FROM users ON CONFLICT (user_id) DO NOTHING; COMMIT;

写法要点：约束（外键、CHECK）在表内一并声明保持自文档化；新表创建后立刻补索引；「播种存量数据」用ON CONFLICT DO NOTHING保证重复执行安全——这与 Blue-Green 回填的幂等思想一致。

Alembic（Python/SQLAlchemy 生态）

对于 Python 项目，Alembic 迁移以upgrade()/downgrade()成对出现，天然支持自动回滚：

"""add_user_preferences Revision ID: 001_user_prefs """ from alembic import op import sqlalchemy as sa from sqlalchemy.dialects import postgresql def upgrade(): op.create_table( 'user_preferences', sa.Column('user_id', postgresql.UUID(as_uuid=True), primary_key=True), sa.Column('theme', sa.VARCHAR(20), nullable=False, server_default='light'), sa.Column('language', sa.VARCHAR(10), nullable=False, server_default='en'), sa.Column('timezone', sa.VARCHAR(50), nullable=False, server_default='UTC'), sa.Column('notifications', postgresql.JSONB, nullable=False, server_default=sa.text("'{}'::jsonb")), sa.ForeignKeyConstraint(['user_id'], ['users.id'], ondelete='CASCADE') ) op.create_index('idx_user_preferences_language', 'user_preferences', ['language']) op.execute(""" INSERT INTO user_preferences (user_id) SELECT id FROM users ON CONFLICT (user_id) DO NOTHING """) def downgrade(): op.drop_table('user_preferences')

注意其中的工程细节：server_default让默认值下沉到数据库层（而非应用层），保证任何写入路径行为一致；server_default=sa.text("'{}'::jsonb")是 PostgreSQL JSONB 类型的标准写法；downgrade()与upgrade()严格对称，这是后面回滚机制能自动化的前提。

数据完整性校验：迁移前防御，迁移后验证

迁移最怕「跑完了才发现数据不对」。命令文档给出两层校验函数，分别拦截迁移前与迁移后的数据问题（摘自 sql-migrations.md）：

def validate_pre_migration(db_connection): checks = [] # Check 1: 关键列不允许出现 NULL null_check = db_connection.execute(""" SELECT table_name, COUNT(*) as null_count FROM users WHERE email IS NULL """).fetchall() if null_check[0]['null_count'] > 0: checks.append({ 'check': 'null_values', 'status': 'FAILED', 'severity': 'CRITICAL', 'message': 'NULL values found in required columns' }) # Check 2: 检测重复值 duplicate_check = db_connection.execute(""" SELECT email, COUNT(*) as count FROM users GROUP BY email HAVING COUNT(*) > 1 """).fetchall() if duplicate_check: checks.append({ 'check': 'duplicates', 'status': 'FAILED', 'severity': 'CRITICAL', 'message': f'{len(duplicate_check)} duplicate emails' }) return checks def validate_post_migration(db_connection, migration_spec): validations = [] # 行数核对：迁移前后 affected_tables 的行数必须与预期一致 for table in migration_spec['affected_tables']: actual_count = db_connection.execute( f"SELECT COUNT(*) FROM {table['name']}" ).fetchone()[0] validations.append({ 'check': 'row_count', 'table': table['name'], 'expected': table['expected_count'], 'actual': actual_count, 'status': 'PASS' if actual_count == table['expected_count'] else 'FAIL' }) return validations

设计要点：

校验结果结构化（check/status/severity/message四字段），方便后续在 CI 或监控面板中机器化消费；
前置校验聚焦「数据是否具备迁移条件」（NULL、重复等），命中CRITICAL直接阻断迁移；
后置校验聚焦「迁移结果是否符合预期」（行数、值域），migration_spec['affected_tables']要求迁移计划显式声明受影响表与期望行数，杜绝「不知道动了哪些表」的模糊操作。

回滚机制：事务保护 + 快照恢复 + 脚本化回滚

MigrationRunner：事务、SAVEPOINT 与快照

命令文档给出的MigrationRunner把「预检 → 备份 → 执行 → 后验 → 清理/回滚」串成一条自动链路：

import psycopg2 from contextlib import contextmanager class MigrationRunner: def __init__(self, db_config): self.db_config = db_config self.conn = None @contextmanager def migration_transaction(self): try: self.conn = psycopg2.connect(**self.db_config) self.conn.autocommit = False cursor = self.conn.cursor() cursor.execute("SAVEPOINT migration_start") yield cursor self.conn.commit() except Exception as e: if self.conn: self.conn.rollback() raise finally: if self.conn: self.conn.close() def run_with_validation(self, migration): try: # 前置校验，FAILED 即中止 pre_checks = self.validate_pre_migration(migration) if any(c['status'] == 'FAILED' for c in pre_checks): raise MigrationError("Pre-migration validation failed") # 执行前创建快照 self.create_snapshot() # 事务内执行迁移 + 后置校验 with self.migration_transaction() as cursor: for statement in migration.forward_sql: cursor.execute(statement) post_checks = self.validate_post_migration(migration, cursor) if any(c['status'] == 'FAIL' for c in post_checks): raise MigrationError("Post-migration validation failed") self.cleanup_snapshot() except Exception as e: self.rollback_from_snapshot() raise

这个类的工程价值：

migration_transaction上下文管理器统一管理连接生命周期，SAVEPOINT migration_start让整个迁移成为一个可原子回退的单元；
run_with_validation把「校验通过才执行、执行后校验失败整体回滚」作为硬性契约：后置校验与迁移在同一事务内，发现异常直接rollback，不会留下半成品状态；
快照（create_snapshot/rollback_from_snapshot/cleanup_snapshot）作为事务之外的第二道保险，应对 DDL 无法被普通事务覆盖（MySQL 隐式提交 DDL）或事务已部分提交的场景。这与仓库中 database-admin Agent 强调的「备份策略：全量/增量/差异 + 时间点恢复」以及「未经过恢复演练的备份等于不存在」的运维哲学一脉相承。

rollback_migration.sh：面向运维的手动回滚脚本

对于 Flyway 这类「up 与 down 成对」的迁移体系，回滚脚本的关键是先核对版本，再备份，再执行 down：

#!/bin/bash # rollback_migration.sh set -e MIGRATION_VERSION=$1 DATABASE=$2 # 校验当前版本，防止误回滚 CURRENT_VERSION=$(psql -d $DATABASE -t -c \ "SELECT version FROM schema_migrations ORDER BY applied_at DESC LIMIT 1" | xargs) if [ "$CURRENT_VERSION" != "$MIGRATION_VERSION" ]; then echo "❌ Version mismatch" exit 1 fi # 回滚前先做备份 BACKUP_FILE="pre_rollback_${MIGRATION_VERSION}_$(date +%Y%m%d_%H%M%S).sql" pg_dump -d $DATABASE -f "$BACKUP_FILE" # 执行回滚 if [ -f "migrations/${MIGRATION_VERSION}.down.sql" ]; then psql -d $DATABASE -f "migrations/${MIGRATION_VERSION}.down.sql" psql -d $DATABASE -c "DELETE FROM schema_migrations WHERE version = '$MIGRATION_VERSION';" echo "✅ Rollback complete" else echo "❌ Rollback file not found" exit 1 fi

要点：set -e让任何一步失败即退出；版本不匹配直接拒绝执行（防止在生产上回滚错版本）；pg_dump的备份文件带版本号与时间戳，天然形成回滚审计轨迹。这也呼应了仓库 review-agent-governance / signed-audit-trails 一类的可追溯治理理念，但本命令自身的重点是「回滚前后有据可查」。

性能优化：批处理游标与并行分区迁移

大表迁移的黄金法则是永远不要一次 UPDATE/INSERT 全部数据。命令文档提供两种互补的优化器：

BatchMigrator：基于游标的分批迁移

class BatchMigrator: def __init__(self, db_connection, batch_size=10000): self.db = db_connection self.batch_size = batch_size def migrate_large_table(self, source_query, target_query, cursor_column='id'): last_cursor = None batch_number = 0 while True: batch_number += 1 if last_cursor is None: batch_query = f"{source_query} ORDER BY {cursor_column} LIMIT {self.batch_size}" params = [] else: batch_query = f"{source_query} AND {cursor_column} > %s ORDER BY {cursor_column} LIMIT {self.batch_size}" params = [last_cursor] rows = self.db.execute(batch_query, params).fetchall() if not rows: break for row in rows: self.db.execute(target_query, row) last_cursor = rows[-1][cursor_column] self.db.commit() print(f"Batch {batch_number}: {len(rows)} rows") time.sleep(0.1)

default batch_size=10000是一个兼顾事务大小与吞吐的常用值（可根据行宽、网络延迟调整到 1000~50000）。每批独立COMMIT，即使中途失败也只需从last_cursor断点续跑——这与 Expand-Contract 阶段二、Blue-Green 回填的游标思想完全同构。

ParallelMigrator：按主键区间并行

from concurrent.futures import ThreadPoolExecutor class ParallelMigrator: def __init__(self, db_config, num_workers=4): self.db_config = db_config self.num_workers = num_workers def migrate_partition(self, partition_spec): table_name, start_id, end_id = partition_spec conn = psycopg2.connect(**self.db_config) cursor = conn.cursor() cursor.execute(f""" INSERT INTO v2_{table_name} (columns...) SELECT columns... FROM {table_name} WHERE id >= %s AND id < %s """, [start_id, end_id]) conn.commit() cursor.close() conn.close() def migrate_table_parallel(self, table_name, partition_size=100000): # 取表的主键边界 conn = psycopg2.connect(**self.db_config) cursor = conn.cursor() cursor.execute(f"SELECT MIN(id), MAX(id) FROM {table_name}") min_id, max_id = cursor.fetchone() # 按 partition_size 切分区间 partitions = [] current_id = min_id while current_id <= max_id: partitions.append((table_name, current_id, current_id + partition_size)) current_id += partition_size # 多线程并行执行 with ThreadPoolExecutor(max_workers=self.num_workers) as executor: results = list(executor.map(self.migrate_partition, partitions)) conn.close()

设计要点：按主键值域切分（而非随机分页）保证各分区间互不重叠，天然可并行且可断点续传；每个 worker 使用独立连接，规避共享连接下的并发事务问题；num_workers与数据库连接池上限、IO 能力需配合调优，盲目加大并发数反而会拖垮目标库。这与仓库中 database-optimizer Agent 的能力定位（索引策略、大表迁移、批量处理、性能基线监控）直接对应。

索引管理：大表批量导入前后的索引策略

对亿级大表做批量导入时，索引维护开销往往超过数据写入本身。标准做法是「先摘索引、批量灌数、再 CONCURRENTLY 重建」：

-- 1) 记录当前表的非主键索引定义 CREATE TEMP TABLE migration_indexes AS SELECT indexname, indexdef FROM pg_indexes WHERE tablename = 'large_table' AND indexname NOT LIKE '%pkey%'; -- 2) 摘除索引 DO $$ DECLARE idx_record RECORD; BEGIN FOR idx_record IN SELECT indexname FROM migration_indexes LOOP EXECUTE format('DROP INDEX IF EXISTS %I', idx_record.indexname); END LOOP; END $$; -- 3) 执行批量操作 INSERT INTO large_table SELECT * FROM source_table; -- 4) 用 CONCURRENTLY 重建（不阻塞线上读写） DO $$ DECLARE idx_record RECORD; BEGIN FOR idx_record IN SELECT indexdef FROM migration_indexes LOOP EXECUTE regexp_replace(idx_record.indexdef, 'CREATE INDEX', 'CREATE INDEX CONCURRENTLY'); END LOOP; END $$;

工程要点：临时表保存indexdef便于精确还原索引（含类型、列序、条件）；保留主键索引（NOT LIKE '%pkey%'）避免破坏约束与复制；重建阶段统一替换为CREATE INDEX CONCURRENTLY，让索引构建在后台进行、不占用排他锁。需要权衡的风险是：摘索引期间相关查询会走全表扫描，因此该方案应配合业务低峰窗口使用，并在摘除前用EXPLAIN评估受影响的查询。

输出交付物与监控联动

命令文档明确要求 Agent 最终输出 7 项交付物，可视为迁移实施的验收清单：

Migration Analysis Report：变更的详细拆解（受影响表、行数、外键依赖）；
Zero-Downtime Implementation Plan：在 Expand-Contract 与 Blue-Green 中给出选择依据；
Migration Scripts：接入 Flyway / Alembic 的版本化脚本；
Validation Suite：前后置校验函数；
Rollback Procedures：自动化与手动回滚脚本；
Performance Optimization：批处理、并行执行方案；
Monitoring Integration：进度跟踪与告警。

第 7 项正是本插件配套命令 migration-observability 的主场：它通过 Prometheus 的 Histogram/Counter/Gauge 采集迁移时长、处理行数、消费滞后与复制延迟指标，提供 Grafana 面板（如rate(migration_rows_total[5m])表示迁移进度、migration_data_lag_seconds阈值面板显示数据滞后）、异常检测（吞吐低于预期 50% 或错误率超 1% 触发告警），并支持通过 Debezium + Kafka 搭建 CDC 管道替代触发器做蓝绿双写同步。在 docs/usage.md 的命令参考中，两个命令被并列在 Database 分类下，正好构成「先迁移、后监控」的完整链路。

小结

SQL 迁移的工程化本质可以概括为四句话：变更分阶段（Expand-Contract / Blue-Green）、脚本版本化（Flyway / Alembic）、执行有校验（前后置检查 + 快照回滚）、大表要分批（游标 + 并行 + 索引摘建）。database-migrations插件的sql-migrations命令把这几条经验固化成可复用的 Agent 工作流，让你在 Claude Code、Codex、Cursor、OpenCode、GitHub Copilot 与 Antigravity 等任一 harness 中，都能获得一份包含分析报告、迁移脚本、校验套件、回滚脚本、性能方案与监控集成的生产级迁移方案。实际使用前请务必在预生产环境完整演练一遍回滚流程——正如 database-admin 所强调的：未经演练的备份与回滚，等于不存在。

创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考