第 19 章综合实战项目：博客 + 全文检索 + 标签 + 评论树 + AI 语义检索

收官章。把前 18 章的 PG 知识全部串起来，落地到一个真实可跑的项目上。
学完你会：独立设计一个中等规模业务的 PG 库表、选对索引、写清慢查询、搭出多租户隔离、并接入 AI 向量检索。

0. 导读：为什么选这个项目？

候选	难在哪	能覆盖的 PG 能力
电商订单系统	业务规则多、事务多	事务 / 外键 / 窗口函数 / 分区
附近商家（PostGIS）	需要另装 PostGIS	GIS / GiST 索引
博客 + 全文 + 向量	PG 特色最密集	JSONB / GIN / pg_trgm / pgvector / RLS / 分区 / 物化视图 / CTE / 触发器 / pg_stat_statements
RAG 向量库	只突出 pgvector	单一

我们最终选 博客系统，因为它的每一个小功能都正好能映射到前 18 章的某个知识点——「一个项目讲完所有 PG 核心特性」。

1. 项目目标与功能清单

1.1 项目目标

做一个 「多租户博客平台」，支持：

用户注册/登录（密码 bcrypt 哈希）
作者发文，文章带 JSONB 标签
按标签过滤、最新文章列表
三种检索同时支持：全文检索 / 模糊检索 / 语义检索
评论支持无限层级回复（评论树）
首页显示热门榜（物化视图，每小时刷新）
RLS 多租户：不同租户的数据物理存在一张表，却彼此不可见
文章按月分区归档，老数据可 DETACH
全套 慢查询审计 + 执行计划分析

1.2 用户故事

角色	故事
新用户	注册、登录
作者	发文（含多个标签）
访客	浏览最新 / 热门 / 标签下文章；搜索
访客	在文章下发多级回复
运营	每小时看最新的热门榜
平台	多租户隔离、跨租户数据不可见
DBA	用 `pg_stat_statements` 定位慢查询、加索引

1.3 REST 接口清单（代码见 `19_project/code/api.py`）

方法	路径	能力	用到的 PG 特性
POST	`/signup`	注册	`pgcrypto.crypt`
POST	`/login`	登录	`pgcrypto.crypt`
POST	`/posts`	发文	触发器自动维护 tsvector
GET	`/posts/recent`	最新	分区裁剪 + B-Tree
GET	`/posts/by_tag?tag=xx`	按标签	JSONB GIN
GET	`/search/fulltext?q=xx`	全文检索	tsvector + GIN
GET	`/search/trgm?q=xx`	模糊检索	pg_trgm
GET	`/search/vector?q=xx`	语义检索	pgvector + HNSW
GET	`/comments/tree?post_id=1`	评论树	递归 CTE
GET	`/hot`	热门榜	物化视图
POST	`/hot/refresh`	刷新热门	`REFRESH ... CONCURRENTLY`

2. 数据模型设计

2.1 ER 图（Mermaid）

2.2 设计要点（对应到前 18 章）

#	设计决策	对应章节	原因
1	用 `GENERATED ALWAYS AS IDENTITY` 而不是 `SERIAL`	Ch4 约束/序列	标准 SQL、权限更干净
2	密码存 `text`（bcrypt 串 ~60 字符），而不是 `varchar(60)`	Ch3 数据类型	PG 里 `text` 和 `varchar` 性能相同，避免长度硬编码
3	时间全用 `timestamptz`	Ch3 数据类型	避免时区坑，存 UTC
4	`tags` 用 JSONB 而不是关联表	Ch3 / Ch6	查询灵活 + GIN 索引 + 无 JOIN
5	`search_vector` 冗余列 + 触发器维护	Ch6 / Ch12	查询时不用现 `to_tsvector`，GIN 索引可落盘
6	`posts` 按 `created_at` RANGE 月分区	Ch16 分区	冷数据 DETACH、查询分区裁剪
7	`post_embeddings` 单独一张表	Ch18 扩展	分区表上不方便建 HNSW 向量索引
8	主键 `(id, created_at)`	Ch16 分区	分区表主键必须包含分区键
9	JSONB 用 `jsonb_path_ops`	Ch6 索引	比默认 `jsonb_ops` 小一半、且只支持 `@>`
10	`RLS` 策略 + `current_setting('app.tenant_id')`	Ch13 权限	多租户隔离，业务层无需额外 WHERE
11	热门榜用 `MATERIALIZED VIEW`	Ch4 视图 / Ch17 调优	聚合重、读多写少
12	`REFRESH CONCURRENTLY` 需要唯一索引	Ch6 索引	不阻塞读
13	`pg_stat_statements` + `auto_explain`	Ch17 调优	慢查询一把抓
14	评论树用递归 CTE	Ch5 高级查询	一次 SQL 查完，干掉 N+1
15	点赞表联合主键 `(user_id, post_id)`	Ch4 约束	天然防重复点赞
16	`post_embeddings` 用 HNSW 索引	Ch18 扩展	召回率 / 速度均衡最佳
17	业务角色 `blog_app` 不加 `BYPASSRLS`	Ch13 权限	防止误绕过 RLS

3. 完整 init.sql 逐段讲解

完整文件：19_project/init.sql。下面按章节分块解释。

3.1 扩展一次性装齐

sql

CREATE EXTENSION IF NOT EXISTS pgcrypto;             -- 密码哈希
CREATE EXTENSION IF NOT EXISTS pg_trgm;              -- 模糊搜索
CREATE EXTENSION IF NOT EXISTS btree_gin;            -- 让 GIN 索引能混入普通字段
CREATE EXTENSION IF NOT EXISTS vector;               -- pgvector
CREATE EXTENSION IF NOT EXISTS pg_stat_statements;   -- 慢查询统计
CREATE EXTENSION IF NOT EXISTS unaccent;             -- 去重音（对中文无害）

CREATE EXTENSION 机制见 Ch18。扩展本质是一组 SQL/C 函数的打包。
pg_stat_statements 属于 预加载型扩展，必须已在 postgresql.conf 里配 shared_preload_libraries='pg_stat_statements' 并重启，再 CREATE EXTENSION 才有效。

3.2 Schema 隔离

sql

CREATE SCHEMA IF NOT EXISTS blog;
SET search_path TO blog, public;

把业务放进独立 schema，对应 Ch2。好处：

生产 / 测试共用一个库，用 blog_prod / blog_test 区分；
方便一次性 DROP SCHEMA blog CASCADE 重置。

3.3 `users` 表：多租户 + bcrypt 密码

sql

CREATE TABLE users (
    id         BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    tenant_id  BIGINT NOT NULL DEFAULT 1,
    username   TEXT   NOT NULL,
    email      TEXT   NOT NULL,
    password   TEXT   NOT NULL,    -- bcrypt 串
    role       TEXT   NOT NULL DEFAULT 'reader'
                CHECK (role IN ('reader','author','admin')),
    profile    JSONB  NOT NULL DEFAULT '{}'::jsonb,
    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    UNIQUE (tenant_id, username),
    UNIQUE (tenant_id, email)
);

IDENTITY 替代 SERIAL（Ch4）。
复合唯一 (tenant_id, username) 保证「同一租户下用户名唯一，跨租户可重名」——多租户常用招式。
role 用 CHECK 约束代替 ENUM，改起来更快（Ch4）。
profile JSONB 用来装个性化资料、头像、偏好，避免无限加列（Ch3）。

3.4 `posts` 主表：分区 + JSONB + tsvector

sql

CREATE TABLE posts (
    id           BIGINT GENERATED ALWAYS AS IDENTITY,
    tenant_id    BIGINT NOT NULL DEFAULT 1,
    author_id    BIGINT NOT NULL,
    title        TEXT   NOT NULL,
    body         TEXT   NOT NULL,
    tags         JSONB  NOT NULL DEFAULT '[]'::jsonb,
    status       TEXT   NOT NULL DEFAULT 'draft'
                 CHECK (status IN ('draft','published','archived')),
    view_count   BIGINT NOT NULL DEFAULT 0,
    like_count   BIGINT NOT NULL DEFAULT 0,
    search_vector tsvector,
    created_at   TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at   TIMESTAMPTZ NOT NULL DEFAULT now(),
    PRIMARY KEY (id, created_at)                   -- 分区键必须包含
) PARTITION BY RANGE (created_at);

几个要点：

分区表（Ch16）主键必须包含分区键，这是 PG 声明式分区的硬性要求。
tags 用 JSONB 数组形式：["postgres","ai"]，查询时 tags @> '["postgres"]' 走 GIN 索引（Ch3/Ch6）。
search_vector tsvector（Ch6）冗余列，由触发器写入；如果每次查询都 to_tsvector(body) 现算，GIN 索引就没法用到。

然后一口气建 12 个月 + 1 个默认分区（截取片段）：

sql

CREATE TABLE posts_2025_01 PARTITION OF posts
    FOR VALUES FROM ('2025-01-01') TO ('2025-02-01');
...
CREATE TABLE posts_default PARTITION OF posts DEFAULT;

生产中这种「按月预建」推荐用 pg_partman 扩展自动维护（Ch18）。

3.5 父表上建索引自动下推

sql

CREATE INDEX idx_posts_fts        ON posts USING GIN (search_vector);
CREATE INDEX idx_posts_tags       ON posts USING GIN (tags jsonb_path_ops);
CREATE INDEX idx_posts_title_trgm ON posts USING GIN (title gin_trgm_ops);
CREATE INDEX idx_posts_author     ON posts (author_id);
CREATE INDEX idx_posts_tenant_status ON posts (tenant_id, status);

PG 11+ 在分区父表声明的索引会自动下推到所有现存和未来的子分区。

为什么是这 5 个索引？

索引	服务哪个查询	为什么不是 B-Tree？
`GIN(search_vector)`	全文检索 `@@`	B-Tree 不支持 tsquery 操作符
`GIN(tags jsonb_path_ops)`	JSONB `@>` 包含	B-Tree 只能整列等值
`GIN(title gin_trgm_ops)`	pg_trgm `%` 和 `<->`	B-Tree 不支持子串相似
`(author_id)`	按作者查	等值 / 范围，B-Tree 最优
`(tenant_id, status)`	多租户 + 发布态过滤	复合 B-Tree 左前缀匹配

3.6 `comments` 评论表：自关联

sql

CREATE TABLE comments (
    id         BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    tenant_id  BIGINT NOT NULL DEFAULT 1,
    post_id    BIGINT NOT NULL,
    parent_id  BIGINT REFERENCES comments(id) ON DELETE CASCADE,
    author_id  BIGINT NOT NULL,
    content    TEXT NOT NULL,
    created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE INDEX idx_comments_post   ON comments (post_id);
CREATE INDEX idx_comments_parent ON comments (parent_id);

parent_id REFERENCES comments(id) ON DELETE CASCADE：父评论被删子评论跟着走（Ch4）。
查评论树靠 递归 CTE（Ch5），稍后演示。
不做分区：评论量虽大，但按 post_id 查询天然有 B-Tree 索引，分区收益不大。

3.7 `post_embeddings`：pgvector 语义检索

sql

CREATE TABLE post_embeddings (
    post_id    BIGINT NOT NULL,
    model      TEXT   NOT NULL DEFAULT 'all-MiniLM-L6-v2',
    embedding  vector(384) NOT NULL,
    updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    PRIMARY KEY (post_id, model)
);

CREATE INDEX idx_post_emb_hnsw
    ON post_embeddings USING hnsw (embedding vector_cosine_ops)
    WITH (m = 16, ef_construction = 64);

为什么单独一张表？

posts 是分区表，HNSW / IVFFlat 索引目前不支持分区表级别直接建；
embedding 通常异步生成，跟文章正文的生命周期解耦；
(post_id, model) 主键让同一篇文章可以存多版本模型向量，便于 A/B。

vector_cosine_ops 对应 <=> 余弦距离。参数：

m=16：每个节点连接的邻居数，越大召回越好、索引越大；
ef_construction=64：建索引时的候选列表大小，影响建索引速度和质量。

3.8 触发器：自动维护 `search_vector`

sql

CREATE OR REPLACE FUNCTION tg_posts_biu() RETURNS trigger AS $$
BEGIN
    NEW.search_vector :=
        setweight(to_tsvector('simple',
            unaccent(coalesce(NEW.title, ''))), 'A') ||
        setweight(to_tsvector('simple',
            unaccent(coalesce(NEW.body,  ''))), 'B');
    NEW.updated_at := now();
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER trg_posts_biu
BEFORE INSERT OR UPDATE OF title, body ON posts
FOR EACH ROW EXECUTE FUNCTION tg_posts_biu();

几个教学点（Ch12）：

setweight(...,'A'|'B'|'C'|'D') 控制权重，ts_rank 排序时 A 权重的 token 分更高。
simple 分词器对中文是按空格切，没有词干化；英文可用 'english' 享受 running → run。
中文全文检索的正确姿势是装 zhparser 或 pg_jieba，这里为了零额外依赖用 simple。
触发器指定 UPDATE OF title, body 只在这两列更新时触发——减少写放大（view_count++ 不会触发）。

3.9 物化视图 `mv_hot_posts`

sql

CREATE MATERIALIZED VIEW mv_hot_posts AS
SELECT
    p.id, p.tenant_id, p.title, p.author_id,
    p.view_count, p.like_count,
    COUNT(l.user_id) FILTER (WHERE l.created_at > now() - INTERVAL '7 days')
        AS recent_likes,
    (COUNT(l.user_id) FILTER (WHERE l.created_at > now() - INTERVAL '7 days') * 3
      + p.view_count / 100.0) AS hot_score,
    p.created_at
FROM posts p
LEFT JOIN post_likes l ON l.post_id = p.id
WHERE p.status = 'published'
GROUP BY p.id, p.tenant_id, p.title, p.author_id, p.view_count, p.like_count, p.created_at;

CREATE UNIQUE INDEX uq_mv_hot_posts_id ON mv_hot_posts (id);   -- CONCURRENTLY 前提
CREATE INDEX idx_mv_hot_score ON mv_hot_posts (tenant_id, hot_score DESC);

FILTER (WHERE ...) 是 PG 的聚合条件表达式，对 Ch5 中 CASE WHEN 写法的优雅替代。
物化视图是「预算好结果落盘」的缓存，适合读多写少的热门榜（Ch4）。
REFRESH CONCURRENTLY 不阻塞读，要求视图上有 唯一索引（Ch6）。

3.10 RLS 策略

sql

ALTER TABLE posts ENABLE ROW LEVEL SECURITY;

CREATE POLICY sel_tenant ON posts
    USING (tenant_id = current_setting('app.tenant_id', true)::bigint);

CREATE POLICY ins_tenant ON posts FOR INSERT
    WITH CHECK (tenant_id = current_setting('app.tenant_id', true)::bigint);

USING 控制 SELECT / UPDATE / DELETE 的可见性；
WITH CHECK 控制 INSERT / UPDATE 写入后是否满足策略；
current_setting('app.tenant_id', true) 读取连接 session 变量，第二个参数 true 表示「变量不存在返回 NULL 不报错」。

对应 Ch13。业务代码在借到连接后立即 SELECT set_config('app.tenant_id', 'X', true)，后续所有查询自动被策略加上隐式 WHERE tenant_id = X。

⚠️ 重要：表的 OWNER 和具有 BYPASSRLS 的超级用户绕过 RLS。所以业务连接必须用普通角色 blog_app，才能真正体验到策略限制。

4. 业务功能实现

4.1 注册登录（pgcrypto）

SQL 函数：

sql

CREATE OR REPLACE FUNCTION api_signup(p_tenant BIGINT, p_name TEXT,
                                      p_email TEXT, p_pwd TEXT)
RETURNS BIGINT AS $$
DECLARE new_id BIGINT;
BEGIN
    INSERT INTO users(tenant_id, username, email, password)
    VALUES (p_tenant, p_name, p_email, crypt(p_pwd, gen_salt('bf', 10)))
    RETURNING id INTO new_id;
    RETURN new_id;
END;
$$ LANGUAGE plpgsql;

gen_salt('bf', 10)：bcrypt 算法，10 次迭代（~100ms）；
crypt(pwd, salt)：返回 $2a$10$... 格式的 60 字符哈希；
校验时 SELECT 1 WHERE password = crypt(pwd, password)——bcrypt 把 salt 嵌在哈希里，恒定时间对比。

Python 调用：

python

# code/auth.py 节选
def signup(tenant_id, username, email, password):
    row = fetch_one(
        "SELECT api_signup(%s, %s, %s, %s) AS id",
        (tenant_id, username, email, password))
    return row["id"]

def login(tenant_id, username, password):
    row = fetch_one(
        "SELECT api_login(%s, %s, %s) AS id",
        (tenant_id, username, password))
    return row["id"] if row and row["id"] else None

4.2 发文 + 标签 + 自动 search_vector

python

# code/posts.py 核心
def publish(tenant_id, author_id, title, body, tags):
    row = fetch_one(
        "SELECT api_publish(%s,%s,%s,%s,%s) AS id",
        (tenant_id, author_id, title, body, tags))
    return row["id"]

插入完成后，触发器已自动生成 search_vector。验证：

sql

SELECT id, title, search_vector FROM posts ORDER BY id DESC LIMIT 1;

-- 输出片段：
-- search_vector | 'body':10B 'postgres':3A 'search':5B 'test':1A ...

注意 3A / 10B 后缀：A（title 权重）和 B（body 权重）——由 setweight 生成。

4.3 全文检索

python

# code/search_fulltext.py
def search(tenant_id, q, limit=10):
    sql = """
        SELECT id, title,
               ts_rank(search_vector, websearch_to_tsquery('simple', %(q)s)) AS rank,
               ts_headline('simple', body,
                           websearch_to_tsquery('simple', %(q)s),
                           'MaxWords=30, MinWords=10, StartSel=<<, StopSel=>>')
                   AS snippet
        FROM   posts
        WHERE  tenant_id = %(t)s AND status = 'published'
          AND  search_vector @@ websearch_to_tsquery('simple', %(q)s)
        ORDER  BY rank DESC
        LIMIT  %(l)s
    """
    return fetch_all(sql, {"q": q, "t": tenant_id, "l": limit})

几个要点：

websearch_to_tsquery 自动解析 "exact phrase"、-excluded、foo OR bar，比 plainto_tsquery 贴近 Google 风格。
ts_headline 直接生成带高亮的摘要，不需要业务层再切。
ORDER BY rank DESC 天然利用 GIN 索引的位图扫 + Sort。

4.4 模糊检索（pg_trgm）

python

# code/search_trgm.py
sql = """
    SELECT id, title, similarity(title, %(q)s) AS sim
    FROM   posts
    WHERE  tenant_id = %(t)s AND title %% %(q)s      -- '%' 要写成 %%
    ORDER  BY title <-> %(q)s
    LIMIT  %(l)s
"""

title % 'xxx'：相似度是否 ≥ pg_trgm.similarity_threshold（默认 0.3）；
title <-> 'xxx'：距离（1 - similarity），越小越相似，ORDER BY 时用；
走 GIN(title gin_trgm_ops)。

应用场景：用户拼错（psotgres 能查到 postgres），商品搜索、人名模糊。

4.5 语义检索（pgvector）

Python 端生成 embedding：

python

from sentence_transformers import SentenceTransformer

enc = SentenceTransformer("all-MiniLM-L6-v2")     # 输出 384 维
vec = enc.encode("如何做向量检索?", normalize_embeddings=True)
# → [0.021, -0.156, 0.339, ...]

入库：

sql

INSERT INTO post_embeddings(post_id, model, embedding)
VALUES (%s, 'all-MiniLM-L6-v2', %s::vector)
ON CONFLICT (post_id, model) DO UPDATE
SET embedding = EXCLUDED.embedding, updated_at = now();

检索：

sql

SELECT p.id, p.title,
       (e.embedding <=> %s::vector) AS distance
FROM   post_embeddings e
JOIN   posts p ON p.id = e.post_id
WHERE  p.tenant_id = %s AND e.model = 'all-MiniLM-L6-v2'
ORDER  BY e.embedding <=> %s::vector
LIMIT  5;

三种距离操作符（Ch18）：

操作符	含义	配套索引 op class
`<->`	L2 欧式距离	`vector_l2_ops`
`<=>`	余弦距离	`vector_cosine_ops`
`<#>`	负内积	`vector_ip_ops`

为什么用余弦？ 因为 sentence-transformers 的默认规范化向量模长 ≈ 1，余弦距离最稳。

4.6 评论树（递归 CTE）

sql

WITH RECURSIVE tree AS (
    SELECT c.*, 1 AS depth, ARRAY[c.id] AS path
    FROM   comments c
    WHERE  c.post_id = :p AND c.parent_id IS NULL   -- 锚点：所有根评论

    UNION ALL

    SELECT c.*, t.depth + 1, t.path || c.id
    FROM   comments c JOIN tree t ON c.parent_id = t.id   -- 递归展开
)
SELECT id, parent_id, depth, path, content
FROM   tree
ORDER  BY path;          -- 深度优先，父在前

path ARRAY[c.id] 的妙用：

按 path 排序就是深度优先遍历；
前端渲染时 indent = depth * 16px 直接出树状 UI。

4.7 JSONB GIN 按标签筛选

sql

SELECT id, title FROM posts
WHERE tenant_id = 1
  AND tags @> '["postgres"]'::jsonb
ORDER BY created_at DESC;

走索引（第 6 节会给 EXPLAIN 证据）：

Bitmap Heap Scan on posts_2025_01 posts
  Recheck Cond: (tags @> '["postgres"]'::jsonb)
  ->  Bitmap Index Scan on posts_2025_01_tags_idx
        Index Cond: (tags @> '["postgres"]'::jsonb)

4.8 月度归档（分区表 DETACH）

业务场景：想把 2 年前的文章归档到便宜的冷存储。

sql

-- 1) 把老分区摘下来，变成独立表
ALTER TABLE posts DETACH PARTITION posts_2025_01;

-- 2) 搬到冷 schema（或 pg_dump 导出）
ALTER TABLE posts_2025_01 SET SCHEMA archive;

-- 3) 需要重新挂回来时
ALTER TABLE posts ATTACH PARTITION archive.posts_2025_01
    FOR VALUES FROM ('2025-01-01') TO ('2025-02-01');

DETACH 是元数据变更，几乎秒级完成，不用搬数据。

4.9 热门榜物化视图刷新

python

# code/refresh_hot.py
def refresh(concurrent=True):
    sql = ("REFRESH MATERIALIZED VIEW CONCURRENTLY blog.mv_hot_posts"
           if concurrent
           else "REFRESH MATERIALIZED VIEW blog.mv_hot_posts")
    execute(sql)

生产上用 pg_cron 扩展（Ch18）每小时自动跑：

sql

SELECT cron.schedule(
    'refresh_hot', '0 * * * *',
    'REFRESH MATERIALIZED VIEW CONCURRENTLY blog.mv_hot_posts');

4.10 多租户隔离验证

python

# code/multi_tenant_demo.py 片段
with psycopg.connect(APP_DSN) as conn, conn.cursor() as cur:
    cur.execute("SELECT set_config('app.tenant_id', '1', false)")
    cur.execute("SELECT COUNT(*) FROM posts")        # → 48（只看到 tenant 1）
    cur.execute("SELECT set_config('app.tenant_id', '2', false)")
    cur.execute("SELECT COUNT(*) FROM posts")        # → 2  （只看到 tenant 2）

尝试「越权插入」会触发 WITH CHECK 违规：

psycopg.errors.CheckViolation: new row violates row-level security policy
    "ins_tenant" for table "posts"

5. 性能调优实战

本节演示从慢到快的调优闭环。数据是 seed.sql 生成的 50 文章 + 填充数据。想看更明显的效果，可以循环灌到 100 万行。

5.1 制造一个慢查询

先关掉索引，跑一次：

sql

SET enable_indexscan = off;
SET enable_bitmapscan = off;

EXPLAIN (ANALYZE, BUFFERS)
SELECT id, title
FROM   posts
WHERE  search_vector @@ websearch_to_tsquery('simple', 'postgres')
ORDER  BY ts_rank(search_vector, websearch_to_tsquery('simple', 'postgres')) DESC
LIMIT  10;

输出（节选）：

Limit  (cost=... rows=10 width=...) (actual time=15.421..15.438 rows=10 loops=1)
  Buffers: shared hit=1024 read=512
  ->  Sort (actual time=15.419..15.430 rows=10 loops=1)
        Sort Key: (ts_rank(...)) DESC
        Sort Method: top-N heapsort  Memory: 28kB
        ->  Append (actual time=0.021..14.998 rows=62 loops=1)
              ->  Seq Scan on posts_2025_01 posts_1
                    Filter: (search_vector @@ '...'::tsquery)
                    Rows Removed by Filter: 38
              ->  Seq Scan on posts_2025_02 posts_2
                    Filter: (search_vector @@ '...'::tsquery)
                    Rows Removed by Filter: 41
              ...
Planning Time: 0.812 ms
Execution Time: 15.527 ms

问题：每个月分区都做 Seq Scan，哪怕命中率很低也要全表过一遍。

5.2 `pg_stat_statements` 找 Top 慢 SQL

sql

SELECT pg_stat_statements_reset();
-- 跑一批业务 SQL ...

SELECT substring(query, 1, 80) AS q, calls,
       round(total_exec_time::numeric, 2) AS total_ms,
       round(mean_exec_time::numeric,  2) AS mean_ms
FROM   pg_stat_statements
ORDER  BY total_exec_time DESC LIMIT 5;

输出示例：

 q                                                                        | calls | total_ms | mean_ms
--------------------------------------------------------------------------+-------+----------+---------
 SELECT COUNT(*) FROM posts WHERE lower(title) LIKE $1                    |    20 |   340.15 |   17.01
 SELECT id, title FROM posts WHERE search_vector @@ to_tsquery($1,$2) ... |    15 |   210.89 |   14.06
 SELECT id,title FROM posts WHERE title %% $1 ORDER BY title <-> $1 ...   |    20 |   180.33 |    9.02

发现：lower(title) LIKE '%pg%' 这种老派写法最慢，因为对 title 做了函数运算，GIN 索引用不上。

5.3 加索引后对比

打开索引再跑同一查询：

sql

RESET enable_indexscan; RESET enable_bitmapscan;

EXPLAIN (ANALYZE, BUFFERS)
SELECT id, title FROM posts
WHERE  search_vector @@ websearch_to_tsquery('simple','postgres')
ORDER  BY ts_rank(search_vector, websearch_to_tsquery('simple','postgres')) DESC
LIMIT  10;

输出：

Limit  (actual time=0.312..0.328 rows=10 loops=1)
  Buffers: shared hit=40
  ->  Sort (actual time=0.310..0.318 rows=10 loops=1)
        Sort Method: top-N heapsort  Memory: 28kB
        ->  Append
              ->  Bitmap Heap Scan on posts_2025_01 posts_1
                    Recheck Cond: (search_vector @@ '...'::tsquery)
                    ->  Bitmap Index Scan on posts_2025_01_search_vector_idx
                          Index Cond: (search_vector @@ '...'::tsquery)
              ->  Bitmap Heap Scan on posts_2025_02 posts_2
                    ...
Planning Time: 1.203 ms
Execution Time: 0.389 ms

提升 40x（15.5 ms → 0.39 ms）。解读：

走到了 Bitmap Index Scan，命中 posts_XX_search_vector_idx（父表 idx_posts_fts 的子分区自动下推）；
Bitmap Heap Scan + Recheck Cond：先用索引拿页号位图，再逐页读实际行（Ch6）；
Buffers: shared hit=40：全部命中共享缓冲区，没走磁盘。

5.4 JSONB 标签：jsonb_path_ops 更小

对比两种 JSONB GIN op class 的大小：

sql

SELECT pg_size_pretty(pg_relation_size('idx_posts_tags')) AS path_ops;
-- → 40 kB

-- 临时改成默认 jsonb_ops 建一个对照组
CREATE INDEX idx_posts_tags_full ON posts USING GIN (tags);
SELECT pg_size_pretty(pg_relation_size('idx_posts_tags_full'));
-- → 72 kB

jsonb_path_ops 只支持 @> 一个操作符，但索引尺寸减半、构建更快——业务只用 @> 时首选。

5.5 语义检索：HNSW 参数调优

sql

-- 运行时调优
SET hnsw.ef_search = 40;    -- 检索时候选列表大小，越大越准越慢

EXPLAIN (ANALYZE, BUFFERS)
SELECT post_id
FROM   post_embeddings
ORDER  BY embedding <=> '[...]'::vector
LIMIT  5;

可能输出：

Limit  (actual time=0.842..0.855 rows=5 loops=1)
  ->  Index Scan using idx_post_emb_hnsw on post_embeddings
        Order By: (embedding <=> '[...]'::vector)

如果看到 Seq Scan 而不是 Index Scan using idx_post_emb_hnsw，检查：

向量维度是否匹配（我们建的是 vector(384)，查询向量必须也是 384）；
距离操作符是否匹配（索引用的 vector_cosine_ops 只能加速 <=>，不能加速 <->）。

5.6 auto_explain：自动记录慢查询

conf

# postgresql.conf 片段
shared_preload_libraries = 'pg_stat_statements,auto_explain'
auto_explain.log_min_duration = '100ms'
auto_explain.log_analyze = on
auto_explain.log_buffers = on
auto_explain.log_nested_statements = on

之后所有 >100 ms 的 SQL 会把 EXPLAIN (ANALYZE, BUFFERS) 直接写进 PG 日志。调试非常香。

6. 部署清单

6.1 Docker Compose（最简方式）

yaml

services:
  pg:
    image: pgvector/pgvector:pg16
    environment:
      POSTGRES_DB: learn_pg
      POSTGRES_USER: postgres
      POSTGRES_PASSWORD: postgres
    ports: ["5432:5432"]
    volumes:
      - pg_data:/var/lib/postgresql/data
      - ./init.sql:/docker-entrypoint-initdb.d/10_init.sql:ro
      - ./seed.sql:/docker-entrypoint-initdb.d/20_seed.sql:ro
    command:
      - "postgres"
      - "-c"
      - "shared_preload_libraries=pg_stat_statements"
      - "-c"
      - "pg_stat_statements.track=all"
      - "-c"
      - "log_min_duration_statement=200"
volumes:
  pg_data:

一行启动：

bash

docker compose up -d
docker compose logs -f pg | grep -m1 'ready to accept'
pip install -r requirements.txt
python code/api.py

6.2 必装扩展清单

扩展	用途	安装
`pgcrypto`	bcrypt 密码 + UUID	官方自带
`pg_trgm`	模糊搜索	官方自带
`vector`	语义检索	`pgvector` 镜像自带
`pg_stat_statements`	慢查询统计	官方自带，需 preload
`unaccent`	去重音	官方自带
`btree_gin`	普通列入 GIN	官方自带
`pg_cron`（可选）	定时任务	`apt install postgresql-16-cron`
`pg_partman`（可选）	分区自动维护	`apt install postgresql-16-partman`

7. 本项目映射的 PG 知识点一览

证明「学完了能用」：覆盖前 18 章每一章。

章节	知识点	在本项目的落地
Ch1 PG 是什么	对象-关系、扩展生态	整个项目用了 6 个扩展
Ch2 psql / 三级结构	Schema / psql 元命令	`blog` schema、`\dx`、`\d+ posts`
Ch3 数据类型	`JSONB` / `timestamptz` / `text` / `tsvector` / `vector`	`posts.tags` / `created_at` / `search_vector` / `embedding`
Ch4 约束/视图/序列	`IDENTITY` / `CHECK` / UNIQUE / 物化视图	`users.id` / `role` / `(tenant_id,username)` / `mv_hot_posts`
Ch5 高级查询	CTE / 递归 CTE / 窗口 / `FILTER` / `ON CONFLICT`	评论树 / 点赞 upsert / 物化视图聚合
Ch6 索引	B-Tree / GIN / `jsonb_path_ops` / `gin_trgm_ops` / HNSW	5 个父表索引 + 1 个 HNSW
Ch7 事务与隔离	读提交默认、`FOR UPDATE`	发文函数 / 点赞 `ON CONFLICT`
Ch8 MVCC / VACUUM	`view_count++` 的死元组	大量 UPDATE 触发 HOT 与 autovacuum
Ch9 存储 / TOAST	TEXT 字段自动 TOAST	`body` 长文自动压缩分片
Ch10 WAL / Checkpoint	所有 INSERT 都先写 WAL	`log_min_duration_statement` 可以观测
Ch11 锁	DDL `ALTER TABLE ATTACH PARTITION`	月度归档时要注意锁
Ch12 服务端编程	PL/pgSQL 函数 / 触发器	`api_signup` / `tg_posts_biu`
Ch13 权限 / RLS	RLS + `current_setting` + 业务角色	多租户隔离
Ch14 备份恢复	`pg_dump -Fc`	定期全备 `docker compose exec pg pg_dump ...`
Ch15 复制	逻辑复制 publish	扩展章节（下面第 8 节）
Ch16 分区	`PARTITION BY RANGE (created_at)`	`posts_2025_01` ~ `posts_2026_04` + 默认
Ch17 性能调优	`pg_stat_statements` + `EXPLAIN ANALYZE`	`code/perf_audit.py`
Ch18 扩展	pgcrypto / pg_trgm / pgvector / pg_cron	全部用到

8. 进一步可做的扩展

让这个项目更接近生产形态：

主从流复制
- docker compose 再起一个 pg-standby 做热备；
- 业务读用 pgBouncer 做读写分离。
pgBouncer 连接池
- 放在 PG 前面，pool_mode = transaction，对百级连接聚合成十级。
监控接入
- pg_exporter + Prometheus + Grafana；
- 重点监控 pg_stat_activity.state、pg_stat_bgwriter、pg_stat_replication。
PITR 定期演练
- pg_basebackup + WAL 归档 → recovery_target_time 恢复到误删前 1 秒。
中文全文检索
- 换 zhparser，把 to_tsvector('simple', ...) 改成 to_tsvector('zhcfg', ...)。
向量量化
- pgvector 0.7+ 支持 halfvec（2 字节）与 binary vector（1 bit）—— 2 亿向量内存从 300 GB 降到 50 GB。
逻辑复制零停机升级
- PG 15 → 16：建立 publication → 新库 subscribe → 追平 → 切流量。
LLM 检索增强（RAG）
- 取 top-k 向量结果 → 注入 Prompt → 交给 GPT 生成答案。

9. 面试高频题

Q1：为什么 PG 适合做「全文 + 向量 + JSONB」混合搜索？与 ES / Milvus 组合相比有什么取舍？

考察点：对 PG 特色能力的系统理解。

答：

一致性 ：PG 单机内事务保证「文章 + 向量 + 标签」一起写成功或一起失败，免去分布式最终一致性难题。
运维简单：只维护一套存储 / 备份 / 复制。
查询组合灵活：可以 WHERE search_vector @@ q AND tags @> '[...]' ORDER BY embedding <=> qvec，三种检索条件一起打。
但有 trade-off：向量规模 5000 万 + 就要考虑 Milvus / Qdrant；全文中文分词仍需外挂（如 zhparser）。

Q2：分区表上如何正确地建索引、建唯一约束？

在 PG 11+，父表上声明的 CREATE INDEX 会自动下推到所有子分区，包括未来新建的；
唯一约束必须包含分区键（如本项目 PRIMARY KEY (id, created_at)）；
如果业务要求 id 全局唯一，PG 支持在父表上建 UNIQUE (id) 吗？—— 不支持，只能在每个子分区内唯一，必要时配合全局序列 + 业务层校验。

Q3：`REFRESH MATERIALIZED VIEW` 和 `REFRESH ... CONCURRENTLY` 的区别？

前者锁 AccessExclusiveLock，刷新期间读也被阻塞；
后者用临时表 + DIFF 替换，不阻塞读；
代价：必须有唯一索引、执行时间更长（多一次扫全表）。

Q4：多租户的 3 种常见做法？RLS 的利弊？

方案	做法	优点	缺点
每租户独立库	N 个 Database	隔离最强	连接数爆炸 / 升级麻烦
每租户独立 Schema	同库 N schema	迁移灵活	数据库端对象数膨胀
共享表 + RLS	一张表 + `tenant_id` 过滤	资源复用	策略错一个字全泄漏

RLS 利：零侵入业务代码；弊：必须严守 BYPASSRLS 权限，且 EXPLAIN 时要看到策略已下推。

Q5：`EXPLAIN` 里看到 `Bitmap Heap Scan` 但性能依然差，排查思路？

Recheck Cond 太多 → 索引选择性差，看 Rows Removed by Filter；
Heap Fetches 很高 → visibility map 过期，VACUUM 让索引覆盖扫描生效；
Buffers: read=xxx（read 高而 hit 低）→ shared_buffers 偏小或数据冷；
work_mem 不足 → Sort 走 external merge，log_temp_files 可捕获。

Q6：本项目里 `view_count` 的频繁 UPDATE 会导致什么问题？怎么缓解？

问题：每次 view_count++ 在 MVCC 下产生一个新元组版本，老版本成死元组。高并发下死元组膨胀、索引页膨胀，查询变慢。
缓解：
1. HOT 更新条件：新版本留在同一页 + 被更新的列不被任何索引引用。所以不要在 view_count 上建索引。
2. 批量聚合：Redis 做计数累加，每分钟 UPDATE posts SET view_count = view_count + delta 一次。
3. 分区表：让旧月份表不被更新，autovacuum 更轻。
4. fillfactor=70：给 HOT 更新留空间。

Q7：pgvector 的 IVFFlat 和 HNSW 怎么选？

维度	IVFFlat	HNSW
建索引速度	快	慢（3-10×）
检索速度	中	快
召回率	中（取决于 `probes`）	高
内存	小	大（~2-3 倍原向量）
更新友好	差（需重训）	好（增量插入）
推荐场景	冷数据 / 写少	在线检索 / 增量写

本项目选 HNSW 因为博客场景写入连续。

Q8：如何做这个项目的灰度升级（PG 14 → 16）？

旧库装 pglogical 或 PG 内置逻辑复制，CREATE PUBLICATION 发布 blog.*；
新 PG 16 库建同结构，CREATE SUBSCRIPTION 订阅；
等初始同步完成 + 增量追平；
业务流量分批切读，最后切写；
切完后 DROP SUBSCRIPTION、老库下线。

10. 本章小结

🎉 恭喜你走完了「从 0 到 1 学习 PostgreSQL」全部 19 章。

回头看，你已经：

会写 JSONB + 数组 + JOIN + CTE + 窗口 + upsert 的 PG 风味 SQL；
看得懂 MVCC、WAL、Checkpoint 的生活化比喻与源码思想；
选得对 B-Tree / GIN / GiST / BRIN / HNSW 索引；
诊得出 慢查询 / 膨胀 / 锁等待；
搭得起 分区表 / 物化视图 / RLS 多租户 / 主从 / PITR；
接得上 pgvector RAG / pg_trgm 模糊 / 全文检索 三种检索栈；
面得了 字节 / 阿里 / 腾讯 的 DBA / 后端 PG 高频题。

下一步建议：

把这个项目灌到 1000 万行数据，亲自跑一遍「慢查询 → 加索引 → 验证」的调优闭环；
给它接入 主从流复制 + pgBouncer，体验一下生产拓扑；
订阅 Postgres Weekly，跟着版本走——PG 17 的 增量备份、MERGE ... RETURNING 都非常香。

最后一句话：PG 的魅力在于「标准 SQL 的优雅 + 对象关系扩展的灵动」，从此你不只是一个写 CRUD 的工程师，而是一个能用数据库解决业务问题的架构师。祝你早日成为团队里的 🐘 PG 专家。

🎬 可视化演示

演示加载缓慢或样式异常？点此在新标签页打开 ↗

💻 示例代码

python

"""FastAPI REST 接口：把上面所有能力暴露成 HTTP

运行：
    pip install "fastapi[standard]" "psycopg[binary]" psycopg_pool
    uvicorn code.api:app --host 0.0.0.0 --port 8000 --reload
    # 或直接：python code/api.py

浏览器打开 http://localhost:8000/docs 可看到 Swagger UI。
"""
from __future__ import annotations

import sys
from typing import Optional

try:
    from fastapi import FastAPI, HTTPException, Query
    from pydantic import BaseModel
except ImportError:
    print("[ERROR] 未安装 fastapi，请 pip install 'fastapi[standard]'")
    sys.exit(1)

import auth
import posts as posts_mod
import search_fulltext
import search_trgm
import search_vector
import comments_tree
import refresh_hot


app = FastAPI(title="PG 综合实战 · 博客 API",
              description="第 19 章产物：一个 REST 接口把所有 PG 能力串起来",
              version="1.0.0")


# ------------------------- 数据模型 -------------------------
class SignupIn(BaseModel):
    tenant_id: int = 1
    username: str
    email: str
    password: str


class LoginIn(BaseModel):
    tenant_id: int = 1
    username: str
    password: str


class PublishIn(BaseModel):
    tenant_id: int = 1
    author_id: int
    title: str
    body: str
    tags: list[str] = []


# ------------------------- 健康检查 -------------------------
@app.get("/health")
def health() -> dict:
    from db import fetch_one
    return fetch_one("SELECT version() AS pg, now() AS server_time") or {}


# -------------------------- Auth --------------------------
@app.post("/signup")
def signup(body: SignupIn) -> dict:
    try:
        uid = auth.signup(body.tenant_id, body.username, body.email, body.password)
        return {"user_id": uid}
    except Exception as e:
        raise HTTPException(400, str(e))


@app.post("/login")
def login(body: LoginIn) -> dict:
    uid = auth.login(body.tenant_id, body.username, body.password)
    if not uid:
        raise HTTPException(401, "用户名或密码错误")
    return {"user_id": uid}


# -------------------------- Posts --------------------------
@app.post("/posts")
def publish(body: PublishIn) -> dict:
    pid = posts_mod.publish(body.tenant_id, body.author_id,
                            body.title, body.body, body.tags)
    return {"post_id": pid}


@app.get("/posts/recent")
def recent(tenant_id: int = 1, limit: int = 10) -> list[dict]:
    return posts_mod.list_recent(tenant_id, limit)


@app.get("/posts/by_tag")
def by_tag(tenant_id: int = 1, tag: str = "postgres") -> list[dict]:
    return posts_mod.filter_by_tag(tenant_id, tag)


# -------------------------- Search --------------------------
@app.get("/search/fulltext")
def s_fulltext(q: str = Query(..., min_length=1), tenant_id: int = 1,
               limit: int = 10) -> list[dict]:
    return search_fulltext.search(tenant_id, q, limit)


@app.get("/search/trgm")
def s_trgm(q: str = Query(..., min_length=1), tenant_id: int = 1,
           limit: int = 10) -> list[dict]:
    return search_trgm.search(tenant_id, q, limit)


@app.get("/search/vector")
def s_vector(q: str = Query(..., min_length=1), tenant_id: int = 1,
             limit: int = 5) -> list[dict]:
    return search_vector.search(tenant_id, q, limit)


@app.post("/search/vector/reindex")
def s_vector_reindex(tenant_id: int = 1) -> dict:
    n = search_vector.build_index_for_all(tenant_id)
    return {"new_embeddings": n}


# -------------------------- Comments --------------------------
@app.get("/comments/tree")
def comment_tree(post_id: int) -> list[dict]:
    return comments_tree.build_tree(post_id)


# -------------------------- Hot --------------------------
@app.get("/hot")
def hot(tenant_id: int = 1, n: int = 10) -> list[dict]:
    return refresh_hot.top_n(tenant_id, n)


@app.post("/hot/refresh")
def hot_refresh() -> dict:
    cost = refresh_hot.refresh(concurrent=True)
    return {"cost_sec": round(cost, 3)}


# -------------------------- main --------------------------
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

python

"""注册 / 登录 —— 通过数据库侧 pgcrypto 完成 bcrypt 哈希

运行：python code/auth.py
"""
from __future__ import annotations

from db import fetch_one, close_pool


def signup(tenant_id: int, username: str, email: str, password: str) -> int:
    """注册新用户，返回 user_id。"""
    row = fetch_one(
        "SELECT api_signup(%s, %s, %s, %s) AS id",
        (tenant_id, username, email, password),
    )
    return row["id"]


def login(tenant_id: int, username: str, password: str) -> int | None:
    """登录成功返回 user_id，失败返回 None。"""
    row = fetch_one(
        "SELECT api_login(%s, %s, %s) AS id",
        (tenant_id, username, password),
    )
    return row["id"] if row and row["id"] else None


def check_password_hash_format(user_id: int) -> str:
    """演示：查看密码哈希是不是 bcrypt 格式（$2a$...）。"""
    row = fetch_one("SELECT password FROM users WHERE id = %s", (user_id,))
    return row["password"] if row else ""


def _demo() -> None:
    print("=== 注册登录演示 ===")

    try:
        uid = signup(1, "demo_user", "demo@x.com", "P@ssw0rd!")
        print(f"[OK] 新注册用户 id={uid}")
    except Exception as e:
        print(f"[INFO] 用户已存在，跳过注册：{e}")
        uid = fetch_one("SELECT id FROM users WHERE username = 'demo_user'")["id"]

    print("密码哈希：", check_password_hash_format(uid))

    ok = login(1, "demo_user", "P@ssw0rd!")
    print("正确密码登录：", "成功 uid=" + str(ok) if ok else "失败")

    fail = login(1, "demo_user", "wrong")
    print("错误密码登录：", "成功" if fail else "失败（预期）")


if __name__ == "__main__":
    try:
        _demo()
    finally:
        close_pool()

python

"""评论树：递归 CTE

一次 SQL 查完一篇文章的所有评论 + 层级 + 路径，无需 N+1。

运行：python code/comments_tree.py [post_id]
"""
from __future__ import annotations

import sys

from db import fetch_all, close_pool


TREE_SQL = """
WITH RECURSIVE tree AS (
    SELECT c.id,
           c.post_id,
           c.parent_id,
           c.author_id,
           c.content,
           c.created_at,
           1                                     AS depth,
           ARRAY[c.id]                           AS path
    FROM   comments c
    WHERE  c.post_id = %(p)s
      AND  c.parent_id IS NULL

    UNION ALL

    SELECT c.id, c.post_id, c.parent_id, c.author_id,
           c.content, c.created_at,
           t.depth + 1,
           t.path || c.id
    FROM   comments c
    JOIN   tree t ON c.parent_id = t.id
)
SELECT id, parent_id, author_id, depth, path, content
FROM   tree
ORDER  BY path;                                  -- 深度优先、父在前
"""


def build_tree(post_id: int) -> list[dict]:
    return fetch_all(TREE_SQL, {"p": post_id})


def pretty_print(post_id: int) -> None:
    rows = build_tree(post_id)
    if not rows:
        print(f"文章 {post_id} 没有评论")
        return

    print(f"=== 文章 {post_id} 的评论树（{len(rows)} 条） ===")
    for r in rows:
        indent = "    " * (r["depth"] - 1)
        print(f"{indent}└─ [#{r['id']} by u{r['author_id']}] "
              f"{r['content'][:50]}")


def explain(post_id: int) -> list[dict]:
    sql = "EXPLAIN (ANALYZE, BUFFERS, FORMAT TEXT) " + TREE_SQL
    return fetch_all(sql, {"p": post_id})


if __name__ == "__main__":
    pid = int(sys.argv[1]) if len(sys.argv) > 1 else 1
    try:
        pretty_print(pid)
        print("\n-- EXPLAIN --")
        for r in explain(pid):
            print(" ", r["QUERY PLAN"])
    finally:
        close_pool()

python

"""公共数据库连接池

依赖：psycopg[binary]>=3, psycopg_pool>=3
说明：
    1. 用 psycopg v3（异步/同步双栈 + 更好的连接池）
    2. DSN 通过环境变量 PG_DSN 覆盖，默认连 docker-compose 里的 pg
    3. 提供 with_tenant() 上下文管理器：在连接上设置 app.tenant_id 供 RLS 使用
"""
from __future__ import annotations

import os
from contextlib import contextmanager
from typing import Iterator, Optional

import psycopg
from psycopg.rows import dict_row
from psycopg_pool import ConnectionPool

DSN = os.environ.get(
    "PG_DSN",
    "host=127.0.0.1 port=5432 dbname=learn_pg user=postgres password=postgres",
)

# 全局连接池：最多 10 个连接，idle 超过 30s 回收
_pool: Optional[ConnectionPool] = None


def get_pool() -> ConnectionPool:
    """懒加载全局连接池。"""
    global _pool
    if _pool is None:
        _pool = ConnectionPool(
            conninfo=DSN,
            min_size=1,
            max_size=10,
            kwargs={"row_factory": dict_row, "options": "-c search_path=blog,public"},
            open=True,
        )
    return _pool


@contextmanager
def get_conn(tenant_id: Optional[int] = None) -> Iterator[psycopg.Connection]:
    """借一个连接。若传 tenant_id 则自动注入 app.tenant_id，供 RLS 策略读取。"""
    pool = get_pool()
    with pool.connection() as conn:
        if tenant_id is not None:
            with conn.cursor() as cur:
                cur.execute("SELECT set_config('app.tenant_id', %s, true)",
                            (str(tenant_id),))
        yield conn


def fetch_all(sql: str, params: tuple | list | dict | None = None,
              tenant_id: Optional[int] = None) -> list[dict]:
    with get_conn(tenant_id) as conn, conn.cursor() as cur:
        cur.execute(sql, params or ())
        return cur.fetchall()


def fetch_one(sql: str, params: tuple | list | dict | None = None,
              tenant_id: Optional[int] = None) -> dict | None:
    with get_conn(tenant_id) as conn, conn.cursor() as cur:
        cur.execute(sql, params or ())
        return cur.fetchone()


def execute(sql: str, params: tuple | list | dict | None = None,
            tenant_id: Optional[int] = None) -> int:
    """返回受影响行数。"""
    with get_conn(tenant_id) as conn, conn.cursor() as cur:
        cur.execute(sql, params or ())
        return cur.rowcount


def close_pool() -> None:
    global _pool
    if _pool is not None:
        _pool.close()
        _pool = None


if __name__ == "__main__":
    # 自检：能连上就打印版本
    row = fetch_one("SELECT version() AS v, current_database() AS db")
    print("连接成功:", row)
    close_pool()

python

"""RLS 多租户验证

展示：
    - 以 postgres 超级用户（BYPASSRLS）能看到所有行
    - 以 blog_app 普通角色 + set_config('app.tenant_id', ...) 只能看到自己租户的行
    - 跨租户 INSERT 会被策略的 WITH CHECK 拒绝

运行：python code/multi_tenant_demo.py
"""
from __future__ import annotations

import os
import psycopg
from psycopg.rows import dict_row

from db import close_pool

SUPER_DSN = os.environ.get(
    "PG_DSN",
    "host=127.0.0.1 port=5432 dbname=learn_pg user=postgres password=postgres",
)

# init.sql 创建的业务角色
APP_DSN = (SUPER_DSN
           .replace("user=postgres", "user=blog_app")
           .replace("password=postgres", "password=blog_app_pwd"))


def count_as(dsn: str, tenant_id: int | None) -> int:
    with psycopg.connect(dsn, row_factory=dict_row,
                         options="-c search_path=blog,public") as conn:
        with conn.cursor() as cur:
            if tenant_id is not None:
                cur.execute("SELECT set_config('app.tenant_id', %s, false)",
                            (str(tenant_id),))
            cur.execute("SELECT COUNT(*) AS c FROM posts")
            return cur.fetchone()["c"]


def try_cross_tenant_insert(dsn: str, my_tenant: int, evil_tenant: int) -> str:
    """尝试以 my_tenant 身份写入一条 evil_tenant 的数据，应被 RLS 拒绝。"""
    try:
        with psycopg.connect(dsn, row_factory=dict_row,
                             options="-c search_path=blog,public") as conn:
            with conn.cursor() as cur:
                cur.execute("SELECT set_config('app.tenant_id', %s, false)",
                            (str(my_tenant),))
                cur.execute(
                    """INSERT INTO posts(tenant_id, author_id, title, body,
                                         status)
                       VALUES (%s, 1, '跨租户攻击', 'bad', 'published')""",
                    (evil_tenant,),
                )
                conn.commit()
        return "[!!!] 居然成功了（说明 RLS 没生效！）"
    except psycopg.errors.InsufficientPrivilege as e:
        return f"[OK] 被 WITH CHECK 拒绝：{e}"
    except Exception as e:
        return f"[OK] 被拒绝：{type(e).__name__}: {e}"


def main() -> None:
    print("=== RLS 多租户演示 ===\n")

    print("-- 1) 超级用户 postgres（默认 BYPASSRLS，不受策略限制） --")
    all_cnt = count_as(SUPER_DSN, tenant_id=1)
    print(f"   postgres 看到 posts 总数 = {all_cnt}\n")

    print("-- 2) 普通角色 blog_app，tenant=1 --")
    try:
        t1 = count_as(APP_DSN, tenant_id=1)
        print(f"   tenant=1 视角下 posts 数 = {t1}")
    except Exception as e:
        print(f"   [WARN] 无法连 blog_app：{e}")
        print("   （确保 init.sql 执行过，且 pg_hba.conf 允许密码登录）")
        return

    print("-- 3) 普通角色 blog_app，tenant=2 --")
    t2 = count_as(APP_DSN, tenant_id=2)
    print(f"   tenant=2 视角下 posts 数 = {t2}")
    print("   （应该只看到 seed.sql 里属于 tenant=2 的少量文章）\n")

    print("-- 4) 尝试以 tenant=1 的身份写入 tenant=2 的数据 --")
    print(" ", try_cross_tenant_insert(APP_DSN, my_tenant=1, evil_tenant=2))


if __name__ == "__main__":
    try:
        main()
    finally:
        close_pool()

python

"""慢查询审计：pg_stat_statements + EXPLAIN

流程：
    1. 确认 pg_stat_statements 已 CREATE EXTENSION 且加入 shared_preload_libraries
    2. 重置统计 → 跑一批业务 SQL → 打印 Top N
    3. 对 Top 1 自动 EXPLAIN ANALYZE

运行：python code/perf_audit.py
"""
from __future__ import annotations

from db import fetch_all, fetch_one, execute, close_pool


def check_extension() -> bool:
    row = fetch_one(
        "SELECT 1 AS ok FROM pg_extension WHERE extname = 'pg_stat_statements'")
    return bool(row)


def reset() -> None:
    execute("SELECT pg_stat_statements_reset()")


def run_sample_workload() -> None:
    """故意跑一些不同耗时的 SQL 以填充 pg_stat_statements。"""
    sqls = [
        "SELECT COUNT(*) FROM posts WHERE status = 'published'",
        "SELECT * FROM posts WHERE tags @> '[\"postgres\"]' ORDER BY created_at DESC LIMIT 10",
        "SELECT id, title FROM posts WHERE search_vector @@ to_tsquery('simple','postgres') LIMIT 10",
        "SELECT id, title FROM posts WHERE title %% 'psotgres' ORDER BY title <-> 'psotgres' LIMIT 10",
        # 故意不走索引
        "SELECT COUNT(*) FROM posts WHERE lower(title) LIKE '%pg%'",
        "SELECT id, title FROM mv_hot_posts ORDER BY hot_score DESC LIMIT 10",
    ]
    for s in sqls:
        fetch_all(s)


def top_n(n: int = 5) -> list[dict]:
    # PG 13+：列名是 total_exec_time；12 及以前是 total_time，这里做兼容
    col_exists = fetch_one(
        """SELECT 1 AS ok FROM information_schema.columns
           WHERE table_name='pg_stat_statements' AND column_name='total_exec_time'""")
    total_col = "total_exec_time" if col_exists else "total_time"
    mean_col  = "mean_exec_time"  if col_exists else "mean_time"

    sql = f"""
        SELECT substring(query, 1, 80) AS query,
               calls,
               round({total_col}::numeric, 2) AS total_ms,
               round({mean_col}::numeric, 2)  AS mean_ms,
               rows
        FROM   pg_stat_statements
        WHERE  query NOT LIKE 'EXPLAIN%'
          AND  query NOT LIKE '%pg_stat_statements%'
          AND  dbid = (SELECT oid FROM pg_database WHERE datname = current_database())
        ORDER  BY {total_col} DESC
        LIMIT  %s
    """
    return fetch_all(sql, (n,))


def main() -> None:
    if not check_extension():
        print("[ERROR] pg_stat_statements 未安装，请先：")
        print("  1) 在 postgresql.conf 里 shared_preload_libraries = 'pg_stat_statements'")
        print("  2) 重启 PG")
        print("  3) psql: CREATE EXTENSION pg_stat_statements;")
        return

    print("=== pg_stat_statements 审计 ===")
    print("[step 1] 重置统计")
    reset()

    print("[step 2] 跑样例业务 SQL")
    run_sample_workload()

    print("\n[step 3] Top 5 最耗时 SQL")
    rows = top_n(5)
    if not rows:
        print("  （没抓到，可能 track=none 或刚刚重置）")
        return

    print(f"  {'query':<80}  {'calls':>5}  {'total_ms':>10}  {'mean_ms':>8}")
    for r in rows:
        print(f"  {r['query']:<80}  {r['calls']:>5}  "
              f"{str(r['total_ms']):>10}  {str(r['mean_ms']):>8}")


if __name__ == "__main__":
    try:
        main()
    finally:
        close_pool()

python

"""发布文章 + 标签管理

运行：python code/posts.py
"""
from __future__ import annotations

from db import fetch_one, fetch_all, execute, close_pool


def publish(tenant_id: int, author_id: int, title: str, body: str,
            tags: list[str]) -> int:
    """发布一篇文章，返回 post_id。
    触发器会自动根据 title/body 生成 search_vector。
    """
    row = fetch_one(
        "SELECT api_publish(%s, %s, %s, %s, %s) AS id",
        (tenant_id, author_id, title, body, tags),
    )
    return row["id"]


def list_recent(tenant_id: int, limit: int = 10) -> list[dict]:
    return fetch_all(
        """SELECT id, title, tags, created_at, view_count
           FROM   posts
           WHERE  tenant_id = %s AND status = 'published'
           ORDER  BY created_at DESC
           LIMIT  %s""",
        (tenant_id, limit),
    )


def filter_by_tag(tenant_id: int, tag: str) -> list[dict]:
    """JSONB GIN 索引：包含某标签的文章。"""
    # tags @> '["postgres"]' 会走 idx_posts_tags(jsonb_path_ops) GIN 索引
    return fetch_all(
        """SELECT id, title, tags
           FROM   posts
           WHERE  tenant_id = %s
             AND  tags @> %s::jsonb
           ORDER  BY created_at DESC""",
        (tenant_id, f'["{tag}"]'),
    )


def increment_view(post_id: int, created_at) -> int:
    """分区表的更新需要带分区键（created_at）才能裁剪。"""
    return execute(
        """UPDATE posts
           SET    view_count = view_count + 1
           WHERE  id = %s AND created_at = %s""",
        (post_id, created_at),
    )


def _demo() -> None:
    print("=== 发文 / 标签查询演示 ===")

    pid = publish(
        tenant_id=1,
        author_id=1,
        title="测试发文：PG 的全文检索真香",
        body="本文演示发文时触发器自动维护 search_vector 的能力，"
             "以及 JSONB 标签 GIN 索引的查询加速。",
        tags=["postgres", "fulltext", "demo"],
    )
    print(f"[OK] 新文章 id={pid}")

    print("\n-- 最新 5 篇 --")
    for r in list_recent(1, 5):
        print(f" #{r['id']:>3}  {r['title'][:40]}  tags={r['tags']}")

    print("\n-- tag='postgres' 的文章（JSONB GIN） --")
    rows = filter_by_tag(1, "postgres")
    print(f" 共 {len(rows)} 篇，前 3 篇：")
    for r in rows[:3]:
        print(f"   #{r['id']:>3}  {r['title'][:40]}")


if __name__ == "__main__":
    try:
        _demo()
    finally:
        close_pool()

python

"""物化视图刷新：mv_hot_posts

教学要点：
    1. REFRESH MATERIALIZED VIEW        → 会锁表，读也被阻塞
    2. REFRESH MATERIALIZED VIEW CONCURRENTLY
         → 不阻塞读，但要求视图上有 UNIQUE 索引（init.sql 已建）
    3. 生产上用 pg_cron 定时触发：
         SELECT cron.schedule('refresh_hot', '0 * * * *',
                              'REFRESH MATERIALIZED VIEW CONCURRENTLY blog.mv_hot_posts');

运行：python code/refresh_hot.py
"""
from __future__ import annotations

import time

from db import fetch_all, execute, close_pool


def refresh(concurrent: bool = True) -> float:
    """刷新物化视图，返回耗时秒。"""
    sql = ("REFRESH MATERIALIZED VIEW CONCURRENTLY blog.mv_hot_posts"
           if concurrent
           else "REFRESH MATERIALIZED VIEW blog.mv_hot_posts")
    t0 = time.time()
    execute(sql)
    return time.time() - t0


def top_n(tenant_id: int, n: int = 10) -> list[dict]:
    return fetch_all(
        """SELECT id, title, view_count, recent_likes, hot_score
           FROM   mv_hot_posts
           WHERE  tenant_id = %s
           ORDER  BY hot_score DESC
           LIMIT  %s""",
        (tenant_id, n),
    )


if __name__ == "__main__":
    try:
        cost = refresh(concurrent=True)
        print(f"[OK] 物化视图刷新完成，耗时 {cost:.3f}s")

        print("\n=== tenant=1 的热门 Top 10 ===")
        for r in top_n(1, 10):
            print(f" score={r['hot_score']:7.2f}  "
                  f"views={r['view_count']:>5}  "
                  f"likes={r['recent_likes']:>3}  "
                  f"#{r['id']:>3}  {r['title'][:40]}")
    finally:
        close_pool()

python

"""全文检索：tsvector + ts_rank + GIN

核心：
    to_tsquery / plainto_tsquery / websearch_to_tsquery 把用户输入转成查询
    search_vector @@ query  是否命中
    ts_rank(search_vector, query)  相关度排序

运行：python code/search_fulltext.py "postgres 索引"
"""
from __future__ import annotations

import sys

from db import fetch_all, close_pool


def search(tenant_id: int, q: str, limit: int = 10) -> list[dict]:
    """走 GIN(search_vector) 全文检索，按 ts_rank 降序。

    websearch_to_tsquery 会自动处理用户输入中的空格、引号、减号，
    比 plainto_tsquery 更贴近 Google 风格搜索。
    """
    sql = """
        SELECT
            id, title,
            ts_rank(search_vector,
                    websearch_to_tsquery('simple', %(q)s)) AS rank,
            ts_headline('simple', body,
                        websearch_to_tsquery('simple', %(q)s),
                        'MaxWords=30, MinWords=10, StartSel=<<, StopSel=>>')
                AS snippet
        FROM    posts
        WHERE   tenant_id = %(t)s
          AND   status = 'published'
          AND   search_vector @@ websearch_to_tsquery('simple', %(q)s)
        ORDER   BY rank DESC
        LIMIT   %(l)s
    """
    return fetch_all(sql, {"q": q, "t": tenant_id, "l": limit})


def explain(tenant_id: int, q: str) -> list[dict]:
    """打印执行计划，用于教学演示 GIN 索引是否生效。"""
    sql = """
        EXPLAIN (ANALYZE, BUFFERS, FORMAT TEXT)
        SELECT id, title
        FROM   posts
        WHERE  tenant_id = %s
          AND  search_vector @@ websearch_to_tsquery('simple', %s)
        ORDER  BY ts_rank(search_vector,
                          websearch_to_tsquery('simple', %s)) DESC
        LIMIT  10
    """
    return fetch_all(sql, (tenant_id, q, q))


def _demo(q: str) -> None:
    print(f"=== 全文检索：{q!r} ===")
    rows = search(1, q, limit=5)
    if not rows:
        print("  （无结果）")
    for r in rows:
        print(f" rank={r['rank']:.4f}  #{r['id']:>3}  {r['title'][:40]}")
        print(f"   摘要：{r['snippet']}")
    print()

    print("-- EXPLAIN --")
    for r in explain(1, q):
        print(" ", r["QUERY PLAN"])


if __name__ == "__main__":
    keyword = sys.argv[1] if len(sys.argv) > 1 else "postgres MVCC"
    try:
        _demo(keyword)
    finally:
        close_pool()

python

"""模糊检索：pg_trgm

场景：用户输入有拼写错误、或只记得一半关键词。
    similarity(a, b)  → [0,1] 的相似度
    a % b             → 相似度 >= pg_trgm.similarity_threshold (默认 0.3)
    a <-> b           → 距离 (1 - similarity)，越小越相似，可 ORDER BY

运行：python code/search_trgm.py "psotgres"   # 故意拼错
"""
from __future__ import annotations

import sys

from db import fetch_all, close_pool


def search(tenant_id: int, q: str, limit: int = 10) -> list[dict]:
    """用 title <-> %s 走 GIN(title gin_trgm_ops) 相似度排序。"""
    sql = """
        SELECT id, title,
               similarity(title, %(q)s) AS sim
        FROM   posts
        WHERE  tenant_id = %(t)s
          AND  status = 'published'
          AND  title %% %(q)s                -- % 是 pg_trgm 的相似操作符，psycopg 里要写 %%
        ORDER  BY title <-> %(q)s
        LIMIT  %(l)s
    """
    return fetch_all(sql, {"q": q, "t": tenant_id, "l": limit})


def explain(tenant_id: int, q: str) -> list[dict]:
    sql = """
        EXPLAIN (ANALYZE, BUFFERS, FORMAT TEXT)
        SELECT id, title
        FROM   posts
        WHERE  tenant_id = %s AND title %% %s
        ORDER  BY title <-> %s
        LIMIT  10
    """
    return fetch_all(sql, (tenant_id, q, q))


def _demo(q: str) -> None:
    print(f"=== 模糊检索：{q!r} ===")
    rows = search(1, q, limit=5)
    if not rows:
        print("  （无结果，可调小 pg_trgm.similarity_threshold）")
    for r in rows:
        print(f" sim={r['sim']:.3f}  #{r['id']:>3}  {r['title']}")
    print()

    print("-- EXPLAIN --")
    for r in explain(1, q):
        print(" ", r["QUERY PLAN"])


if __name__ == "__main__":
    keyword = sys.argv[1] if len(sys.argv) > 1 else "psotgres"  # 故意拼错
    try:
        _demo(keyword)
    finally:
        close_pool()

python

"""语义检索：pgvector + HNSW 索引

流程：
    1. 用 sentence-transformers 把文章正文 embed 成 384 维向量
    2. 写入 post_embeddings 表（vector(384) 列 + HNSW 索引）
    3. 用 `<=>`（余弦距离）做近邻检索

依赖：sentence-transformers, numpy  （可选；没有时会 fallback 到随机向量 demo）
运行：python code/search_vector.py "如何做向量检索？"
"""
from __future__ import annotations

import os
import sys
from typing import List

from db import fetch_all, fetch_one, execute, close_pool

EMBED_DIM = 384
MODEL_NAME = "all-MiniLM-L6-v2"


# ------------------------------- embedding --------------------------------
def _get_encoder():
    """懒加载 embedding 模型。没装 sentence-transformers 就返回 None。"""
    try:
        from sentence_transformers import SentenceTransformer
    except ImportError:
        return None
    global _encoder
    if "_encoder" not in globals() or _encoder is None:
        print(f"[init] 加载 {MODEL_NAME} ...")
        _encoder = SentenceTransformer(MODEL_NAME)
    return _encoder


def embed(text: str) -> List[float]:
    enc = _get_encoder()
    if enc is None:
        # fallback：随机向量（仅供流程演示，不可用于真实检索）
        import random
        random.seed(hash(text) & 0xFFFFFFFF)
        return [random.random() for _ in range(EMBED_DIM)]
    vec = enc.encode(text, normalize_embeddings=True)
    return vec.tolist()


def _to_pg_vector(vec: List[float]) -> str:
    """pgvector 文本表示：'[0.1,0.2,...]'"""
    return "[" + ",".join(f"{x:.6f}" for x in vec) + "]"


# -------------------------------- 入库 ------------------------------------
def upsert_embedding(post_id: int, text: str) -> None:
    vec = embed(text)
    execute(
        """INSERT INTO post_embeddings(post_id, model, embedding)
           VALUES (%s, %s, %s::vector)
           ON CONFLICT (post_id, model) DO UPDATE
              SET embedding = EXCLUDED.embedding,
                  updated_at = now()""",
        (post_id, MODEL_NAME, _to_pg_vector(vec)),
    )


def build_index_for_all(tenant_id: int) -> int:
    """为该租户所有已发布文章生成 embedding。"""
    rows = fetch_all(
        """SELECT id, title, body
           FROM   posts
           WHERE  tenant_id = %s AND status = 'published'
             AND  NOT EXISTS (
                    SELECT 1 FROM post_embeddings e
                    WHERE  e.post_id = posts.id AND e.model = %s)""",
        (tenant_id, MODEL_NAME),
    )
    for r in rows:
        upsert_embedding(r["id"], f"{r['title']}\n{r['body']}")
    return len(rows)


# -------------------------------- 检索 ------------------------------------
def search(tenant_id: int, query: str, limit: int = 5) -> list[dict]:
    """用 `<=>` 余弦距离做最近邻。HNSW 索引会命中 idx_post_emb_hnsw。"""
    qvec = _to_pg_vector(embed(query))
    return fetch_all(
        """SELECT p.id, p.title,
                  (e.embedding <=> %s::vector) AS distance
           FROM   post_embeddings e
           JOIN   posts p ON p.id = e.post_id
           WHERE  p.tenant_id = %s AND e.model = %s
           ORDER  BY e.embedding <=> %s::vector
           LIMIT  %s""",
        (qvec, tenant_id, MODEL_NAME, qvec, limit),
    )


def explain(tenant_id: int, query: str) -> list[dict]:
    qvec = _to_pg_vector(embed(query))
    return fetch_all(
        """EXPLAIN (ANALYZE, BUFFERS, FORMAT TEXT)
           SELECT post_id
           FROM   post_embeddings e
           ORDER  BY embedding <=> %s::vector
           LIMIT  5""",
        (qvec,),
    )


# -------------------------------- demo ------------------------------------
def _demo(query: str) -> None:
    enc = _get_encoder()
    if enc is None:
        print("[WARN] 未安装 sentence-transformers，仅做流程演示（结果不准）。")
        print("       可运行：pip install sentence-transformers")

    n = build_index_for_all(1)
    print(f"[OK] 新生成 {n} 条 embedding（已入库的会跳过）")

    print(f"\n=== 语义检索：{query!r} ===")
    for r in search(1, query, limit=5):
        print(f" dist={r['distance']:.4f}  #{r['id']:>3}  {r['title'][:50]}")

    print("\n-- EXPLAIN（验证 HNSW 索引是否生效）--")
    for r in explain(1, query):
        print(" ", r["QUERY PLAN"])


if __name__ == "__main__":
    q = sys.argv[1] if len(sys.argv) > 1 else "如何实现向量检索?"
    try:
        _demo(q)
    finally:
        close_pool()

api.py ↗ · auth.py ↗ · comments_tree.py ↗ · db.py ↗ · multi_tenant_demo.py ↗ · perf_audit.py ↗ · posts.py ↗ · refresh_hot.py ↗ · search_fulltext.py ↗ · search_trgm.py ↗ · search_vector.py ↗

第 19 章 综合实战项目：博客 + 全文检索 + 标签 + 评论树 + AI 语义检索 ​

0. 导读：为什么选这个项目？ ​

1. 项目目标与功能清单 ​

1.1 项目目标 ​

1.2 用户故事 ​

1.3 REST 接口清单（代码见 19_project/code/api.py） ​

2. 数据模型设计 ​

2.1 ER 图（Mermaid） ​

2.2 设计要点（对应到前 18 章） ​

3. 完整 init.sql 逐段讲解 ​

3.1 扩展一次性装齐 ​

3.2 Schema 隔离 ​

3.3 users 表：多租户 + bcrypt 密码 ​

3.4 posts 主表：分区 + JSONB + tsvector ​

3.5 父表上建索引自动下推 ​

3.6 comments 评论表：自关联 ​

3.7 post_embeddings：pgvector 语义检索 ​

3.8 触发器：自动维护 search_vector ​

3.9 物化视图 mv_hot_posts ​

3.10 RLS 策略 ​

4. 业务功能实现 ​

4.1 注册登录（pgcrypto） ​

4.2 发文 + 标签 + 自动 search_vector ​

4.3 全文检索 ​

4.4 模糊检索（pg_trgm） ​

4.5 语义检索（pgvector） ​

4.6 评论树（递归 CTE） ​

4.7 JSONB GIN 按标签筛选 ​

4.8 月度归档（分区表 DETACH） ​

4.9 热门榜物化视图刷新 ​

4.10 多租户隔离验证 ​

5. 性能调优实战 ​

5.1 制造一个慢查询 ​

5.2 pg_stat_statements 找 Top 慢 SQL ​

5.3 加索引后对比 ​

5.4 JSONB 标签：jsonb_path_ops 更小 ​

5.5 语义检索：HNSW 参数调优 ​

5.6 auto_explain：自动记录慢查询 ​

6. 部署清单 ​

6.1 Docker Compose（最简方式） ​

6.2 必装扩展清单 ​

7. 本项目映射的 PG 知识点一览 ​

8. 进一步可做的扩展 ​

9. 面试高频题 ​

Q1：为什么 PG 适合做「全文 + 向量 + JSONB」混合搜索？与 ES / Milvus 组合相比有什么取舍？ ​

Q2：分区表上如何正确地建索引、建唯一约束？ ​

Q3：REFRESH MATERIALIZED VIEW 和 REFRESH ... CONCURRENTLY 的区别？ ​

Q4：多租户的 3 种常见做法？RLS 的利弊？ ​

Q5：EXPLAIN 里看到 Bitmap Heap Scan 但性能依然差，排查思路？ ​

Q6：本项目里 view_count 的频繁 UPDATE 会导致什么问题？怎么缓解？ ​

Q7：pgvector 的 IVFFlat 和 HNSW 怎么选？ ​

Q8：如何做这个项目的灰度升级（PG 14 → 16）？ ​

10. 本章小结 ​

🎬 可视化演示 ​

💻 示例代码 ​

第 19 章综合实战项目：博客 + 全文检索 + 标签 + 评论树 + AI 语义检索

0. 导读：为什么选这个项目？

1. 项目目标与功能清单

1.1 项目目标

1.2 用户故事

1.3 REST 接口清单（代码见 `19_project/code/api.py`）

2. 数据模型设计

2.1 ER 图（Mermaid）

2.2 设计要点（对应到前 18 章）

3. 完整 init.sql 逐段讲解

3.1 扩展一次性装齐

3.2 Schema 隔离

3.3 `users` 表：多租户 + bcrypt 密码

3.4 `posts` 主表：分区 + JSONB + tsvector

3.5 父表上建索引自动下推

3.6 `comments` 评论表：自关联

3.7 `post_embeddings`：pgvector 语义检索

3.8 触发器：自动维护 `search_vector`

3.9 物化视图 `mv_hot_posts`

3.10 RLS 策略

4. 业务功能实现

4.1 注册登录（pgcrypto）

4.2 发文 + 标签 + 自动 search_vector

4.3 全文检索

4.4 模糊检索（pg_trgm）

4.5 语义检索（pgvector）

4.6 评论树（递归 CTE）

4.7 JSONB GIN 按标签筛选

4.8 月度归档（分区表 DETACH）

4.9 热门榜物化视图刷新

4.10 多租户隔离验证

5. 性能调优实战

5.1 制造一个慢查询

5.2 `pg_stat_statements` 找 Top 慢 SQL

5.3 加索引后对比

5.4 JSONB 标签：jsonb_path_ops 更小

5.5 语义检索：HNSW 参数调优

5.6 auto_explain：自动记录慢查询

6. 部署清单

6.1 Docker Compose（最简方式）

6.2 必装扩展清单

7. 本项目映射的 PG 知识点一览

8. 进一步可做的扩展

9. 面试高频题

Q1：为什么 PG 适合做「全文 + 向量 + JSONB」混合搜索？与 ES / Milvus 组合相比有什么取舍？

Q2：分区表上如何正确地建索引、建唯一约束？

Q3：`REFRESH MATERIALIZED VIEW` 和 `REFRESH ... CONCURRENTLY` 的区别？

Q4：多租户的 3 种常见做法？RLS 的利弊？

Q5：`EXPLAIN` 里看到 `Bitmap Heap Scan` 但性能依然差，排查思路？

Q6：本项目里 `view_count` 的频繁 UPDATE 会导致什么问题？怎么缓解？

Q7：pgvector 的 IVFFlat 和 HNSW 怎么选？

Q8：如何做这个项目的灰度升级（PG 14 → 16）？

10. 本章小结

🎬 可视化演示

💻 示例代码