数据库迁移模式
面向生产系统，安全且可逆的数据库 Schema 变更。
何时激活
- 创建或修改数据库表
- 增加/删除列或索引
- 执行数据迁移（回填、转换）
- 规划零停机 Schema 变更
- 为新项目搭建迁移工具链
核心原则
- 每次更改都是一次迁移 — 切勿手动修改生产数据库
- 迁移在线上环境仅向前推进 — 回滚应通过新的正向迁移实现
- Schema 迁移与数据迁移分离 — 切勿将 DDL 和 DML 混合在一次迁移中
- 使用生产级数据量测试迁移 — 在 100 行数据上正常的迁移可能在 1000 万行上导致锁表
- 迁移一旦部署即为不可变 — 绝不要编辑已在生产环境运行过的迁移
迁移安全检查清单
应用任何迁移之前：
- 迁移同时包含 UP 和 DOWN（或明确标记为不可逆）
- 大表上无全表锁（使用并发操作）
- 新列有默认值或可为空（切勿无默认值添加 NOT NULL）
- 索引并发创建（对已有表不要内联 CREATE TABLE）
- 数据回填与 Schema 变更是分离的迁移
- 已使用生产数据的副本测试
- 回滚方案已记录
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 不能在事务块内运行
-- 大多数迁移工具需要特殊处理
重命名列（零停机）
绝不在线上直接重命名。使用 Expand-Contract 模式：
-- 步骤 1：添加新列（迁移 001）
ALTER TABLE users ADD COLUMN display_name TEXT;
-- 步骤 2：回填数据（迁移 002，数据迁移）
UPDATE users SET display_name = username WHERE display_name IS NULL;
-- 步骤 3：更新应用代码以读写两个列
-- 部署应用变更
-- 步骤 4：停止写入旧列，删除它（迁移 003）
ALTER TABLE users DROP COLUMN username;
安全删除列
-- 步骤 1：移除所有对列的引用
-- 步骤 2：部署不含该列引用的应用
-- 步骤 3：在下次迁移中删除列
ALTER TABLE orders DROP COLUMN legacy_status;
-- 对于 Django：使用 SeparateDatabaseAndState 从模型中移除，
-- 而不生成 DROP COLUMN（然后在下次迁移中删除）
大数据迁移
-- 坏：在一个事务中更新所有行（锁表）
UPDATE users SET normalized_email = LOWER(email);
-- 好：分批更新并显示进度
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 变更创建迁移
npx prisma migrate dev --name add_user_avatar
# 在线应用待处理的迁移
npx prisma migrate deploy
# 重置数据库（仅开发）
npx prisma migrate reset
# Schema 变更后生成客户端
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 迁移
对于 Prisma 无法表达的操作（并发索引、数据回填）：
# 创建空迁移，然后手动编辑 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 变更生成迁移
npx drizzle-kit generate
# 应用迁移
npx drizzle-kit migrate
# 直接推送 Schema（仅开发，不生成迁移文件）
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
# 创建新的迁移文件
kysely migrate make add_user_avatar
# 应用所有待处理的迁移
kysely migrate latest
# 回滚上一次迁移
kysely migrate down
# 显示迁移状态
kysely migrate list
迁移文件
// migrations/2024_01_15_001_create_user_profile.ts
import { type Kysely, sql } from 'kysely'
// 重要：始终使用 Kysely<any>，而不是你的类型化 DB 接口。
// 迁移是时间冻结的，绝不能依赖于当前的 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()
}
编程式 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 漂移。
// 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)
工作流
# 从模型变更生成迁移
python manage.py makemigrations
# 应用迁移
python manage.py migrate
# 显示迁移状态
python manage.py showmigrations
# 为自定义 SQL 生成空迁移
python manage.py makemigrations --empty app_name -n description
数据迁移
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 模型中删除列，但不立即从数据库中删除：
class Migration(migrations.Migration):
operations = [
migrations.SeparateDatabaseAndState(
state_operations=[
migrations.RemoveField(model_name="user", name="legacy_field"),
],
database_operations=[], # 暂时不触碰数据库
),
]
golang-migrate (Go)
工作流
# 创建迁移对
migrate create -ext sql -dir migrations -seq add_user_avatar
# 应用所有待处理的迁移
migrate -path migrations -database "$DATABASE_URL" up
# 回滚上一次迁移
migrate -path migrations -database "$DATABASE_URL" down 1
# 强制版本（修复脏状态）
migrate -path migrations -database "$DATABASE_URL" force VERSION
迁移文件
-- 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;
零停机迁移策略
对于关键生产变更，遵循 Expand-Contract 模式：
阶段 1：扩展 (EXPAND)
- 添加新列/表（可为空或带默认值）
- 部署：应用同时写入新旧两列
- 回填现有数据
阶段 2：迁移 (MIGRATE)
- 部署：应用从新列读取，同时写入新旧两列
- 验证数据一致性
阶段 3：收缩 (CONTRACT)
- 部署：应用仅使用新列
- 在单独的迁移中删除旧列/表
时间线示例
第 1 天：迁移添加 new_status 列（可为空）
第 1 天：部署应用 v2 — 同时写入 status 和 new_status
第 2 天：运行回填迁移以填充现有行
第 3 天：部署应用 v3 — 仅从 new_status 读取
第 7 天：迁移删除旧的 status 列
反模式
| 反模式 | 为何失败 | 更好的方法 |
|---|---|---|
| 手动在线上执行 SQL | 无审计追踪，不可重复 | 始终使用迁移文件 |
| 编辑已部署的迁移 | 导致环境间漂移 | 创建新迁移替代 |
| 无默认值添加 NOT NULL | 锁表，重写所有行 | 添加可为空列，回填，再添加约束 |
| 在大表上内联创建索引 | 构建期间阻塞写入 | CREATE INDEX CONCURRENTLY |
| Schema 与数据混在同一个迁移中 | 难以回滚，事务时间长 | 分离迁移 |
| 在移除代码前删除列 | 应用因缺失列而报错 | 先移除代码，下次部署再删列 |