database-migrations

資料庫遷移模式（Database Migration Patterns）

適用於正式環境的安全、可逆資料庫 Schema 變更指南。

何時啟用

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

核心原則

1. 所有變更都是一次遷移 — 切勿手動修改正式環境資料庫
2. 正式環境中的遷移只能向前 — 回復操作應透過新的向前遷移（forward migration）實現
3. Schema 與資料遷移必須分離 — 切勿在同一筆遷移中混合 DDL 與 DML
4. 務必使用接近正式環境規模的資料進行測試 — 在 100 筆資料測試正常的遷移，放到 1,000 萬筆資料時可能會鎖定資料表
5. 已部署的遷移不可變更 — 切勿修改已在正式環境執行過的遷移檔案

遷移安全檢查清單

套用任何遷移前：

• [ ] 遷移包含 UP 與 DOWN（或已明確標註為不可逆）
• [ ] 大表中不會觸發全表鎖定（使用並行/非阻塞操作）
• [ ] 新欄位設有預設值或允許為空（切勿新增無預設值的 NOT NULL 欄位）
• [ ] 索引採用並行方式建立（對於既有資料表，切勿在 CREATE TABLE 內直接建立）
• [ ] 資料回填與 Schema 變更拆分為獨立的遷移
• [ ] 已在正式環境資料的副本上完成測試
• [ ] 已記錄復原（Rollback）計畫

PostgreSQL 最佳模式

安全新增欄位

-- 好：允許為 Null 的欄位，不會鎖表
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）內執行
-- 大多數遷移工具需要特別處理此狀況

重新命名欄位（零停機）

切勿直接在正式環境重新命名欄位，請使用擴充-收縮模式（Expand-Contract Pattern）：

-- 步驟 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 從 Model 中移除
-- 而不產生 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 變更後重新產生 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 遷移

處理 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 偏離（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 變更產生遷移檔案
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 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

# 套用所有待處理的遷移
migrate -path migrations -database "$DATABASE_URL" up

# 復原上一次遷移
migrate -path migrations -database "$DATABASE_URL" down 1

# 強制指定版本（修復 dirty 狀態）
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;

零停機遷移策略（Zero-Downtime Migration Strategy）

針對關鍵的正式環境變更，請採用擴充-收縮模式（Expand-Contract Pattern）：

階段 1：擴充（EXPAND）
  - 新增欄位/資料表（允許為 Null 或包含預設值）
  - 部署：應用程式同時寫入舊欄位與新欄位
  - 回填既有資料

階段 2：遷移（MIGRATE）
  - 部署：應用程式改從新欄位讀取，但依然同時寫入新舊欄位
  - 驗證資料一致性

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

時程範例

第 1 天：遷移新增 new_status 欄位（允許為 Null）
第 1 天：部署 App v2 — 同時寫入 status 與 new_status
第 2 天：針對既有資料列執行回填遷移
第 3 天：部署 App v3 — 僅從 new_status 讀取
第 7 天：遷移刪除舊的 status 欄位

反模式（Anti-Patterns）

反模式	為何失敗	更好的作法
在正式環境直接手動執行 SQL	缺乏稽核軌跡，無法重複執行	務必使用遷移檔案
修改已部署的遷移	導致不同環境間出現 Schema 偏離	建立新的遷移檔案
新增無預設值的 NOT NULL	鎖定資料表，重寫所有資料列	先新增為可空，回填資料後再加約束
在大表上直接建立內嵌索引	建置期間會阻塞寫入	使用 CREATE INDEX CONCURRENTLY
在同一筆遷移中混合 Schema 與資料	難以復原，且交易時間過長	拆分為獨立的遷移
尚未移除程式碼就刪除欄位	應用程式因找不到欄位而報錯	先移除程式碼，下次部署再刪除欄位