database-migrations

数据库迁移设计模式

为生产环境提供安全、可逆的数据库 Schema 变更指南。

何时触发此 Skill

• 创建或修改数据库表结构
• 新增/删除字段或索引
• 执行数据迁移（数据回填 backfill、清洗转换 transform）
• 规划零停机（Zero-Downtime）Schema 变更
• 为新项目搭建数据库迁移工具链（Migration Tooling）

核心原则

1. 一切变更皆 Migration — 绝对禁止在生产数据库上手动敲 SQL 修改结构。
2. 生产环境 Migration 只能单向向前（Forward-only） — 遇到回滚需求，统一提交新的 Forward Migration 来修正。
3. 结构变更（DDL）与数据变更（DML）必须隔离 — 绝不要把 DDL 和 DML 混在同一个 Migration 里。
4. 针对生产级数据量进行测试 — 在 100 条数据上毫秒级通过的 Migration，在 1000 万条数据上可能会把表锁死。
5. 已部署的 Migration 具有不可变性（Immutable） — 已经在生产环境运行过的 Migration 文件，严禁二次修改。

迁移安全检查清单

在应用任何 Migration 前，请确认：

• [ ] Migration 包含 UP 和 DOWN 操作（或显式标记为不可逆）
• [ ] 大表变更绝不触发全表锁（使用并发/异步操作）
• [ ] 新增字段必须设默认值或允许为 NULL（禁止在已有表上直接加无默认值的 NOT NULL 字段）
• [ ] 已有表上的索引必须使用并发创建（CONCURRENTLY，而不是在 CREATE TABLE 内直接内联定义）
• [ ] 数据回填（Data Backfill）已剥离为独立的 Migration
• [ ] 已在生产环境数据副本上完成实测
• [ ] 已制定并记录明确的回滚方案

PostgreSQL 最佳实践

安全地添加字段

-- 推荐：允许为空的字段，无全表锁
ALTER TABLE users ADD COLUMN avatar_url TEXT;

-- 推荐：带默认值的字段（Postgres 11+ 支持瞬间完成，不需要全表重写）
ALTER TABLE users ADD COLUMN is_active BOOLEAN NOT NULL DEFAULT true;

-- 极差：在已有大表上直接添加无默认值的 NOT NULL 字段（会导致全表重写并锁表）
ALTER TABLE users ADD COLUMN role TEXT NOT NULL;
-- 这会锁定全表并重写每一行数据

无停机时间添加索引

-- 极差：在大表上会阻塞写操作
CREATE INDEX idx_users_email ON users (email);

-- 推荐：非阻塞，允许并发写入
CREATE INDEX CONCURRENTLY idx_users_email ON users (email);

-- 注意：CONCURRENTLY 无法在事务块（Transaction block）内部运行
-- 大部分 Migration 工具需要对其进行特殊配置或特殊处理

重命名字段（零停机）

切勿在生产环境直接对字段重命名。请采用“扩张-收缩模式”（Expand-Contract Pattern）：

-- 第 1 步：添加新字段（Migration 001）
ALTER TABLE users ADD COLUMN display_name TEXT;

-- 第 2 步：回填数据（Migration 002，数据迁移）
UPDATE users SET display_name = username WHERE display_name IS NULL;

-- 第 3 步：更新应用代码，让应用同时读写新旧两个字段
-- 部署应用程序变动

-- 第 4 步：停止写入旧字段，在数据库中删掉旧字段（Migration 003）
ALTER TABLE users DROP COLUMN username;

安全地删除字段

-- 第 1 步：从应用代码中彻底移除对该字段的所有引用
-- 第 2 步：部署不再引用该字段的应用版本
-- 第 3 步：在随后的 Migration 中删除该字段
ALTER TABLE orders DROP COLUMN legacy_status;

-- Django 提示：可使用 SeparateDatabaseAndState 从 Model 中移除字段，
-- 而不立即触发 DROP COLUMN，等下一次部署再真正删除

海量数据迁移

-- 极差：在单个事务中更新所有行（会导致长时间锁表）
UPDATE users SET normalized_email = LOWER(email);

-- 推荐：分批更新（Batching）并输出进度
DO $$
DECLARE
  batch_size INT := 10000;
  rows_updated INT;
BEGIN
  LOOP
    UPDATE users
    SET normalized_email = LOWER(email)
    WHERE id IN (
      SELECT id FROM users
      WHERE normalized_email IS NULL
      LIMIT batch_size
      FOR UPDATE SKIP LOCKED
    );
    GET DIAGNOSTICS rows_updated = ROW_COUNT;
    RAISE NOTICE 'Updated % rows', rows_updated;
    EXIT WHEN rows_updated = 0;
    COMMIT;
  END LOOP;
END $$;

Prisma (TypeScript/Node.js)

工作流

# 根据 Schema 变更生成 Migration
npx prisma migrate dev --name add_user_avatar

# 在生产环境中应用未执行的 Migration
npx prisma migrate deploy

# 重置数据库（仅限开发环境）
npx prisma migrate reset

# Schema 变更后生成 Client
npx prisma generate

Schema 示例

model User {
  id        String   @id @default(cuid())
  email     String   @unique
  name      String?
  avatarUrl String?  @map("avatar_url")
  createdAt DateTime @default(now()) @map("created_at")
  updatedAt DateTime @updatedAt @map("updated_at")
  orders    Order[]

  @@map("users")
  @@index([email])
}

自定义 SQL Migration

适用于 Prisma 无法原生表达的底层操作（如并发索引创建 CONCURRENTLY、复杂数据回填）：

# 生成空的 Migration 文件，随后手动编辑 SQL
npx prisma migrate dev --create-only --name add_email_index

-- migrations/20240115_add_email_index/migration.sql
-- Prisma 无法自动生成 CONCURRENTLY 语法，需手动补充
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_users_email ON users (email);

Drizzle (TypeScript/Node.js)

工作流

# 根据 Schema 变动生成 Migration
npx drizzle-kit generate

# 执行 Migration 变更
npx drizzle-kit migrate

# 直接同步 Schema 到数据库（仅限开发环境，不生成 Migration 文件）
npx drizzle-kit push

Schema 示例

import { pgTable, text, timestamp, uuid, boolean } from "drizzle-orm/pg-core";

export const users = pgTable("users", {
  id: uuid("id").primaryKey().defaultRandom(),
  email: text("email").notNull().unique(),
  name: text("name"),
  isActive: boolean("is_active").notNull().default(true),
  createdAt: timestamp("created_at").notNull().defaultNow(),
  updatedAt: timestamp("updated_at").notNull().defaultNow(),
});

Kysely (TypeScript/Node.js)

工作流 (kysely-ctl)

# 初始化配置文件 (kysely.config.ts)
kysely init

# 创建新的 Migration 文件
kysely migrate make add_user_avatar

# 执行所有未落地的 Migration
kysely migrate latest

# 回滚最后一个 Migration
kysely migrate down

# 查看当前 Migration 状态
kysely migrate list

Migration 脚本结构

// migrations/2024_01_15_001_create_user_profile.ts
import { type Kysely, sql } from 'kysely'

// 重要：始终使用 Kysely<any>，不要使用具体类型化的 DB 接口。
// Migration 是历史快照，绝不能依赖当前最新的 Schema 类型。
export async function up(db: Kysely<any>): Promise<void> {
  await db.schema
    .createTable('user_profile')
    .addColumn('id', 'serial', (col) => col.primaryKey())
    .addColumn('email', 'varchar(255)', (col) => col.notNull().unique())
    .addColumn('avatar_url', 'text')
    .addColumn('created_at', 'timestamp', (col) =>
      col.defaultTo(sql`now()`).notNull()
    )
    .execute()

  await db.schema
    .createIndex('idx_user_profile_avatar')
    .on('user_profile')
    .column('avatar_url')
    .execute()
}

export async function down(db: Kysely<any>): Promise<void> {
  await db.schema.dropTable('user_profile').execute()
}

代码级迁移执行器（Programmatic Migrator）

import { Migrator, FileMigrationProvider } from 'kysely'
import { promises as fs } from 'fs'
import * as path from 'path'
// 仅限 ESM 模块 — CJS 环境可直接使用 __dirname
import { fileURLToPath } from 'url'
const migrationFolder = path.join(
  path.dirname(fileURLToPath(import.meta.url)),
  './migrations',
)

// `db` 为实例化的 Kysely<any> 数据库对象
const migrator = new Migrator({
  db,
  provider: new FileMigrationProvider({
    fs,
    path,
    migrationFolder,
  }),
  // 警告：该选项仅允许在开发环境开启。开启后会禁用基于时间戳的顺序校验，
  // 容易导致不同环境之间的 Schema 出现漂移（Schema Drift）。
  // allowUnorderedMigrations: true,
})

const { error, results } = await migrator.migrateToLatest()

results?.forEach((it) => {
  if (it.status === 'Success') {
    console.log(`migration "${it.migrationName}" executed successfully`)
  } else if (it.status === 'Error') {
    console.error(`failed to execute migration "${it.migrationName}"`)
  }
})

if (error) {
  console.error('migration failed', error)
  process.exit(1)
}

Django (Python)

工作流

# 根据 Model 变动生成 Migration 文件
python manage.py makemigrations

# 执行数据库迁移
python manage.py migrate

# 查看迁移状态列表
python manage.py showmigrations

# 生成空 Migration 文件以编写自定义原生 SQL
python manage.py makemigrations --empty app_name -n description

数据迁移（Data Migration）

from django.db import migrations

def backfill_display_names(apps, schema_editor):
    User = apps.get_model("accounts", "User")
    batch_size = 5000
    users = User.objects.filter(display_name="")
    while users.exists():
        batch = list(users[:batch_size])
        for user in batch:
            user.display_name = user.username
        User.objects.bulk_update(batch, ["display_name"], batch_size=batch_size)

def reverse_backfill(apps, schema_editor):
    pass  # 数据迁移，无需回滚逻辑

class Migration(migrations.Migration):
    dependencies = [("accounts", "0015_add_display_name")]

    operations = [
        migrations.RunPython(backfill_display_names, reverse_backfill),
    ]

SeparateDatabaseAndState

从 Django Model 中移除字段，但暂不在物理数据库中删除该列：

class Migration(migrations.Migration):
    operations = [
        migrations.SeparateDatabaseAndState(
            state_operations=[
                migrations.RemoveField(model_name="user", name="legacy_field"),
            ],
            database_operations=[],  # 暂不触动真实数据库
        ),
    ]

golang-migrate (Go)

工作流

# 创建一对 UP/DOWN 迁移脚本
migrate create -ext sql -dir migrations -seq add_user_avatar

# 执行所有未落地 Migration
migrate -path migrations -database "$DATABASE_URL" up

# 回滚最近一次 Migration
migrate -path migrations -database "$DATABASE_URL" down 1

# 强行设置版本号（用于修复 Dirty 状态）
migrate -path migrations -database "$DATABASE_URL" force VERSION

Migration 脚本示例

-- migrations/000003_add_user_avatar.up.sql
ALTER TABLE users ADD COLUMN avatar_url TEXT;
CREATE INDEX CONCURRENTLY idx_users_avatar ON users (avatar_url) WHERE avatar_url IS NOT NULL;

-- migrations/000003_add_user_avatar.down.sql
DROP INDEX IF EXISTS idx_users_avatar;
ALTER TABLE users DROP COLUMN IF EXISTS avatar_url;

零停机迁移策略（Zero-Downtime Migration Strategy）

针对核心生产环境的重磅变更，建议严格遵循“扩张-收缩模式”（Expand-Contract Pattern）：

阶段 1：扩张（EXPAND）
  - 新增字段/表（可空或带有默认值）
  - 应用上线发布：应用同时向新旧两个字段/表双写（Dual-write）
  - 对已有存量数据进行回填（Backfill）

阶段 2：过渡（MIGRATE）
  - 应用上线发布：应用改为仅读取新字段/表，但依然保持双写
  - 校验数据一致性

阶段 3：收缩（CONTRACT）
  - 应用上线发布：应用彻底只使用新字段/表，移除写入旧字段的代码
  - 在后续独立的 Migration 中真正 Drop 掉旧字段/表

部署时间轴示例

第 1 天：提交 Migration 添加 new_status 字段（允许为空）
第 1 天：发布 App v2 —— 同时对 status 与 new_status 进行双写
第 2 天：运行回填 Migration 补齐存量行数据
第 3 天：发布 App v3 —— 业务全面切到仅读取 new_status
第 7 天：提交 Migration 彻底删除旧 status 字段

典型反模式（Anti-Patterns）

反模式（Anti-Pattern）	危害/失败原因	正确应对方案
在生产环境手动敲 SQL	缺乏审计追踪，无法自动化重现	强制统一使用标准 Migration 文件
修改已发布的 Migration	导致不同环境间数据库 Schema 漂移（Drift）	永远通过新建 Migration 来修正问题
添加无默认值的 NOT NULL 字段	触发全表长时间锁表，触发全表重写	先加允许为空字段，回填数据后再加约束
在大表上内联同步创建索引	创建索引期间会全程阻塞写操作	统一使用 CREATE INDEX CONCURRENTLY
混写 Schema 与数据变更	回滚困难，极易引发大事务锁死	严格拆分为独立的 Migration 操作
在移除引用代码前提前删字段	导致在线应用程序引发列找不到报错	必须遵循“先发布移除引用的代码，下一次部署再删字段”