name: database-optimization description: "資料庫效能最佳化專家技能。涵蓋 SQL 查詢最佳化、索引策略設計、查詢計劃分析、ORM 最佳化、連線池調優。支援 PostgreSQL、MySQL、SQLite。當用戶遇到：慢查詢、資料庫效能問題、需要設計索引、分析查詢計劃、最佳化 SQL 語句、解決 N+1 查詢問題、調優連線池、分析資料庫鎖、最佳化 ORM 查詢、資料庫監控診斷時，立即使用此技能。即使使用者只說'查詢很慢'、'加個索引'、'SQL 最佳化'、'資料庫卡了'，也應觸發此技能。" version: 1.0.0

Database Optimization — 資料庫效能最佳化

最佳化方法論

1. 度量 → 2. 分析 → 3. 最佳化 → 4. 驗證
   ↑                                ↓
   └────────── 持續監控 ←───────────┘

鐵律：不度量就不最佳化。 先拿到基線資料，再做改動，最後驗證效果。

一、查詢分析（第一步永遠是這個）

1.1 PostgreSQL — EXPLAIN ANALYZE

-- 基礎：檢視執行計劃
EXPLAIN SELECT * FROM users WHERE email = 'test@example.com';

-- 完整：實際執行 + 真實耗時（推薦）
EXPLAIN (ANALYZE, BUFFERS, FORMAT TEXT) 
SELECT * FROM users WHERE email = 'test@example.com';

-- JSON 格式（便於程式解析）
EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) 
SELECT * FROM users WHERE email = 'test@example.com';

關鍵指標解讀：

1.2 MySQL — EXPLAIN

-- 基礎執行計劃
EXPLAIN SELECT * FROM users WHERE email = 'test@example.com';

-- 更詳細（MySQL 8.0+）
EXPLAIN ANALYZE SELECT * FROM users WHERE email = 'test@example.com';

-- 格式化輸出
EXPLAIN FORMAT=JSON SELECT * FROM users WHERE email = 'test@example.com';

關鍵指標：

1.3 SQLite

-- 開啟查詢計劃
EXPLAIN QUERY PLAN SELECT * FROM users WHERE email = 'test@example.com';

-- 開啟統計
.timer on
SELECT * FROM users WHERE email = 'test@example.com';

二、索引策略

2.1 索引型別速查

2.2 索引設計原則

原則 1：為查詢設計索引，不是為表

-- ❌ 錯誤：給每列單獨建索引
CREATE INDEX idx_users_email ON users(email);
CREATE INDEX idx_users_status ON users(status);
CREATE INDEX idx_users_created ON users(created_at);

-- ✅ 正確：分析查詢模式，設計組合索引
-- 查詢：WHERE status = 'active' ORDER BY created_at DESC
CREATE INDEX idx_users_status_created ON users(status, created_at DESC);

原則 2：最左字首匹配

-- 索引 (a, b, c)
-- ✅ 能使用：WHERE a=1 / WHERE a=1 AND b=2 / WHERE a=1 AND b=2 AND c=3
-- ❌ 不能使用：WHERE b=2 / WHERE c=3 / WHERE b=2 AND c=3

原則 3：覆蓋索引避免回表

-- 查詢只需要 email, name
-- ❌ 需要回表
CREATE INDEX idx_users_email ON users(email);
SELECT email, name FROM users WHERE email = 'test@example.com';

-- ✅ 覆蓋索引（PostgreSQL）
CREATE INDEX idx_users_email_cover ON users(email) INCLUDE (name);
SELECT email, name FROM users WHERE email = 'test@example.com';

-- ✅ 覆蓋索引（MySQL）
CREATE INDEX idx_users_email_name ON users(email, name);

原則 4：部分索引（PostgreSQL）

-- 只索引活躍使用者（減少索引大小）
CREATE INDEX idx_users_active_email ON users(email) WHERE is_active = true;

-- 只索引未刪除的訂單
CREATE INDEX idx_orders_pending ON orders(created_at) WHERE status = 'pending';

2.3 索引反模式

2.4 查詢未使用的索引

-- PostgreSQL
SELECT schemaname, tablename, indexname, idx_scan, idx_tup_read, idx_tup_fetch
FROM pg_stat_user_indexes
WHERE idx_scan = 0  -- 從未被使用
ORDER BY pg_relation_size(indexrelid) DESC;

-- MySQL (需要 performance_schema)
SELECT object_schema, object_name, index_name, count_star
FROM performance_schema.events_statements_summary_by_digest
WHERE index_name NOT IN ('PRIMARY', 'NULL')
AND count_star = 0;

三、SQL 最佳化模式

3.1 分頁最佳化

-- ❌ 深分頁效能差
SELECT * FROM posts ORDER BY id LIMIT 10 OFFSET 100000;

-- ✅ 游標分頁（推薦）
SELECT * FROM posts WHERE id > :last_id ORDER BY id LIMIT 10;

-- ✅ 延遲關聯（PostgreSQL）
SELECT p.* FROM posts p
JOIN (SELECT id FROM posts ORDER BY id LIMIT 10 OFFSET 100000) t ON p.id = t.id;

3.2 COUNT 最佳化

-- ❌ 全表掃描
SELECT COUNT(*) FROM orders WHERE status = 'pending';

-- ✅ 方案 1：維護計數表
CREATE TABLE order_counters (status TEXT PRIMARY KEY, count BIGINT DEFAULT 0);
-- 用觸發器維護

-- ✅ 方案 2：估算（PostgreSQL）
SELECT reltuples::BIGINT AS estimate FROM pg_class WHERE relname = 'orders';

-- ✅ 方案 3：部分索引 + COUNT
CREATE INDEX idx_orders_pending ON orders(id) WHERE status = 'pending';
SELECT COUNT(*) FROM orders WHERE status = 'pending';  -- 只掃描索引

3.3 JOIN 最佳化

-- ❌ 笛卡爾積風險
SELECT * FROM users, orders WHERE users.id = orders.user_id;

-- ✅ 明確 JOIN
SELECT u.name, o.total 
FROM users u
INNER JOIN orders o ON u.id = o.user_id
WHERE o.created_at > NOW() - INTERVAL '7 days';

-- ✅ 先過濾再 JOIN
SELECT u.name, recent.total
FROM users u
JOIN (SELECT user_id, SUM(amount) as total 
      FROM orders 
      WHERE created_at > NOW() - INTERVAL '7 days'
      GROUP BY user_id) recent ON u.id = recent.user_id;

3.4 子查詢 vs JOIN

-- ❌ 相關子查詢（每行執行一次）
SELECT name, (SELECT COUNT(*) FROM orders WHERE user_id = users.id) as order_count
FROM users;

-- ✅ JOIN + GROUP BY
SELECT u.name, COUNT(o.id) as order_count
FROM users u
LEFT JOIN orders o ON u.id = o.user_id
GROUP BY u.id, u.name;

-- ✅  lateral JOIN（PostgreSQL，複雜場景）
SELECT u.name, recent.order_count
FROM users u
JOIN LATERAL (
    SELECT COUNT(*) as order_count 
    FROM orders WHERE user_id = u.id
) recent ON true;

3.5 UPSERT 最佳化

-- PostgreSQL: INSERT ... ON CONFLICT
INSERT INTO users (email, name) VALUES ('test@example.com', 'Test')
ON CONFLICT (email) DO UPDATE SET name = EXCLUDED.name;

-- MySQL: INSERT ... ON DUPLICATE KEY
INSERT INTO users (email, name) VALUES ('test@example.com', 'Test')
ON DUPLICATE KEY UPDATE name = VALUES(name);

四、ORM 最佳化（SQLAlchemy）

4.1 N+1 問題

# ❌ N+1 查詢：1次查使用者 + N次查訂單
users = db.execute(select(User)).scalars().all()
for user in users:
    print(user.orders)  # 每個 user 觸發一次查詢

# ✅ 預載入（selectin）
users = db.execute(
    select(User).options(selectinload(User.orders))
).scalars().all()

# ✅ 預載入（joined）— 單條 SQL
users = db.execute(
    select(User).options(joinedload(User.orders))
).scalars().all()

4.2 只查需要的列

# ❌ 查所有列
users = db.execute(select(User)).scalars().all()

# ✅ 只查需要的
user_names = db.execute(select(User.id, User.name)).all()

# ✅ 使用 with_expression 計算欄位
from sqlalchemy import case
stmt = select(
    User.name,
    case((User.is_active == True, '活躍'), else_='停用').label('status')
)

4.3 批次操作

# ❌ 逐條插入
for item in items:
    db.add(Item(**item))
await db.flush()

# ✅ 批次插入
from sqlalchemy import insert
await db.execute(insert(Item), items)

# ✅ 批次更新（SQLAlchemy 2.0）
from sqlalchemy import update
await db.execute(
    update(Item).where(Item.status == 'pending').values(status='processing')
)

4.4 避免隱式查詢

# ❌ 觸發額外查詢（延遲載入）
user = db.get(User, 1)
print(user.posts)  # 又觸發一次查詢

# ✅ 明確載入策略
user = db.execute(
    select(User).options(selectinload(User.posts)).where(User.id == 1)
).scalar_one()

# ✅ 或者使用 joinedload
user = db.execute(
    select(User).options(joinedload(User.posts)).where(User.id == 1)
).scalar_one()

五、連線池調優

5.1 引數說明

5.2 SQLAlchemy 配置

engine = create_async_engine(
    DATABASE_URL,
    pool_size=20,           # 保持 20 個連線
    max_overflow=10,        # 最多溢位 10 個
    pool_timeout=30,        # 等待 30 秒
    pool_recycle=3600,      # 1 小時回收
    pool_pre_ping=True,     # 自動重連
    connect_args={
        "command_timeout": 60,  # 查詢超時
    },
)

5.3 監控連線池

# SQLAlchemy 連線池統計
pool = engine.pool
print(f"Size: {pool.size()}")
print(f"Checked in: {pool.checkedin()}")
print(f"Checked out: {pool.checkedout()}")
print(f"Overflow: {pool.overflow()}")

六、鎖與併發

6.1 死鎖診斷

-- PostgreSQL：檢視當前鎖
SELECT pid, usename, query, state, wait_event_type, wait_event
FROM pg_stat_activity
WHERE state = 'active' AND wait_event_type = 'Lock';

-- PostgreSQL：檢視死鎖
SELECT blocked.pid AS blocked_pid,
       blocked.query AS blocked_query,
       blocking.pid AS blocking_pid,
       blocking.query AS blocking_query
FROM pg_stat_activity blocked
JOIN pg_locks bl ON bl.pid = blocked.pid AND NOT bl.granted
JOIN pg_locks gl ON gl.locktype = bl.locktype AND gl.relation = bl.relation AND gl.granted
JOIN pg_stat_activity blocking ON gl.pid = blocking.pid
WHERE blocked.pid != blocking.pid;

-- MySQL：檢視鎖等待
SELECT * FROM information_schema.innodb_lock_waits;

6.2 樂觀鎖 vs 悲觀鎖

# 悲觀鎖（SELECT FOR UPDATE）
async with db.begin():
    result = await db.execute(
        select(Account).where(Account.id == account_id).with_for_update()
    )
    account = result.scalar_one()
    account.balance -= amount

# 樂觀鎖（版本號）
class Account(Base):
    __tablename__ = "accounts"
    id: Mapped[int] = mapped_column(primary_key=True)
    balance: Mapped[Decimal] = mapped_column(Numeric(10, 2))
    version: Mapped[int] = mapped_column(Integer, default=0)

    __mapper_args__ = {"version_id_col": version}

# 更新時自動檢查版本
try:
    account.balance -= amount
    await db.commit()
except OptimisticError:
    # 版本衝突，重試

七、慢查詢診斷清單

7.1 PostgreSQL

-- 檢視慢查詢（需要 pg_stat_statements 擴充套件）
CREATE EXTENSION IF NOT EXISTS pg_stat_statements;

SELECT query, calls, total_exec_time / 1000 as total_sec,
       mean_exec_time / 1000 as avg_sec,
       rows
FROM pg_stat_statements
ORDER BY total_exec_time DESC
LIMIT 20;

-- 查看錶大小
SELECT relname, 
       pg_size_pretty(pg_total_relation_size(relid)) as total_size,
       pg_size_pretty(pg_relation_size(relid)) as data_size,
       pg_size_pretty(pg_total_relation_size(relid) - pg_relation_size(relid)) as index_size
FROM pg_catalog.pg_statio_all_tables
ORDER BY pg_total_relation_size(relid) DESC
LIMIT 20;

-- 檢視快取命中率
SELECT sum(heap_blks_read) as heap_read,
       sum(heap_blks_hit) as heap_hit,
       round(sum(heap_blks_hit) * 100.0 / (sum(heap_blks_hit) + sum(heap_blks_read)), 2) as hit_ratio
FROM pg_statio_user_tables;

7.2 MySQL

小蔥技能站7w4.net每天更新，海量AI技能等你發現。

-- 慢查詢日誌
SET GLOBAL slow_query_log = 'ON';
SET GLOBAL long_query_time = 1;  -- 超過1秒記錄

-- 檢視當前執行的查詢
SHOW PROCESSLIST;

-- 查看錶狀態
SHOW TABLE STATUS LIKE 'users';

-- InnoDB 狀態
SHOW ENGINE INNODB STATUS;

八、最佳化決策樹

查詢慢？
│
├─ EXPLAIN 看了嗎？
│  └─ 沒看 → 先 EXPLAIN ANALYZE
│
├─ 全表掃描 (Seq Scan / ALL)？
│  ├─ 表很小 (<1000行) → 正常，不用管
│  └─ 表大 → 需要索引
│     ├─ WHERE 條件列 → 加 B-Tree 索引
│     ├─ ORDER BY 列 → 加索引或調整索引順序
│     └─ JOIN 條件 → 加索引
│
├─ 索引存在但沒用到？
│  ├─ 型別不匹配 → 檢查隱式轉換
│  ├─ 函式包裹 → 用函式索引
│  ├─ OR 條件 → 改 UNION 或分別建索引
│  └─ 統計資訊過期 → ANALYZE table
│
├─ 排序慢 (filesort / external merge)？
│  ├─ work_mem 太小 → 增大 work_mem
│  └─ 無索引排序 → 加覆蓋索引
│
├─ N+1 查詢？
│  └─ 用 selectinload / joinedload
│
└─ 鎖等待？
   ├─ 長事務 → 縮短事務範圍
   ├─ 死鎖 → 統一加鎖順序
   └─ 熱點行 → 樂觀鎖 / 佇列化

九、快速參考

質量良好。覆蓋主流資料庫的查詢最佳化、索引設計等核心場景，方法論清晰、示例實用，風險提示到位。但缺少真實案例演示，且僅有中文版本。適合需要系統性學習資料庫最佳化的開發者使用。日常使用夠用，但進階使用者可能覺得深度和廣度都有提升空間。