案例 004：用 Agent 做数据库迁移前的风险评估

数据库迁移是生产事故的高发区：一条 ALTER TABLE 可能锁表 30 分钟，一个没有索引的查询可能拖垮整个数据库。本文用 Agent 在迁移执行前自动完成 schema diff 分析、索引影响评估、锁风险检测、回滚脚本生成和 dry-run 验证，把"凭经验判断"变成"数据驱动决策"。

一、业务背景

一个运营 3 年的 SaaS 系统，PostgreSQL 数据库 45 张表、最大的表（orders）有 2000 万行。每次数据库迁移都由 DBA 人工审查，平均耗时 2 小时。

痛点：

问题	具体表现	影响
锁表风险	开发者不知道 `ALTER TABLE` 在大表上会锁表	生产故障 2 次
索引遗漏	新查询没加索引，上线后慢查询暴增	页面超时
回滚困难	迁移脚本没有配套回滚脚本	出事只能恢复备份
审查瓶颈	DBA 是单点，所有迁移排队等审查	发布延迟

目标：让 Agent 完成 80% 的审查工作，DBA 只审批高风险操作。

二、风险评估流程

text

┌──────────────┐    ┌──────────────┐    ┌──────────────┐    ┌──────────────┐
│  Migration    │──▶│  Schema Diff │──▶│  风险评估     │──▶│  Dry-Run     │
│  脚本输入     │    │  分析         │    │  (Agent)     │    │  验证        │
└──────────────┘    └──────────────┘    └──────┬───────┘    └──────┬───────┘
                                               │                    │
                                        ┌──────▼───────┐    ┌──────▼───────┐
                                        │  报告生成     │    │  回滚脚本    │
                                        │  (风险报告)   │    │  生成        │
                                        └──────┬───────┘    └──────────────┘
                                               │
                                        ┌──────▼───────┐
                                        │  审批门禁     │
                                        │  (高风险→DBA) │
                                        └──────────────┘

三、Agent 风险评估核心

3.1 评估 Prompt 与输入

yaml

# db-migration-review.yaml
name: "数据库迁移风险评估"
version: "1.0"

inputs:
  migration_file: "migrations/20240615_add_order_tags.sql"
  db_schema: "docs/schema-snapshot.json"     # 当前 schema 快照
  table_stats: "docs/table-stats.json"       # 表大小、行数、索引统计
  query_patterns: "docs/slow-queries.json"   # 已知慢查询

prompt: |
  你是一个高级 DBA。审查以下数据库迁移脚本，评估风险并生成报告。

  ## 迁移脚本
  {{migration_file}}

  ## 当前数据库状态
  - Schema 快照：{{db_schema}}
  - 表统计信息：{{table_stats}}
  - 已知慢查询：{{query_patterns}}

  ## 评估维度
  1. **锁风险**：ALTER TABLE 在大表（>100万行）上是否会锁表？需要 ONLINE 迁移吗？
  2. **索引影响**：新查询是否需要索引？新增索引的创建成本？
  3. **数据完整性**：新增约束（NOT NULL、UNIQUE、FK）是否有存量数据违反？
  4. **回滚可行性**：能生成安全的回滚脚本吗？
  5. **性能影响**：对现有查询的影响？

  ## 风险等级
  - LOW：可自动执行，不需要 DBA 审批
  - MEDIUM：需要 DBA 快速审查（< 15 分钟）
  - HIGH：需要 DBA 完整审查 + dry-run 验证
  - CRITICAL：需要在维护窗口执行，需要提前通知

  ## 输出
  按指定的 JSON 格式输出评估报告。

3.2 风险评估报告示例

json

{
  "migration_file": "20240615_add_order_tags.sql",
  "risk_level": "HIGH",
  "summary": "在 2000 万行的 orders 表上新增 tags 列（JSONB），并创建 GIN 索引",
  
  "assessments": [
    {
      "dimension": "锁风险",
      "level": "HIGH",
      "detail": "ALTER TABLE orders ADD COLUMN tags JSONB 会对 2000 万行的表加 ACCESS EXCLUSIVE 锁，预计锁表时间 5-10 分钟",
      "recommendation": "使用 pg_repack 或 pt-online-schema-change 进行在线迁移，或者在维护窗口执行"
    },
    {
      "dimension": "索引影响",
      "level": "MEDIUM",
      "detail": "CREATE INDEX CONCURRENTLY tags_gin_idx ON orders USING GIN(tags) 不会锁表，但创建过程消耗大量 IO，预计 15-30 分钟",
      "recommendation": "使用 CONCURRENTLY 选项，在低峰期执行，监控 IO 使用率"
    },
    {
      "dimension": "数据完整性",
      "level": "LOW",
      "detail": "JSONB 列默认值设为 '[]'，不会有 NULL 值问题",
      "recommendation": "无需额外处理"
    },
    {
      "dimension": "回滚可行性",
      "level": "LOW",
      "detail": "可以安全回滚：DROP COLUMN 删除列和索引",
      "recommendation": "回滚前确认没有代码已经开始使用 tags 列"
    },
    {
      "dimension": "性能影响",
      "level": "MEDIUM",
      "detail": "新增 JSONB 列会增加每行约 50-100 字节的存储，GIN 索引增加约 2GB 磁盘占用",
      "recommendation": "确认磁盘空间充足（需要额外 3GB），监控 VACUUM 频率"
    }
  ],

  "rollback_script": "-- 回滚脚本\nDROP INDEX IF EXISTS tags_gin_idx;\nALTER TABLE orders DROP COLUMN IF EXISTS tags;",
  
  "execution_plan": {
    "phase_1": "在维护窗口执行 ALTER TABLE ADD COLUMN（锁表 5-10 分钟）",
    "phase_2": "执行 CREATE INDEX CONCURRENTLY（不锁表，15-30 分钟）",
    "phase_3": "验证索引创建完成，运行验证查询",
    "estimated_time": "40 分钟",
    "suggested_window": "周日 03:00-04:00（流量最低时段）"
  },

  "dry_run_commands": [
    "EXPLAIN ANALYZE SELECT * FROM orders WHERE tags @> '[\"vip\"]' LIMIT 10;",
    "SELECT pg_size_pretty(pg_total_relation_size('orders'));",
    "SELECT count(*) FROM orders WHERE tags IS NULL;"
  ],

  "approval_required": true,
  "approvers": ["@dba-lead", "@tech-lead"]
}

3.3 Schema Diff 与锁检测脚本

python

# scripts/db-migration-analyzer.py
"""
分析 SQL 迁移脚本的风险等级。
核心规则基于 PostgreSQL 的锁行为和表大小。
"""
import re
import json

# PostgreSQL 锁风险规则
LOCK_RISK_RULES = {
    "ALTER TABLE .* ADD COLUMN": {
        "lock_level": "ACCESS EXCLUSIVE",
        "risk_on_large_table": "HIGH",
        "safe_alternative": "使用 DEFAULT 值可以避免重写整表（PG 11+）"
    },
    "ALTER TABLE .* DROP COLUMN": {
        "lock_level": "ACCESS EXCLUSIVE",
        "risk_on_large_table": "HIGH",
        "safe_alternative": "先标记废弃，下个版本再物理删除"
    },
    "ALTER TABLE .* ALTER COLUMN .* TYPE": {
        "lock_level": "ACCESS EXCLUSIVE",
        "risk_on_large_table": "CRITICAL",
        "safe_alternative": "需要重写整表，必须用维护窗口或 pg_repack"
    },
    "ALTER TABLE .* ADD CONSTRAINT": {
        "lock_level": "SHARE UPDATE EXCLUSIVE",
        "risk_on_large_table": "MEDIUM",
        "safe_alternative": "使用 NOT VALID 先加约束，再 VALIDATE CONSTRAINT"
    },
    "CREATE INDEX": {
        "lock_level": "SHARE",
        "risk_on_large_table": "MEDIUM",
        "safe_alternative": "使用 CONCURRENTLY 避免锁表"
    },
    "CREATE INDEX CONCURRENTLY": {
        "lock_level": "无锁",
        "risk_on_large_table": "LOW",
        "safe_alternative": "已经是安全方式"
    },
}

LARGE_TABLE_THRESHOLD = 1_000_000  # 超过 100 万行视为大表

def analyze_migration(sql_content: str, table_stats: dict) -> dict:
    """分析迁移脚本的风险"""
    risks = []
    
    for statement in split_statements(sql_content):
        statement = statement.strip()
        if not statement or statement.startswith("--"):
            continue
        
        # 提取涉及的表
        tables = extract_tables(statement)
        
        for rule_pattern, rule in LOCK_RISK_RULES.items():
            if re.match(rule_pattern, statement, re.IGNORECASE):
                for table in tables:
                    row_count = table_stats.get(table, {}).get("rows", 0)
                    is_large = row_count > LARGE_TABLE_THRESHOLD
                    
                    risk = {
                        "statement": statement[:100] + "..." if len(statement) > 100 else statement,
                        "table": table,
                        "lock_level": rule["lock_level"],
                        "row_count": row_count,
                        "safe_alternative": rule["safe_alternative"],
                    }
                    
                    if is_large:
                        risk["level"] = rule["risk_on_large_table"]
                        risk["warning"] = f"表 {table} 有 {row_count:,} 行，属于大表"
                    else:
                        risk["level"] = "LOW"
                    
                    risks.append(risk)
                break
    
    # 计算整体风险等级
    level_order = {"LOW": 0, "MEDIUM": 1, "HIGH": 2, "CRITICAL": 3}
    max_level = max((level_order.get(r["level"], 0) for r in risks), default=0)
    overall = {v: k for k, v in level_order.items()}[max_level]
    
    return {
        "overall_risk": overall,
        "risks": risks,
        "requires_approval": overall in ["HIGH", "CRITICAL"],
        "requires_maintenance_window": overall == "CRITICAL",
    }

def split_statements(sql: str) -> list:
    """按分号分割 SQL 语句，忽略注释和字符串内的分号"""
    statements = []
    current = []
    in_string = False
    
    for char in sql:
        if char == "'" and not in_string:
            in_string = True
        elif char == "'" and in_string:
            in_string = False
        elif char == ";" and not in_string:
            stmt = "".join(current).strip()
            if stmt:
                statements.append(stmt)
            current = []
        else:
            current.append(char)
    
    return statements

def extract_tables(statement: str) -> list:
    """从 SQL 语句中提取表名"""
    tables = []
    patterns = [
        r"ALTER\s+TABLE\s+(\w+)",
        r"CREATE\s+INDEX.*?ON\s+(\w+)",
        r"INSERT\s+INTO\s+(\w+)",
        r"UPDATE\s+(\w+)",
        r"DELETE\s+FROM\s+(\w+)",
    ]
    for pattern in patterns:
        matches = re.findall(pattern, statement, re.IGNORECASE)
        tables.extend(matches)
    return list(set(tables))

# 示例使用
if __name__ == "__main__":
    sql = """
    ALTER TABLE orders ADD COLUMN tags JSONB DEFAULT '[]';
    CREATE INDEX CONCURRENTLY tags_gin_idx ON orders USING GIN(tags);
    """
    stats = {
        "orders": {"rows": 20_000_000, "size_mb": 4500},
        "users": {"rows": 500_000, "size_mb": 120},
    }
    result = analyze_migration(sql, stats)
    print(json.dumps(result, indent=2, ensure_ascii=False))

四、回滚脚本生成

yaml

# rollback-generation.yaml
name: "回滚脚本生成"
prompt: |
  根据以下迁移脚本和风险评估结果，生成安全的回滚脚本。

  ## 迁移脚本
  {{migration_sql}}

  ## 回滚规则
  1. ADD COLUMN → DROP COLUMN
  2. CREATE INDEX → DROP INDEX
  3. DROP COLUMN → 回滚脚本中重新 ADD COLUMN（需要原始类型和默认值）
  4. 数据迁移（UPDATE/INSERT）→ 需要提前备份受影响的数据
  5. 重命名列/表 → 重命名回去

  ## 约束
  - 回滚脚本必须是幂等的（多次执行不报错）
  - 使用 IF EXISTS / IF NOT EXISTS 防止报错
  - 回滚前检查是否有代码依赖被修改的对象

  ## 输出
  ```sql
  -- 回滚脚本：{{migration_file}}
  -- 生成时间：{{timestamp}}
  -- 回滚前检查：{{pre_checks}}

  {{rollback_sql}}

  -- 回滚后验证
  {{verification_sql}}

text


## 五、Dry-Run 验证

```bash
# scripts/dry-run-migration.sh
#!/bin/bash
# 在测试数据库上执行迁移，验证不报错且性能可接受

set -e

MIGRATION_FILE=$1
TEST_DB="migration_test_$(date +%s)"

echo "=== Dry-Run: $MIGRATION_FILE ==="

# 1. 创建测试数据库（从生产 schema 复制）
echo "1. 创建测试数据库..."
pg_dump --schema-only production | psql -d $TEST_DB

# 2. 导入测试数据（采样子集）
echo "2. 导入测试数据..."
pg_dump --data-only --table=orders --where="id % 100 = 0" production | psql -d $TEST_DB

# 3. 记录迁移前的状态
echo "3. 记录迁移前状态..."
psql -d $TEST_DB -c "
  SELECT tablename, pg_size_pretty(pg_total_relation_size(tablename::regclass))
  FROM pg_tables WHERE schemaname = 'public'
  ORDER BY pg_total_relation_size(tablename::regclass) DESC LIMIT 5;
" > /tmp/before-migration.txt

# 4. 执行迁移，记录耗时
echo "4. 执行迁移..."
START_TIME=$(date +%s)
psql -d $TEST_DB -f $MIGRATION_FILE
END_TIME=$(date +%s)
echo "   迁移耗时: $((END_TIME - START_TIME)) 秒"

# 5. 验证索引创建
echo "5. 验证索引..."
psql -d $TEST_DB -c "
  SELECT indexname, indexdef
  FROM pg_indexes
  WHERE tablename IN (SELECT tablename FROM pg_tables WHERE schemaname = 'public')
  ORDER BY tablename;
" > /tmp/after-indexes.txt

# 6. 运行已知慢查询，验证性能
echo "6. 性能验证..."
for query_file in tests/queries/*.sql; do
    echo "  执行: $query_file"
    psql -d $TEST_DB -c "EXPLAIN ANALYZE $(cat $query_file)" 2>&1 | \
      grep "Execution Time" || echo "  ⚠️ 查询失败"
done

# 7. 执行回滚脚本，验证可回滚
echo "7. 验证回滚..."
ROLLBACK_FILE="${MIGRATION_FILE%.sql}.rollback.sql"
if [ -f "$ROLLBACK_FILE" ]; then
    psql -d $TEST_DB -f "$ROLLBACK_FILE"
    echo "   ✅ 回滚成功"
else
    echo "   ⚠️ 回滚脚本不存在"
fi

# 8. 清理测试数据库
echo "8. 清理..."
dropdb $TEST_DB
echo "=== Dry-Run 完成 ==="

六、真实经验与踩坑

6.1 DEFAULT 值在 PG 11 前后行为不同

场景：迁移脚本 ALTER TABLE orders ADD COLUMN status VARCHAR DEFAULT 'active'。Agent 评估为 LOW 风险。问题：在 PostgreSQL 10 及以下，给大表加带 DEFAULT 的列会重写整表（2000 万行需要 30 分钟）。PG 11+ 才支持元数据级别的默认值（瞬间完成）。 解决方案：Agent 的风险评估必须检查 PostgreSQL 版本号。如果是 PG < 11，所有带 DEFAULT 的 ALTER TABLE 都标记为 HIGH 风险，建议使用两阶段迁移：先 ADD COLUMN（无默认值），再 UPDATE 分批填充，最后 SET DEFAULT。

6.2 CREATE INDEX CONCURRENTLY 不能在事务中

场景：迁移框架（如 Alembic / Flyway）把所有 SQL 放在一个事务中执行。问题：CREATE INDEX CONCURRENTLY 不能在事务块中运行，直接报错。 解决方案：迁移脚本中把 CREATE INDEX CONCURRENTLY 单独拆出来，在事务外执行。迁移工具通常支持 --autocommit 或 run_in_transaction: false 配置。在评估报告中明确标注这个约束。

6.3 回滚脚本要在迁移前测试，不是迁移后

场景：迁移成功执行了，然后"等一下再测回滚脚本"。结果回滚脚本有语法错误，真需要回滚时手忙脚乱。问题：回滚脚本的价值在于紧急情况下快速恢复，如果没测过就等于没有。 解决方案：dry-run 流程中必须包含回滚测试——先在测试数据库执行迁移，然后立即执行回滚脚本，验证能完全恢复。回滚脚本和迁移脚本一起提交、一起审查。

七、参数说明表

参数	类型	默认值	说明
`migration_file`	string	必填	迁移 SQL 文件路径
`table_stats`	string	必填	表统计信息（行数、大小）
`pg_version`	string	`"14"`	PostgreSQL 版本号
`large_table_threshold`	int	`1000000`	大表行数阈值
`lock_timeout`	string	`"30s"`	锁等待超时
`require_rollback`	bool	`true`	是否必须提供回滚脚本
`dry_run`	bool	`true`	是否执行 dry-run 验证
`auto_approve_low`	bool	`true`	LOW 风险是否自动审批
`maintenance_window`	string	无	建议的维护窗口时间
`approvers`	list	`["@dba"]`	高风险操作的审批人

八、落地检查清单

迁移脚本的每条语句都有风险评估结果
大表（>100 万行）上的 ALTER TABLE 被正确标记为高风险
回滚脚本已生成且经过 dry-run 验证
CREATE INDEX 使用 CONCURRENTLY 选项（不在事务中）
PG 版本号已确认，DEFAULT 值行为已考虑
dry-run 在测试数据库上执行通过
磁盘空间充足（新增列和索引的存储开销）
维护窗口已预约（如果是 HIGH/CRITICAL 风险）
DBA 已审批高风险迁移
回滚后验证查询已准备

九、系列导航

上一篇：案例 003：用 Agent 生成和维护 OpenAPI 文档下一篇：GitHub 集成实战：Issue、PR、Checks 与 Agent 任务流