Schema 遷移工具完全比較：Flyway、Alembic、Prisma、TypeORM 與零停機遷移策略 | PostgreSQL

-- V1__create_users_table.sql
CREATE TABLE users (
    id          BIGSERIAL PRIMARY KEY,
    username    VARCHAR(50) NOT NULL UNIQUE,
    email       VARCHAR(255) NOT NULL UNIQUE,
    created_at  TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at  TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

CREATE INDEX idx_users_email ON users (email);

COMMENT ON TABLE users IS '使用者帳號主表';

-- V2__add_user_profile.sql
ALTER TABLE users
    ADD COLUMN display_name  VARCHAR(100),
    ADD COLUMN avatar_url    TEXT,
    ADD COLUMN is_active     BOOLEAN NOT NULL DEFAULT TRUE;

-- V3__create_posts_table.sql
CREATE TABLE posts (
    id          BIGSERIAL PRIMARY KEY,
    author_id   BIGINT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
    title       VARCHAR(500) NOT NULL,
    content     TEXT,
    status      VARCHAR(20) NOT NULL DEFAULT 'draft'
                    CHECK (status IN ('draft', 'published', 'archived')),
    published_at TIMESTAMPTZ,
    created_at  TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

CREATE INDEX idx_posts_author_id ON posts (author_id);
CREATE INDEX idx_posts_status ON posts (status) WHERE status = 'published';

# application.yml
spring:
  flyway:
    enabled: true
    locations: classpath:db/migration
    baseline-on-migrate: false
    validate-on-migrate: true
    out-of-order: false
    table: flyway_schema_history

flyway migrate   # 執行所有待執行的遷移
flyway info      # 查看遷移狀態
flyway validate  # 驗證已執行遷移的 checksum
flyway repair    # 修復失敗的遷移記錄
flyway clean     # 清空整個資料庫（危險！僅用於開發）

# db/changelog/001-create-users.yaml
databaseChangeLog:
  - changeSet:
      id: 001
      author: alice
      comment: "建立使用者主表"
      changes:
        - createTable:
            tableName: users
            columns:
              - column:
                  name: id
                  type: BIGINT
                  autoIncrement: true
                  constraints:
                    primaryKey: true
                    nullable: false
              - column:
                  name: username
                  type: VARCHAR(50)
                  constraints:
                    nullable: false
                    unique: true
              - column:
                  name: email
                  type: VARCHAR(255)
                  constraints:
                    nullable: false
                    unique: true

  - changeSet:
      id: 003
      author: bob
      comment: "新增個人資料欄位"
      changes:
        - addColumn:
            tableName: users
            columns:
              - column:
                  name: display_name
                  type: VARCHAR(100)
              - column:
                  name: is_active
                  type: BOOLEAN
                  defaultValueBoolean: true
                  constraints:
                    nullable: false
      rollback:
        - dropColumn:
            tableName: users
            columnName: display_name
        - dropColumn:
            tableName: users
            columnName: is_active

pip install alembic sqlalchemy psycopg2-binary

# 初始化
alembic init alembic

# alembic/env.py（關鍵配置）
from myapp.models import Base

target_metadata = Base.metadata

def run_migrations_online():
    connectable = engine_from_config(
        config.get_section(config.config_ini_section),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )
    with connectable.connect() as connection:
        context.configure(
            connection=connection,
            target_metadata=target_metadata,
            compare_type=True,           # 偵測欄位型別變更
            compare_server_default=True   # 偵測預設值變更
        )
        with context.begin_transaction():
            context.run_migrations()

# 自動比對模型與資料庫，生成遷移腳本
alembic revision --autogenerate -m "add posts table"

# 執行所有待執行的遷移
alembic upgrade head

# 回滾一個版本
alembic downgrade -1

# 查看當前版本
alembic current

# 查看遷移歷史
alembic history --verbose

"""add posts table

Revision ID: ae1027a6acf
Revises: 1234567890ab
"""
from alembic import op
import sqlalchemy as sa

revision = 'ae1027a6acf'
down_revision = '1234567890ab'


def upgrade() -> None:
    op.create_table(
        'posts',
        sa.Column('id', sa.BigInteger(), autoincrement=True, nullable=False),
        sa.Column('author_id', sa.BigInteger(), nullable=False),
        sa.Column('title', sa.String(500), nullable=False),
        sa.Column('content', sa.Text(), nullable=True),
        sa.Column('status', sa.String(20), nullable=False, server_default='draft'),
        sa.Column('created_at', sa.TIMESTAMP(timezone=True),
                  server_default=sa.text('NOW()'), nullable=False),
        sa.ForeignKeyConstraint(['author_id'], ['users.id'], ondelete='CASCADE'),
        sa.PrimaryKeyConstraint('id'),
        sa.CheckConstraint(
            "status IN ('draft', 'published', 'archived')",
            name='posts_status_check'
        )
    )
    op.create_index('idx_posts_author_id', 'posts', ['author_id'])


def downgrade() -> None:
    op.drop_index('idx_posts_author_id')
    op.drop_table('posts')

# 根據 Model 變更自動生成遷移
python manage.py makemigrations

# 執行所有待執行的遷移
python manage.py migrate

# 查看遷移生成的 SQL（不執行）
python manage.py sqlmigrate myapp 0003

# 查看遷移狀態
python manage.py showmigrations

# 在遷移中執行任意 SQL
migrations.RunSQL(
    sql="ALTER TABLE posts ADD COLUMN search_vector tsvector",
    reverse_sql="ALTER TABLE posts DROP COLUMN search_vector"
),

# 在遷移中執行 Python 邏輯（資料遷移）
def populate_full_name(apps, schema_editor):
    User = apps.get_model('myapp', 'User')
    for user in User.objects.all():
        user.full_name = f"{user.first_name} {user.last_name}".strip()
        user.save(update_fields=['full_name'])

migrations.RunPython(
    populate_full_name,
    reverse_code=migrations.RunPython.noop
),

// DataSource 配置
export const AppDataSource = new DataSource({
    type: 'postgres',
    host: process.env.DB_HOST,
    port: parseInt(process.env.DB_PORT ?? '5432'),
    username: process.env.DB_USER,
    password: process.env.DB_PASSWORD,
    database: process.env.DB_NAME,
    entities: ['src/entities/**/*.ts'],
    migrations: ['src/migrations/**/*.ts'],
    synchronize: false,  // 生產環境絕對不要設為 true
});

# 比對 Entity 與資料庫，自動生成遷移
npx typeorm migration:generate src/migrations/AddPostsTable -d ormconfig.ts

# 執行遷移
npx typeorm migration:run -d ormconfig.ts

# 回滾最後一個遷移
npx typeorm migration:revert -d ormconfig.ts

# 開發環境：生成遷移並套用
npx prisma migrate dev --name add-posts-table

# 生產環境：只套用已存在的遷移
npx prisma migrate deploy

# 查看遷移狀態
npx prisma migrate status

# 初始化
sqitch init myproject --engine pg

# 新增一個變更（生成 deploy/revert/verify 三個檔案）
sqitch add create-users-table -n '建立使用者主表'

-- deploy/create-users-table.sql
BEGIN;
CREATE TABLE users (
    id          BIGSERIAL PRIMARY KEY,
    username    VARCHAR(50) NOT NULL UNIQUE,
    email       VARCHAR(255) NOT NULL UNIQUE,
    created_at  TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
COMMIT;

-- revert/create-users-table.sql
BEGIN;
DROP TABLE IF EXISTS users CASCADE;
COMMIT;

-- verify/create-users-table.sql
SELECT id, username, email, created_at
FROM users
WHERE FALSE;  -- 只驗證結構，不查詢資料

sqitch deploy db:pg://user:pass@localhost/mydb   # 部署
sqitch revert db:pg://user:pass@localhost/mydb   # 回滾
sqitch verify db:pg://user:pass@localhost/mydb   # 驗證
sqitch status db:pg://user:pass@localhost/mydb   # 狀態

-- ========= 階段 1：Expand =========

-- 步驟 1a：新增新欄位
ALTER TABLE users ADD COLUMN username VARCHAR(50);

-- 步驟 1b：分批回填（避免長時間鎖定大表）
DO $$
DECLARE
    batch_size INT := 1000;
    last_id    BIGINT := 0;
    max_id     BIGINT;
BEGIN
    SELECT MAX(id) INTO max_id FROM users;
    WHILE last_id < max_id LOOP
        UPDATE users
        SET username = user_name
        WHERE id > last_id
          AND id <= last_id + batch_size
          AND username IS NULL;
        last_id := last_id + batch_size;
        PERFORM pg_sleep(0.01);
    END LOOP;
END;
$$;

-- 步驟 1c：CONCURRENTLY 建索引（不鎖定讀寫）
CREATE UNIQUE INDEX CONCURRENTLY idx_users_username ON users (username);

-- 步驟 1d：部署 v2（同時讀寫 user_name 與 username）

-- ========= 階段 2：驗證期 =========

SELECT COUNT(*) FROM users WHERE username IS DISTINCT FROM user_name;
-- 應為 0

-- ========= 階段 3：Contract =========

-- 步驟 3a：部署 v3（只讀寫 username）

-- 步驟 3b：加上 NOT NULL 約束（分兩步驟避免全表鎖）
ALTER TABLE users
    ADD CONSTRAINT users_username_not_null
    CHECK (username IS NOT NULL) NOT VALID;

ALTER TABLE users VALIDATE CONSTRAINT users_username_not_null;

ALTER TABLE users ALTER COLUMN username SET NOT NULL;

-- 步驟 3c：刪除舊欄位
ALTER TABLE users DROP COLUMN user_name;
ALTER TABLE users DROP CONSTRAINT users_username_not_null;

-- PG<11 全表重寫
ALTER TABLE large_table ADD COLUMN new_col TEXT DEFAULT 'value';
-- 全表重寫
ALTER TABLE large_table ALTER COLUMN col TYPE new_type;
-- 非 CONCURRENTLY 模式
CREATE INDEX ON large_table (col);

-- PG 11+：ADD COLUMN 有預設值不再需要全表重寫（瞬間完成）
ALTER TABLE large_table ADD COLUMN new_col TEXT DEFAULT 'safe';

-- CONCURRENTLY 建立索引（不鎖定讀寫）
CREATE INDEX CONCURRENTLY idx_name ON large_table (col);

-- NOT VALID + VALIDATE 分離加約束（不全表鎖）
ALTER TABLE large_table
    ADD CONSTRAINT col_not_null CHECK (col IS NOT NULL) NOT VALID;
ALTER TABLE large_table VALIDATE CONSTRAINT col_not_null;

-- 外鍵也用 NOT VALID + VALIDATE
ALTER TABLE posts
    ADD CONSTRAINT fk_author FOREIGN KEY (author_id) REFERENCES users(id)
    NOT VALID;
ALTER TABLE posts VALIDATE CONSTRAINT fk_author;

-- 在遷移腳本開頭設置，避免遷移長時間等待鎖
SET lock_timeout = '3s';
SET statement_timeout = '60s';

ALTER TABLE users ADD COLUMN new_col TEXT;

RESET lock_timeout;
RESET statement_timeout;

# 使用 Testcontainers 進行遷移測試
from testcontainers.postgres import PostgresContainer
from alembic import command
from alembic.config import Config

def test_migrations_up_and_down():
    with PostgresContainer("postgres:16") as pg:
        alembic_cfg = Config("alembic.ini")
        alembic_cfg.set_main_option("sqlalchemy.url", pg.get_connection_url())

        # 測試升級
        command.upgrade(alembic_cfg, "head")

        # 測試降級（回滾所有）
        command.downgrade(alembic_cfg, "base")

        # 再次升級，確保冪等性
        command.upgrade(alembic_cfg, "head")