資料庫遷移

資料庫遷移模式

安全、可逆的資料庫架構變更，專為生產系統設計。

何時啟用

• 建立或修改資料表
• 新增/移除欄位或索引
• 執行資料遷移（回填、轉換）
• 規劃零停機架構變更
• 為新專案設定遷移工具

核心原則

1. 每個變更都是一次遷移 — 絕不手動修改正式環境資料庫
2. 遷移在正式環境中僅向前進行 — 回滾使用新的向前遷移
3. 架構遷移與資料遷移分開 — 絕不在單一遷移中混合 DDL 和 DML
4. 針對正式環境規模的資料測試遷移 — 在 100 行上運作正常的遷移，可能在 1000 萬行上造成鎖定
5. 遷移一旦部署即不可修改 — 絕不編輯已在正式環境中執行過的遷移

遷移安全檢查清單

套用任何遷移之前：

• 遷移同時包含 UP 與 DOWN（或明確標記為不可逆）
• 大型資料表沒有全表鎖定（使用並行操作）
• 新欄位具有預設值或可為空（絕不新增 NOT NULL 卻不帶預設值）
• 索引已並行建立（不是與 CREATE TABLE 內嵌在現有資料表中）
• 資料回填是獨立於架構變更的遷移
• 已針對正式環境資料的副本進行測試
• 回滾計畫已記錄

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 無法在交易區塊內執行
-- 大多數遷移工具需要特殊處理

重新命名欄位（零停機）

絕不在正式環境中直接重新命名。使用擴展-收縮模式：

-- 步驟 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)

工作流程

# 從架構變更建立遷移
npx prisma migrate dev --name add_user_avatar

# 在正式環境中套用待處理的遷移
npx prisma migrate deploy

# 重設資料庫（僅限開發環境）
npx prisma migrate reset

# 在架構變更後產生客戶端
npx prisma generate

架構範例

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)

工作流程

# 從架構變更產生遷移
npx drizzle-kit generate

# 套用遷移
npx drizzle-kit migrate

# 直接推送架構（僅限開發環境，不產生遷移檔）
npx drizzle-kit push

架構範例

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 介面。
// 遷移是凍結在時間點上的，不得依賴於當前的架構型別。
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()
}

程式化遷移器

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,
  }),
  // 警告：僅在開發環境中啟用。停用時間戳排序
  // 驗證，可能會導致環境之間的架構偏移。
  // 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;

零停機遷移策略

對於關鍵的正式環境變更，請遵循擴展-收縮模式：

階段 1：擴展
  - 新增欄位/資料表（可為空或帶有預設值）
  - 部署：應用程式同時寫入「舊」與「新」欄位
  - 回填現有資料

階段 2：遷移
  - 部署：應用程式從「新」欄位讀取，同時寫入「舊」與「新」
  - 驗證資料一致性

階段 3：收縮
  - 部署：應用程式僅使用「新」欄位
  - 在單獨的遷移中刪除舊的欄位/資料表

時間表範例

第 1 天：遷移新增 new_status 欄位（可為空）
第 1 天：部署應用程式 v2 — 同時寫入 status 和 new_status
第 2 天：執行現有資料列的回填遷移
第 3 天：部署應用程式 v3 — 僅從 new_status 讀取
第 7 天：遷移刪除舊的 status 欄位

反模式

反模式	失敗原因	更好的作法
在正式環境中手動執行 SQL	無審計軌跡，無法重複	始終使用遷移檔
編輯已部署的遷移	導致環境之間偏移	改為建立新的遷移
NOT NULL 不帶預設值	鎖定資料表，重寫所有行	先新增可為空，回填，再新增約束
在大型資料表上內嵌索引	在建置期間封鎖寫入	CREATE INDEX CONCURRENTLY
在同一遷移中混合架構與資料	難以回滾，交易時間過長	分開成不同的遷移
在移除程式碼之前刪除欄位	應用程式因缺少欄位而出錯	先移除程式碼，下次部署再刪除欄位