database

数据库管理

优化查询、管理架构并确保可靠性。

适用场景

• 慢查询优化
• 架构设计与迁移
• 索引策略
• 数据库运维
• 性能调优

查询优化

分析查询

-- PostgreSQL
EXPLAIN (ANALYZE, BUFFERS, FORMAT TEXT)
SELECT * FROM orders WHERE user_id = 123;

-- MySQL
EXPLAIN ANALYZE
SELECT * FROM orders WHERE user_id = 123;

常见优化

-- Add missing index
CREATE INDEX CONCURRENTLY idx_orders_user_id
ON orders(user_id);

-- Composite index for common queries
CREATE INDEX idx_orders_user_date
ON orders(user_id, created_at DESC);

-- Partial index for filtered queries
CREATE INDEX idx_active_orders
ON orders(user_id) WHERE status = 'active';

-- Cover index to avoid table lookup
CREATE INDEX idx_orders_covering
ON orders(user_id) INCLUDE (total, status);

迁移最佳实践

-- Safe column addition (no lock)
ALTER TABLE users ADD COLUMN preferences JSONB;

-- Safe column rename (use view for compatibility)
ALTER TABLE users RENAME COLUMN name TO full_name;
CREATE VIEW users_compat AS
  SELECT *, full_name as name FROM users;

-- Safe index creation
CREATE INDEX CONCURRENTLY idx_new ON table(column);

-- Backfill in batches
UPDATE users SET new_col = compute(old_col)
WHERE id BETWEEN 1 AND 10000;
-- Repeat for next batch

运维查询

健康检查

-- Active connections (PostgreSQL)
SELECT state, count(*)
FROM pg_stat_activity
GROUP BY state;

-- Long running queries
SELECT pid, now() - query_start as duration, query
FROM pg_stat_activity
WHERE state = 'active' AND now() - query_start > '5 minutes'::interval;

-- Table sizes
SELECT relname, pg_size_pretty(pg_total_relation_size(relid))
FROM pg_catalog.pg_statio_user_tables
ORDER BY pg_total_relation_size(relid) DESC
LIMIT 10;

-- Index usage
SELECT indexrelname, idx_scan, idx_tup_read
FROM pg_stat_user_indexes
ORDER BY idx_scan DESC;

维护

-- PostgreSQL vacuum and analyze
VACUUM ANALYZE table_name;

-- Reindex
REINDEX INDEX CONCURRENTLY idx_name;

-- Kill long query
SELECT pg_terminate_backend(pid);

缓存策略

import redis

cache = redis.Redis()

def get_user(user_id: int) -> dict:
    # Try cache first
    cached = cache.get(f"user:{user_id}")
    if cached:
        return json.loads(cached)

    # Query database
    user = db.query("SELECT * FROM users WHERE id = %s", user_id)

    # Cache with TTL
    cache.setex(f"user:{user_id}", 3600, json.dumps(user))
    return user

def invalidate_user(user_id: int):
    cache.delete(f"user:{user_id}")

示例

输入：“这个查询很慢” **操作：**运行 EXPLAIN，找出缺失的索引或不佳的执行计划，并进行优化

输入：“设置数据库备份” **操作：**配置 pg_dump/mysqldump，设置计划任务并测试恢复