PostgreSQL 完全指南 2026：SQL 进阶 + 索引优化 + 分区表 + 高可用方案

1
-- 基础 CTE：提高可读性，避免子查询嵌套
2
WITH active_users AS (
3
    SELECT id, name, email, created_at
4
    FROM users
5
    WHERE last_login > NOW() - INTERVAL '30 days'
6
      AND status = 'active'
7
),
8
user_orders AS (
9
    SELECT user_id, COUNT(*) AS order_count, SUM(amount) AS total_amount
10
    FROM orders
11
    WHERE created_at > NOW() - INTERVAL '30 days'
12
    GROUP BY user_id
13
)
14
SELECT
15
    u.name,
16
    u.email,
17
    COALESCE(o.order_count, 0)  AS orders_30d,
18
    COALESCE(o.total_amount, 0) AS revenue_30d
19
FROM active_users u
20
LEFT JOIN user_orders o ON u.id = o.user_id
21
ORDER BY revenue_30d DESC;
22

23
-- 递归 CTE：遍历树形结构（组织架构、分类目录等）
24
WITH RECURSIVE org_tree AS (
25
    -- 锚点：从 CEO 开始
26
    SELECT id, name, manager_id, 0 AS depth, ARRAY[name] AS path
27
    FROM employees
28
    WHERE manager_id IS NULL
29

30
    UNION ALL
31

32
    -- 递归：逐层获取下属
33
    SELECT e.id, e.name, e.manager_id, t.depth + 1, t.path || e.name
34
    FROM employees e
35
    INNER JOIN org_tree t ON e.manager_id = t.id
36
    WHERE t.depth < 10   -- 防止无限递归
37
)
38
SELECT depth, path, name
39
FROM org_tree
40
ORDER BY path;

1
-- 常用窗口函数示例（员工薪资分析）
2
SELECT
3
    dept,
4
    name,
5
    salary,
6
    -- 部门内排名（相同薪资并列，有跳号）
7
    RANK()        OVER (PARTITION BY dept ORDER BY salary DESC) AS dept_rank,
8
    -- 无跳号排名
9
    DENSE_RANK()  OVER (PARTITION BY dept ORDER BY salary DESC) AS dept_dense_rank,
10
    -- 行号（按薪资排序的全局行号）
11
    ROW_NUMBER()  OVER (ORDER BY salary DESC)                   AS global_row,
12
    -- 部门平均薪资（保留所有行）
13
    ROUND(AVG(salary) OVER (PARTITION BY dept), 2)              AS dept_avg,
14
    -- 当前行薪资 vs 部门平均的百分比
15
    ROUND(salary / AVG(salary) OVER (PARTITION BY dept) * 100, 1) AS pct_of_avg,
16
    -- 前一位员工薪资（LAG）
17
    LAG(salary,  1, 0) OVER (PARTITION BY dept ORDER BY salary DESC) AS prev_salary,
18
    -- 后一位员工薪资（LEAD）
19
    LEAD(salary, 1, 0) OVER (PARTITION BY dept ORDER BY salary DESC) AS next_salary,
20
    -- 累计薪资（滚动求和）
21
    SUM(salary) OVER (PARTITION BY dept ORDER BY salary DESC ROWS UNBOUNDED PRECEDING) AS running_total
22
FROM employees;
23

24

25
-- 实用示例：找每个分类下最新的 5 篇文章（TOP N per group）
26
WITH ranked AS (
27
    SELECT *,
28
           ROW_NUMBER() OVER (PARTITION BY category ORDER BY published_at DESC) AS rn
29
    FROM articles
30
    WHERE status = 'published'
31
)
32
SELECT id, title, category, published_at
33
FROM ranked
34
WHERE rn <= 5;

1
-- 创建带 JSONB 列的表
2
CREATE TABLE events (
3
    id         BIGSERIAL PRIMARY KEY,
4
    event_type TEXT NOT NULL,
5
    payload    JSONB NOT NULL,
6
    created_at TIMESTAMPTZ DEFAULT NOW()
7
);
8

9
-- 插入数据
10
INSERT INTO events (event_type, payload) VALUES
11
('user.login',   '{"user_id": 42, "ip": "1.2.3.4", "device": "mobile"}'),
12
('order.placed', '{"order_id": 101, "items": [{"id": 5, "qty": 2}], "total": 99.9}');
13

14
-- ── 查询操作符 ───────────────────────────────────────────────
15
-- -> 返回 JSON 对象（JSON 类型）
16
-- ->> 返回文本（TEXT 类型）
17
SELECT payload -> 'user_id'    AS user_id_json,
18
       payload ->> 'user_id'   AS user_id_text,
19
       payload -> 'device'     AS device
20
FROM events WHERE event_type = 'user.login';
21

22
-- 多层路径
23
SELECT payload #>> '{items,0,id}' AS first_item_id
24
FROM events WHERE event_type = 'order.placed';
25

26
-- @> 包含操作符（检查 JSON 包含关系）
27
SELECT * FROM events
28
WHERE payload @> '{"device": "mobile"}';
29

30
-- ? 键存在性检查
31
SELECT * FROM events WHERE payload ? 'ip';
32

33
-- jsonb_array_elements：展开数组
34
SELECT e.id, item->>'id' AS item_id, item->>'qty' AS qty
35
FROM events e, jsonb_array_elements(payload->'items') AS item
36
WHERE event_type = 'order.placed';
37

38
-- 更新 JSONB 字段
39
UPDATE events
40
SET payload = payload || '{"processed": true}'   -- 合并更新
41
WHERE id = 1;
42

43
UPDATE events
44
SET payload = payload - 'ip'                      -- 删除字段
45
WHERE event_type = 'user.login';
46

47
-- ── GIN 索引（JSONB 全路径查询加速）────────────────────────
48
CREATE INDEX idx_events_payload ON events USING GIN (payload);
49
-- 或只索引特定路径（更小更快）
50
CREATE INDEX idx_events_user_id ON events USING GIN ((payload -> 'user_id'));

1
-- 普通 B-Tree
2
CREATE INDEX idx_users_email ON users (email);
3

4
-- 复合索引（列顺序重要：高选择性列在前，常用过滤条件在前）
5
CREATE INDEX idx_orders_user_status ON orders (user_id, status, created_at);
6

7
-- 部分索引（只索引满足条件的行，体积小效果好）
8
CREATE INDEX idx_orders_pending ON orders (created_at)
9
WHERE status = 'pending';
10

11
-- 表达式索引
12
CREATE INDEX idx_users_lower_email ON users (LOWER(email));
13
-- 对应查询必须使用 LOWER(email) 才能命中
14
SELECT * FROM users WHERE LOWER(email) = 'alice@example.com';
15

16
-- GIN 全文搜索索引
17
ALTER TABLE articles ADD COLUMN search_vec TSVECTOR
18
    GENERATED ALWAYS AS (to_tsvector('chinese', title || ' ' || content)) STORED;
19
CREATE INDEX idx_articles_fts ON articles USING GIN (search_vec);
20

21
-- 全文搜索查询
22
SELECT title FROM articles
23
WHERE search_vec @@ to_tsquery('chinese', '数据库 & 优化')
24
ORDER BY ts_rank(search_vec, to_tsquery('chinese', '数据库 & 优化')) DESC;
25

26
-- BRIN 索引（时序大表，极省空间）
27
CREATE INDEX idx_logs_created_brin ON logs USING BRIN (created_at)
28
WITH (pages_per_range = 128);

1
-- 开启慢查询日志（postgresql.conf）
2
-- log_min_duration_statement = 1000  -- 超过1秒的查询写日志
3

4
-- 查看当前正在执行的查询
5
SELECT pid, now() - pg_stat_activity.query_start AS duration,
6
       query, state
7
FROM pg_stat_activity
8
WHERE state != 'idle' AND query_start < NOW() - INTERVAL '5 seconds'
9
ORDER BY duration DESC;
10

11
-- 查看缺少索引的慢查询（pg_stat_statements 扩展）
12
SELECT query,
13
       calls,
14
       ROUND(total_exec_time / calls, 2) AS avg_ms,
15
       ROUND(total_exec_time, 2)          AS total_ms,
16
       rows
17
FROM pg_stat_statements
18
ORDER BY avg_ms DESC
19
LIMIT 20;
20

21
-- EXPLAIN ANALYZE（实际执行计划 + 耗时）
22
EXPLAIN (ANALYZE, BUFFERS, FORMAT TEXT)
23
SELECT u.name, COUNT(o.id)
24
FROM users u
25
LEFT JOIN orders o ON u.id = o.user_id
26
WHERE u.created_at > '2026-01-01'
27
GROUP BY u.id
28
ORDER BY COUNT(o.id) DESC
29
LIMIT 10;

1
-- 重置统计信息
2
ANALYZE users;          -- 更新单表统计
3
ANALYZE;                -- 更新所有表统计
4

5
-- 查看表索引使用情况（找无用索引）
6
SELECT schemaname, tablename, indexname,
7
       idx_scan,       -- 索引被使用的次数（0 = 从未使用！）
8
       idx_tup_read,
9
       idx_tup_fetch
10
FROM pg_stat_user_indexes
11
ORDER BY idx_scan ASC;

1
import asyncio
2
import psycopg
3
from psycopg.rows import dict_row
4

5
async def main():
6
    # 连接字符串
7
    conn_str = "postgresql://user:password@localhost/mydb"
8

9
    async with await psycopg.AsyncConnection.connect(
10
        conn_str, row_factory=dict_row
11
    ) as conn:
12
        # 查询
13
        async with conn.cursor() as cur:
14
            await cur.execute(
15
                "SELECT id, name, email FROM users WHERE status = %s LIMIT %s",
16
                ("active", 10)
17
            )
18
            users = await cur.fetchall()
19
            for user in users:
20
                print(user["name"], user["email"])
21

22
        # 事务
23
        async with conn.transaction():
24
            await conn.execute(
25
                "INSERT INTO orders (user_id, amount) VALUES (%s, %s)",
26
                (42, 99.9)
27
            )
28
            await conn.execute(
29
                "UPDATE users SET order_count = order_count + 1 WHERE id = %s",
30
                (42,)
31
            )
32

33

34
# 异步连接池（生产推荐）
35
from psycopg_pool import AsyncConnectionPool
36

37
pool = AsyncConnectionPool(
38
    conn_str,
39
    min_size=5,
40
    max_size=20,
41
    kwargs={"row_factory": dict_row},
42
)
43

44
async def get_user(user_id: int) -> dict | None:
45
    async with pool.connection() as conn:
46
        result = await conn.fetchone(
47
            "SELECT * FROM users WHERE id = %s", (user_id,)
48
        )
49
        return result

1
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
2
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
3
from sqlalchemy import String, Integer, ForeignKey, select
4
from datetime import datetime
5

6
# 引擎（异步）
7
engine = create_async_engine(
8
    "postgresql+psycopg://user:password@localhost/mydb",
9
    pool_size=20,
10
    max_overflow=10,
11
    echo=False,
12
)
13

14
AsyncSessionFactory = async_sessionmaker(engine, expire_on_commit=False)
15

16
# ── ORM 模型（SQLAlchemy 2.0 风格）────────────────────────────
17
class Base(DeclarativeBase):
18
    pass
19

20
class User(Base):
21
    __tablename__ = "users"
22

23
    id:         Mapped[int]           = mapped_column(primary_key=True)
24
    name:       Mapped[str]           = mapped_column(String(100))
25
    email:      Mapped[str]           = mapped_column(String(200), unique=True)
26
    created_at: Mapped[datetime]      = mapped_column(default=datetime.utcnow)
27
    orders:     Mapped[list["Order"]] = relationship(back_populates="user")
28

29
class Order(Base):
30
    __tablename__ = "orders"
31

32
    id:      Mapped[int]   = mapped_column(primary_key=True)
33
    user_id: Mapped[int]   = mapped_column(ForeignKey("users.id"))
34
    amount:  Mapped[float]
35
    user:    Mapped[User]  = relationship(back_populates="orders")
36

37
# ── 异步 CRUD ─────────────────────────────────────────────────
38
async def get_users_with_orders(min_amount: float) -> list[User]:
39
    async with AsyncSessionFactory() as session:
40
        stmt = (
41
            select(User)
42
            .join(User.orders)
43
            .where(Order.amount >= min_amount)
44
            .order_by(User.name)
45
        )
46
        result = await session.execute(stmt)
47
        return result.scalars().unique().all()
48

49
async def create_user_with_order(name: str, email: str, amount: float) -> User:
50
    async with AsyncSessionFactory() as session:
51
        async with session.begin():     # 自动提交/回滚
52
            user = User(name=name, email=email)
53
            session.add(user)
54
            await session.flush()       # 获取 user.id（未提交）
55

56
            order = Order(user_id=user.id, amount=amount)
57
            session.add(order)
58
            # session.begin() 退出时自动 COMMIT
59
        return user