第 18 章常用扩展生态：PG 真正的「瑞士军刀」

目标读者：已经能熟练用 SQL，想知道「为什么大家都说 PostgreSQL 这么强」「PostGIS / pgvector 这些到底怎么用」。
学完你会：知道 PG 有哪些「明星扩展」、各自解决什么问题、怎么 CREATE EXTENSION 装上；会用 PostGIS 写「附近商家」；会用 pgvector 做语义检索 / RAG；会用 pg_cron 替代 OS crontab。

0. 导读：扩展才是 PG 的「灵魂」

很多人对 PostgreSQL 的最大误解是：「不就是个开源 SQL 数据库吗？跟 MySQL 比有啥本质区别？」

答案是扩展生态：

PostgreSQL 从设计之初就把「可扩展」当成一等公民。你可以在不动内核的情况下，给它新增类型、操作符、索引方法、过程语言、外部表协议、后台进程。
全球有 200+ 高质量扩展，把 PG 变成：地理库、时序库、向量库、图数据库、消息队列、分布式数据库、CDC 平台……

一句话总结这章想让你建立的认知：

你需要	装一个扩展即可
地理位置	PostGIS
AI 向量检索 / RAG	pgvector
时序指标	TimescaleDB
模糊搜索 / 拼写纠错	pg_trgm
定时任务	pg_cron
自动分区	pg_partman
水平分片	Citus
跑 Python 函数	plpython3u
图数据查询	Apache AGE
把 WAL 变 JSON 给 Kafka	wal2json

📌 与 MySQL 的对比：MySQL 没有「官方推荐 + 一键 CREATE EXTENSION」的扩展生态。MySQL 类似的能力分散在：MySQL Spatial（空间，弱很多）、ProxySQL（路由）、第三方插件（不一定稳定）。PG 的扩展机制是写在内核里的，扩展加载后能新增类型、操作符、甚至索引访问方法 —— 这是 MySQL 做不到的。

1. 扩展机制概览

1.1 一个扩展由什么组成？

某扩展 = 一组 .sql 脚本 + 一组 .so 动态库（如有 C 代码） + 一份 .control 元数据
        放在 $PG_LIB_DIR/extension/ 下

组件	作用
`xxx.control`	扩展元数据：版本、依赖、是否 superuser only
`xxx--1.0.sql`	创建函数 / 类型 / 操作符 / 索引方法的 SQL
`xxx.so`	（可选）C 写的动态库，提供 PG 函数实现

1.2 三个核心命令

sql

-- 1. 看本地系统装了什么扩展（contrib + 第三方）
SELECT * FROM pg_available_extensions;

-- 2. 看当前数据库已经启用哪些扩展
SELECT extname, extversion FROM pg_extension;

-- 3. 启用 / 卸载
CREATE EXTENSION pg_trgm;
CREATE EXTENSION pg_trgm WITH SCHEMA myschema;       -- 指定 schema
CREATE EXTENSION pg_trgm CASCADE;                    -- 自动装依赖
DROP EXTENSION pg_trgm;
ALTER EXTENSION pg_trgm UPDATE TO '1.6';             -- 升级版本

1.3 安装流程（OS 层面）

① contrib 内置扩展：

bash

# Ubuntu/Debian
apt install postgresql-contrib

# RHEL / CentOS
yum install postgresql-contrib

contrib 装好后，pg_trgm、pgcrypto、btree_gin、tablefunc、hstore、pg_stat_statements 等就摆在文件系统里了，去 psql 里 CREATE EXTENSION 即可启用。

② 第三方扩展（PostGIS、pgvector、TimescaleDB、Citus）：

bash

# 以 PostGIS 为例（Ubuntu）
apt install postgresql-16-postgis-3
psql -U postgres -c 'CREATE EXTENSION postgis;'

# 以 pgvector 为例（源码）
git clone --branch v0.7.4 https://github.com/pgvector/pgvector
cd pgvector && make && sudo make install
psql -U postgres -c 'CREATE EXTENSION vector;'

1.4 `shared_preload_libraries`：必须重启的扩展

某些扩展（pg_stat_statements、auto_explain、pg_cron、citus、timescaledb）需要 PG 启动时就加载它们的钩子，必须配在 postgresql.conf：

ini

shared_preload_libraries = 'pg_stat_statements,auto_explain,pg_cron'

改完必须重启 PG，然后再 CREATE EXTENSION xxx;。

配套代码 code/01_extensions_inventory.py 一键打印「已启用 / 可启用 / 关注的明星扩展是否就绪」三类清单。

2. 必装的内置扩展（contrib）

2.1 `pg_stat_statements`：SQL 雷达

第 17 章已经详细讲过，所有 PG 实例都应该装。

2.2 `pgcrypto`：加密 + UUID + HMAC

sql

CREATE EXTENSION pgcrypto;

-- 1) 随机 UUID（替代 SERIAL 主键的常见选择）
SELECT gen_random_uuid();
-- → 8e6e0c2b-0a4a-46cb-ba1a-3a0a2c66f8b6

-- 2) 对称加密
SELECT pgp_sym_encrypt('我的银行卡号', 'master-key');
SELECT pgp_sym_decrypt(pgp_sym_encrypt('hi','k')::TEXT::bytea, 'k');

-- 3) 哈希（密码存储）
SELECT crypt('user_password', gen_salt('bf', 10));

-- 4) HMAC 签名
SELECT encode(hmac('webhook payload', 'secret-key', 'sha256'), 'hex');

📌 与 MySQL 的区别：MySQL 也有 MD5/SHA2/AES_ENCRYPT，但 gen_random_uuid() 在 MySQL 里要手动用 UUID() 或 8.0.13+ 的 UUID_TO_BIN()。PG 的 pgcrypto 提供了完整的 PGP 体系。

2.3 `pageinspect`：看页/元组的二进制结构

学习内核必装：

sql

CREATE EXTENSION pageinspect;

-- 看 perf_orders 第 0 页的元组（xmin/xmax/ctid 一目了然）
SELECT lp, t_xmin, t_xmax, t_ctid, t_infomask
FROM heap_page_items(get_raw_page('perf_orders', 0))
LIMIT 5;

-- 看一个 B-Tree 索引页内部结构
SELECT * FROM bt_page_items('perf_orders_pkey', 1) LIMIT 5;

2.4 `pgstattuple`：精确膨胀测量

sql

CREATE EXTENSION pgstattuple;

SELECT * FROM pgstattuple('perf_orders');
-- table_len | tuple_count | tuple_len | dead_tuple_count | dead_tuple_len | free_space

比 pg_relation_size 更精准，但要全表扫，谨慎在大表上跑。

2.5 `pg_trgm`：模糊搜索神器

PG 的「模糊搜索三件套」之一。把字符串切成连续 3 字符片段（trigram），然后做集合相似度计算。最强的能力是让 LIKE '%xxx%' 走 GIN 索引。

sql

CREATE EXTENSION pg_trgm;

-- 关键：建 GIN trigram 索引
CREATE INDEX ch18_idx_goods_name_trgm ON ch18_goods USING gin (name gin_trgm_ops);

-- ✅ 即使是「左右都模糊」也能用上索引
SELECT * FROM ch18_goods WHERE name LIKE '%iPhone 15%';

-- 拼写纠错
SELECT name, similarity(name, 'iPhne 15 Pr') AS sim
FROM ch18_goods
WHERE name % 'iPhne 15 Pr'              -- '%' 等价 similarity > 阈值
ORDER BY sim DESC LIMIT 5;

-- KNN 距离（无 WHERE）
SELECT name, name <-> 'iPhne' AS dist
FROM ch18_goods
ORDER BY name <-> 'iPhne'
LIMIT 5;

阈值控制：

sql

SHOW pg_trgm.similarity_threshold;             -- 默认 0.3
SET pg_trgm.similarity_threshold = 0.2;        -- 调宽

📌 中文也能用，但因为中文不分词，建议输入 ≥ 3 个字。生产中文搜索一般再叠加 zhparser（PG 中文分词扩展）。

2.6 `btree_gin` / `btree_gist`：让 GIN/GiST 也能索引普通类型

GIN 默认只索引 JSONB / 数组 / 全文等「集合类」类型；GiST 默认只索引几何 / 范围。装上 btree_gin 就能在 GIN 索引里加普通字段当前缀：

sql

CREATE EXTENSION btree_gin;

-- 复合 GIN：先按 status 过滤，再按 payload 模糊匹配
CREATE INDEX idx_t_status_payload
  ON events USING gin (status, payload jsonb_path_ops);

btree_gist 的经典用途是排他约束：

sql

CREATE EXTENSION btree_gist;

CREATE TABLE booking (
    id      BIGSERIAL PRIMARY KEY,
    room_id INT NOT NULL,
    during  TSRANGE NOT NULL,
    EXCLUDE USING gist (room_id WITH =, during WITH &&)  -- 同房间时间不能重叠
);

2.7 `hstore`：键值对类型

PG 9.4 之前 JSONB 还没出，hstore 是首选 KV 存储。现在大多场景被 JSONB 取代，但仍很轻量（不嵌套、不存数组），对「只存简单字符串 KV」场景反而更高效。

sql

CREATE EXTENSION hstore;
SELECT 'a=>1, b=>2'::hstore -> 'a';   -- 1

2.8 `tablefunc`：交叉表 crosstab

把行转成列，相当于 Excel 数据透视表：

sql

CREATE EXTENSION tablefunc;

SELECT * FROM crosstab(
  $$SELECT month, category, sales FROM monthly_sales ORDER BY 1, 2$$,
  $$VALUES ('food'), ('book'), ('tech')$$
) AS ct(month TEXT, food NUMERIC, book NUMERIC, tech NUMERIC);

2.9 `unaccent`、`intarray`、`dblink`、`postgres_fdw`、`file_fdw`

扩展	一句话	典型用法
`unaccent`	去重音	`SELECT unaccent('café résumé'); -- cafe resume`
`intarray`	整数数组高级运算	`WHERE tag_ids && ARRAY[1,2]` 走 GIN 索引
`dblink`	函数式调远程 PG	`SELECT * FROM dblink('host=...', 'SELECT id FROM t') AS (id INT);`
`postgres_fdw`	把远程 PG 表当本地表（推荐）	`IMPORT FOREIGN SCHEMA public FROM SERVER s INTO public;`
`file_fdw`	把 CSV 当外部表查	`CREATE FOREIGN TABLE logs (...) SERVER files OPTIONS (filename '/var/log/a.csv', format 'csv');`

3. PostGIS（地理信息系统）

PostGIS 是 PG 上第一个杀手级扩展，在地理信息领域是事实标准。任何「附近商家」「物流路径」「打车定位」都能用它。

3.1 安装与启用

bash

# Ubuntu
apt install postgresql-16-postgis-3

sql

CREATE EXTENSION postgis;
SELECT PostGIS_Full_Version();

3.2 几何类型与 SRID

PostGIS 提供两套类型：

类型	单位	适用	性能
`geometry`	度（平面坐标）	大量空间运算（intersect / union / 缓冲区）	快
`geography`	米（球面坐标）	距离 / 临近搜索	慢但准确

SRID（Spatial Reference System ID）是「坐标系编号」：

SRID	名称	何时用
`4326`	WGS84 经纬度	GPS / 移动端 / GeoJSON 默认
`3857`	Web Mercator	地图渲染（Google/Mapbox）
`2154`	法国本土 Lambert	地区性精确测量

📌 新手必踩坑：插入 POI 时一定要 ST_SetSRID(ST_MakePoint(lon, lat), 4326) 显式声明 SRID，否则后续做距离计算会得出毫无意义的天文数字。

3.3 基础类型

sql

SELECT 'POINT(116.4 39.9)'::geometry;
SELECT 'LINESTRING(0 0, 1 1, 2 0)'::geometry;
SELECT 'POLYGON((0 0, 1 0, 1 1, 0 1, 0 0))'::geometry;

-- 推荐写法（带 SRID）
SELECT ST_GeomFromText('POINT(116.4 39.9)', 4326);
SELECT ST_MakePoint(116.4, 39.9);

3.4 实战：附近 3km 商家

sql

CREATE TABLE ch18_poi (
    id   BIGSERIAL PRIMARY KEY,
    name TEXT NOT NULL,
    kind TEXT NOT NULL,
    geom geography(POINT, 4326) NOT NULL
);

CREATE INDEX ch18_idx_poi_geom ON ch18_poi USING gist (geom);

-- 查询：用户在天安门附近 3000 米
SELECT id, name, kind,
       ST_Distance(geom, ST_MakePoint(116.397428, 39.90923)::geography) AS dist_m
FROM ch18_poi
WHERE ST_DWithin(geom, ST_MakePoint(116.397428, 39.90923)::geography, 3000)
ORDER BY geom <-> ST_MakePoint(116.397428, 39.90923)::geography
LIMIT 20;

关键 API：

函数 / 操作符	含义
`ST_DWithin(g1, g2, dist)`	距离 ≤ dist 米（走 GiST 索引，强烈推荐）
`ST_Distance(g1, g2)`	精确距离（米，因为是 geography）
`ST_Within / ST_Contains / ST_Intersects`	包含 / 相交
`g1 <-> g2`	KNN 距离操作符，配合 `ORDER BY` 走索引 Top N
`ST_AsGeoJSON(g)`	输出 GeoJSON 给前端
`ST_Buffer(g, r)`	缓冲区（圆 / 多边形向外扩 r 米）
`ST_Transform(g, srid)`	坐标系转换

配套代码 code/02_postgis_nearby.py 实测「无索引 vs GiST 索引」的耗时对比。

3.5 GeoJSON 端到端

前端通常用 Leaflet / Mapbox，吃 GeoJSON：

sql

SELECT json_build_object(
  'type', 'FeatureCollection',
  'features', json_agg(json_build_object(
      'type', 'Feature',
      'geometry', ST_AsGeoJSON(geom)::json,
      'properties', json_build_object('name', name, 'kind', kind)
  ))
) FROM ch18_poi WHERE ST_DWithin(geom, ...);

3.6 进阶能力

pgrouting 扩展：在 PostGIS 之上做路网寻路（A*、Dijkstra），可以做「最短驾车路线」
MobilityDB 扩展：移动对象时空数据库
ST_ClusterDBSCAN：空间聚类（找热点区域）

4. pgvector（向量数据库）

AI 时代必学。pgvector 让 PG 直接支持向量检索，省去「另搭一个 Pinecone / Milvus / Weaviate」的运维成本。

4.1 安装

bash

git clone --branch v0.7.4 https://github.com/pgvector/pgvector
cd pgvector && make && sudo make install

sql

CREATE EXTENSION vector;

云上：AWS RDS / Aliyun RDS / Supabase 都已默认支持 CREATE EXTENSION vector;。

4.2 类型与基础

sql

-- 1536 维（OpenAI text-embedding-3-small）
-- 384 维（sentence-transformers/all-MiniLM-L6-v2，便宜小巧）
CREATE TABLE ch18_docs (
    id        BIGSERIAL PRIMARY KEY,
    title     TEXT,
    content   TEXT,
    embedding vector(1536)
);

INSERT INTO ch18_docs (title, content, embedding) VALUES
  ('文档1', '...', '[0.1, 0.2, 0.3, ...]'::vector);

4.3 三种距离操作符

操作符	含义	适用
`<->`	L2 欧氏距离	一般情况
`<#>`	负内积（值越小越相似）	embedding 已归一化
`<=>`	cosine 距离	文本 embedding 推荐

sql

SELECT id, title, embedding <=> '[0.1,...]'::vector AS dist
FROM ch18_docs
ORDER BY embedding <=> '[0.1,...]'::vector
LIMIT 5;

4.4 索引：HNSW vs IVFFlat

sql

-- HNSW（pgvector 0.5.0+，推荐）：召回好、速度快
CREATE INDEX ON ch18_docs USING hnsw (embedding vector_cosine_ops);

-- IVFFlat：节省内存，但要先 ANALYZE 才能建好倒排
CREATE INDEX ON ch18_docs USING ivfflat (embedding vector_cosine_ops) WITH (lists = 100);

索引	优点	缺点	何时用
HNSW	召回 > 95%，查询 < 5ms	内存占用大，建索引慢	数据量 < 1000w
IVFFlat	占用小，建索引快	召回略低，需要选 lists 参数	数据量大、内存敏感

经验：HNSW + cosine 距离是 90% 文本 RAG 场景的默认。

4.5 实战：RAG 检索

python

# 伪代码
from sentence_transformers import SentenceTransformer
model = SentenceTransformer("sentence-transformers/all-MiniLM-L6-v2")

def retrieve(query: str, k: int = 5):
    qemb = model.encode([query], normalize_embeddings=True)[0]
    qvec = "[" + ",".join(f"{v:.6f}" for v in qemb) + "]"
    cur.execute("""
        SELECT id, title, content,
               embedding <=> %s::vector AS dist
        FROM ch18_docs
        ORDER BY embedding <=> %s::vector
        LIMIT %s
    """, (qvec, qvec, k))
    return cur.fetchall()

把 retrieve() 的结果喂给 LLM 做 prompt context，就是最简单的 RAG。

配套代码 code/03_pgvector_rag.py 是完整可跑示例（含 sentence-transformers 真实 embedding；模型下不下来时自动降级到伪 embedding）。

5. TimescaleDB（时序数据库）

PG 的「时序专用扩展」。适合监控指标、IoT 设备数据、点击流、金融 tick 等「按时间高速写入 + 按时间窗口聚合查询」的场景。

5.1 核心概念：hypertable

普通表：所有行都在一张物理表里，时间长了表巨大、查询慢。

hypertable = 按时间自动切分的「逻辑表」：

sql

CREATE EXTENSION timescaledb;

CREATE TABLE metrics (
    time   TIMESTAMPTZ NOT NULL,
    device TEXT NOT NULL,
    cpu    DOUBLE PRECISION,
    mem    DOUBLE PRECISION
);

SELECT create_hypertable('metrics', 'time',
                         chunk_time_interval => INTERVAL '1 day');

之后你照常 INSERT / SELECT，TimescaleDB 自动按天切分到内部子表（chunk）。查询带时间范围时自动裁剪只扫相关 chunk。

5.2 连续聚合（continuous aggregate）

在原始指标上实时维护滚动窗口聚合：

sql

CREATE MATERIALIZED VIEW metrics_1h
WITH (timescaledb.continuous) AS
SELECT time_bucket('1 hour', time) AS bucket,
       device,
       avg(cpu) AS avg_cpu,
       max(mem) AS max_mem
FROM metrics
GROUP BY 1, 2;

-- 自动刷新策略
SELECT add_continuous_aggregate_policy('metrics_1h',
       start_offset => INTERVAL '3 hours',
       end_offset   => INTERVAL '1 hour',
       schedule_interval => INTERVAL '15 minutes');

5.3 自动压缩

sql

ALTER TABLE metrics SET (
  timescaledb.compress,
  timescaledb.compress_segmentby = 'device',
  timescaledb.compress_orderby = 'time DESC'
);

SELECT add_compression_policy('metrics', INTERVAL '7 days');
-- → 7 天前的数据自动压缩，体积可降到 1/10

5.4 vs InfluxDB / Prometheus

维度	TimescaleDB	InfluxDB
查询语言	标准 SQL	InfluxQL / Flux
ACID	✅ 完整事务	❌
JOIN	✅ 与普通 PG 表关联	❌
学习成本	已会 SQL = 0	要学新语言

经验：有大量已有 PG 业务数据、需要 JOIN 关联维度表就选 TimescaleDB；纯指标 + 高基数标签且只有一套监控就选 Prometheus / InfluxDB。

6. pg_cron（PG 内的定时任务）

不用再装 crontab + psql -c，直接在 PG 内调度任务。

6.1 安装

bash

apt install postgresql-16-cron

ini

# postgresql.conf
shared_preload_libraries = 'pg_cron'
cron.database_name = 'learn_pg'        # 任务表所在库

重启 PG → CREATE EXTENSION pg_cron;

6.2 注册任务

sql

-- 每天凌晨 3 点清理 30 天前的日志
SELECT cron.schedule(
  '清理日志',
  '0 3 * * *',
  $$DELETE FROM logs WHERE ts < now() - interval '30 days'$$
);

-- 指定目标库（推荐）
SELECT cron.schedule_in_database(
  'refresh_mv', '*/5 * * * *',
  'REFRESH MATERIALIZED VIEW CONCURRENTLY user_stats',
  'learn_pg'
);

-- 查询任务表
SELECT * FROM cron.job;
SELECT * FROM cron.job_run_details ORDER BY start_time DESC LIMIT 10;

-- 注销
SELECT cron.unschedule('清理日志');

配套代码 code/04_pg_cron_demo.py 演示了完整的注册 / 列表 / 历史 / 注销流程，并在 pg_cron 没装时给出清晰的安装提示。

6.3 vs OS crontab

维度	pg_cron	OS crontab
调度器	PG 内进程	OS 进程
任务历史	`cron.job_run_details` 视图	自己写日志
主备切换	主库才执行（自动）	要手动管理
跨机器一致	自动（任务存在 PG 里）	要分发脚本

主备场景下 pg_cron 自动只在 primary 上执行，故障切换后新主库立刻接手任务，这是 OS crontab 完全做不到的。

7. pg_partman（自动分区管理）

第 16 章讲过 PG 的声明式分区，但手动建分区子表很烦：日表要每天建一个新子表、删旧子表。pg_partman 自动化这件事。

sql

CREATE EXTENSION pg_partman;

-- 把 events 改成「按 created_at 按天分区」
CREATE TABLE events (
    id BIGSERIAL,
    created_at TIMESTAMPTZ NOT NULL,
    payload JSONB
) PARTITION BY RANGE (created_at);

SELECT partman.create_parent(
  p_parent_table => 'public.events',
  p_control      => 'created_at',
  p_type         => 'range',
  p_interval     => '1 day',
  p_premake      => 4              -- 预建未来 4 天的分区
);

-- 配合 pg_cron：每天凌晨自动建明天分区 + 删 30 天前老分区
UPDATE partman.part_config
SET retention = '30 days', retention_keep_table = false
WHERE parent_table = 'public.events';

SELECT cron.schedule('partman_maintenance', '0 1 * * *',
                     $$SELECT partman.run_maintenance(p_analyze => true)$$);

效果：「每天涌进 1 亿条事件」也能在 PG 里轻松管 5 年。

8. Citus（分布式 PG）

详见第 16 章，简介：

sql

CREATE EXTENSION citus;
SELECT create_distributed_table('orders', 'user_id');  -- 按 user_id 分片到 worker

Citus 让 PG 变成水平分布式数据库，PB 级 OLAP / 多租户 SaaS 都能扛。

9. PL/Python、PL/V8（多语言函数）

PG 默认支持 PL/pgSQL（第 12 章讲过）。装上语言扩展后还能跑别的：

sql

CREATE EXTENSION plpython3u;            -- u = untrusted（superuser 才能装）

CREATE FUNCTION py_word_count(s TEXT) RETURNS INT AS $$
    return len(s.split())
$$ LANGUAGE plpython3u;

SELECT py_word_count('Hello world from PG');   -- 4

PL/V8 类似，可以跑 JavaScript（含完整 V8 引擎，速度极快）。

⚠️ 这些 untrusted 语言能直接调系统调用 / 读文件，生产慎用，最好放在专门的 schema + 权限隔离。

10. Apache AGE（图数据库）

把 PG 变成图数据库，支持 OpenCypher（Neo4j 的查询语言）：

sql

CREATE EXTENSION age;
LOAD 'age';
SET search_path = ag_catalog, "$user", public;

SELECT create_graph('social');

SELECT * FROM cypher('social', $$
  CREATE (:Person {name: 'Alice'})-[:KNOWS]->(:Person {name: 'Bob'})
$$) AS (a agtype);

SELECT * FROM cypher('social', $$
  MATCH (a:Person)-[:KNOWS*1..3]-(b:Person)
  RETURN a.name, b.name
$$) AS (a agtype, b agtype);

适合做社交关系、知识图谱、推荐路径。规模一般不如 Neo4j，但省一个组件。

11. ZomboDB（PG 索引接 Elasticsearch）

sql

CREATE EXTENSION zombodb;
CREATE INDEX idx_es ON products USING zombodb ((products.*))
  WITH (url='http://localhost:9200/');

-- 用 ES query string DSL 直接查
SELECT * FROM products WHERE products ==> 'name:iPhone AND price:[100 TO 500]';

适合：已有 ES 集群、想省客户端两路写。索引由 PG 维护，事务一致性比手动双写好。

12. wal2json：CDC 神器

把 WAL 解码为 JSON 流，供 Kafka / Flink / Debezium 消费做 CDC（Change Data Capture）：

ini

# postgresql.conf
wal_level = logical
shared_preload_libraries = 'wal2json'

sql

SELECT * FROM pg_create_logical_replication_slot('cdc_slot', 'wal2json');

-- 在另一个 session 做 INSERT/UPDATE
-- 然后取变更流：
SELECT data FROM pg_logical_slot_get_changes('cdc_slot', NULL, NULL);

输出示例：

json

{
  "change": [{
    "kind": "update",
    "schema": "public",
    "table": "orders",
    "columnnames": ["id","status"],
    "columnvalues": [42, 2],
    "oldkeys": {"keynames":["id"], "keyvalues":[42]}
  }]
}

Debezium PG Connector 默认就基于 wal2json / pgoutput，是构建实时数仓 / 跨库同步的标配。

13. 选型决策树速查

场景	推荐扩展
全文 / 模糊搜索	`pg_trgm` + GIN（轻量）；中文加 `zhparser`；高级用 `zombodb`
空间地理	`postgis`；路网寻路加 `pgrouting`
时序 / 监控指标	`timescaledb`（推荐） / 仅按日分区用 `pg_partman`
AI / RAG / 语义检索	`pgvector`（HNSW + cosine）
定时任务	`pg_cron`
CDC	`wal2json` + Debezium
图数据	`age`（OpenCypher）
加密 / UUID	`pgcrypto` + `gen_random_uuid()`
跨库查询	频繁：`postgres_fdw`；偶尔：`dblink`
排他约束（时间不重叠等）	`btree_gist` + `EXCLUDE USING gist`
性能调优	`pg_stat_statements` + `auto_explain` + `pg_buffercache`（必装三件套）

14. 与 MySQL 的对比

能力	PostgreSQL	MySQL
扩展机制	`CREATE EXTENSION`，原生支持新增类型/操作符/索引方法	UDF + 插件机制，能力受限
空间能力	PostGIS（业界标杆）	MySQL Spatial（弱很多，不支持地理坐标距离）
向量检索	pgvector（生态完善）	8.0.32+ 有 VECTOR 类型，但不如 pgvector 成熟
时序	TimescaleDB	无官方对应（一般用 ProxySQL/分区表勉强）
全文搜索	pg_trgm / 内置 GIN tsvector	FULLTEXT 索引，能力较弱
定时任务	pg_cron	EVENT SCHEDULER（内置，但功能简陋）
自动分区	pg_partman	手动 ALTER TABLE
分布式	Citus	MySQL Cluster / Vitess
图数据	Apache AGE	无
CDC	wal2json + Debezium	binlog + Debezium

总结：MySQL 走「内核做大做全」路线，PG 走「内核精简 + 扩展生态」路线。PG 一旦遇到非典型需求（地理、向量、时序、图），扩展几乎都能 100% 解决；MySQL 往往得「另搭一个数据库」。

15. 小结

CREATE EXTENSION 是 PG 的灵魂指令：装上扩展即可拥有新类型、新索引、新函数，不动内核。
必装内置三件套：pg_stat_statements、pg_trgm、pgcrypto。
PostGIS 让 PG 在地理领域无敌；pgvector 让 PG 在 AI 领域立刻能用。
TimescaleDB / pg_cron / pg_partman 是「时序生产力组合」：自动分区 + 自动调度。
过程语言扩展（PL/Python、PL/V8）让你能在 PG 里写复杂业务逻辑，但生产慎用 untrusted 版。
wal2json / Debezium 是 CDC 标准方案，把 PG 变成事件源。
选型决策树：先看「我要解决什么问题」，再去地图里找扩展。不要为了用扩展而用扩展。
生态 vs MySQL：PG 的扩展机制是 MySQL 真正难以追平的护城河。

🎮 配套演示

用浏览器打开 ./18_extension/demo.html，跟着可视化动画再走一遍本章核心概念。
配套代码在 ./18_extension/code/，每个脚本都可以独立 python xxx.py 运行，先跑 init.sql 准备 POI / docs / goods 演示表（注意要先安装对应扩展）。

16. 面试高频题

Q1：PG 的「扩展（extension）」机制和 MySQL 的插件有什么本质区别？

考察点：扩展机制设计哲学、对内核的影响、可扩展类型/操作符。

答案：

PG 扩展是「数据库对象的集合」：一个扩展可以定义新的类型、函数、操作符、索引访问方法、过程语言、外部数据包装器（FDW）、后台 worker。比如 PostGIS 加了 geometry 类型 + GiST 索引方法 + 几百个 ST_* 函数。
MySQL 插件机制有限：主要支持「存储引擎、UDF（用户函数）、审计/认证插件」，不能新增类型、不能新增索引方法、不能新增操作符。所以 MySQL 没法做出 PostGIS 这种深度集成的扩展。
打包安装：PG 扩展 = .control + .sql + .so，统一放 $PG_LIB_DIR/extension，CREATE EXTENSION 一行启用、DROP EXTENSION 一行卸载，依赖 / 版本 / 升级一应俱全。
三种装载时机：
- 普通扩展：CREATE EXTENSION 后立刻可用
- 钩子型扩展（pg_stat_statements / auto_explain / pg_cron / Citus）：必须配 shared_preload_libraries 后重启 PG
- 两段式扩展（PostGIS）：装包 + CREATE EXTENSION 即可，不需重启
后果：PG 在「地理、时序、向量、图、全文」等垂直领域都有一流方案；MySQL 一旦遇到这些需求往往得「另起一个 DB」。
运维体验：pg_available_extensions 一键看本地有什么、pg_extension 看启用了什么，对比 SHOW PLUGINS 信息丰富得多。

加分项：能说出 PostGIS 是怎么注册新索引方法 GIST 的（pg_am 表）；能区分 "trusted" 和 "untrusted" 扩展（trusted 普通用户也能装）。

Q2：PostGIS 里 `ST_DWithin` 和 `ST_Distance` 有什么区别？为什么「附近搜索」要用前者？

考察点：PostGIS 索引使用、空间查询性能。

答案：

ST_Distance(a, b) 返回精确距离（米），但这是个函数，PG 没法用它和某常量比较时直接走索引。
ST_DWithin(a, b, dist) 返回布尔，专门设计成索引可优化：PG 拿到这个表达式会自动构造一个边界框（bbox），先用 GiST 索引筛掉不可能的点，再做精确判断。
没有索引时两者都得全表扫，差别不大；有 GiST 索引时，ST_DWithin 比 ST_Distance(...) <= dist 快几个数量级。

正确写法：

sql

CREATE INDEX ON ch18_poi USING gist (geom);
SELECT * FROM ch18_poi
WHERE ST_DWithin(geom, ST_MakePoint(lon, lat)::geography, 3000)
ORDER BY geom <-> ST_MakePoint(lon, lat)::geography
LIMIT 20;

<-> KNN 操作符：与 ORDER BY 配合能让 GiST 索引按距离顺序返回，做 Top N 不用全部排序。
geography vs geometry：用 geography(POINT, 4326)，距离自动是米，自动按球面计算（适合 GPS 经纬度）；用 geometry 必须自己 ST_Transform 到投影坐标系才能算米。新手强烈推荐 geography。
SRID 必填：ST_SetSRID(ST_MakePoint(lon, lat), 4326)，否则距离计算结果是垃圾。

加分项：能说出「<-> 与 ST_Distance 在没有索引时几乎等价；有索引时 <-> 走 KNN 索引一次扫，效率最优」。

Q3：pgvector 的 HNSW 和 IVFFlat 索引有什么区别？怎么选？

考察点：向量检索算法、索引取舍。

答案：

HNSW（Hierarchical Navigable Small World）：层次化的可导航小世界图。速度快、召回高（>95%），是目前行业默认选择。
- 缺点：内存占用大（≈ 向量本身的 1.5~2 倍），建索引慢。
- 参数：m（每层连接数，默认 16）、ef_construction（建索引时探索宽度，默认 64）、ef_search（查询时探索宽度，越大越准越慢）。
IVFFlat（Inverted File with Flat quantization）：把向量空间分 lists 个簇，查询时只在最近的 N 个簇里搜。
- 优点：内存占用小、建索引快。
- 缺点：召回低于 HNSW，需要选好 lists（推荐 rows / 1000），且必须在数据写完后再建索引（否则簇划分是空的）。
怎么选？
- 数据量 < 1000w + 在意召回 → HNSW
- 数据量很大、内存敏感、召回略降可接受 → IVFFlat
距离操作符配套：建索引时要指定 ops class，和查询时用的距离操作符一致：
- vector_l2_ops ↔ <->
- vector_ip_ops ↔ <#>
- vector_cosine_ops ↔ <=>（文本 embedding 推荐）
过滤 + 向量混合查询的坑：WHERE category = 'tech' ORDER BY emb <=> ? 在向量索引选择性差时可能不走索引，要用 partial index 或先过滤后用 LIMIT。
维度限制：HNSW 当前最多 2000 维（PG 0.7+），IVFFlat 也类似。OpenAI text-embedding-3-large（3072 维）需要降维或选别的模型。

加分项：能说出 PG 16+ 的「parallel index build」；能解释为什么 cosine 距离 = 1 - cos(θ)，越小越相似。

Q4：怎么用 PG 做「附近商家」的全套技术方案？请说出表结构、索引、SQL 和性能数量级。

考察点：PostGIS 综合落地能力。

答案：

sql

-- 1) 启用扩展
CREATE EXTENSION postgis;

-- 2) 表结构：geography 让距离单位为米
CREATE TABLE ch18_poi (
    id   BIGSERIAL PRIMARY KEY,
    name TEXT NOT NULL,
    kind TEXT NOT NULL,
    geom geography(POINT, 4326) NOT NULL
);

-- 3) 必建空间索引
CREATE INDEX ch18_idx_poi_geom ON ch18_poi USING gist (geom);

-- 4) 查询：用户在 (lon, lat)，找半径 3km 内 Top 20
SELECT id, name, kind,
       ST_Distance(geom, ST_MakePoint(:lon, :lat)::geography) AS dist_m
FROM ch18_poi
WHERE ST_DWithin(geom, ST_MakePoint(:lon, :lat)::geography, 3000)
ORDER BY geom <-> ST_MakePoint(:lon, :lat)::geography
LIMIT 20;

性能：

5000 ~ 100w 行 POI，本地 SSD：单次查询 < 5ms
关键：ST_DWithin 走 GiST 索引筛掉绝大多数点，只对边界候选做精确距离

还会被追问的细节：

更新策略：POI 经纬度变更 → UPDATE ... SET geom = ST_SetSRID(ST_MakePoint(...), 4326)::geography；GiST 索引自动维护。
GeoJSON 输出：ST_AsGeoJSON(geom) 给前端，可直接喂 Leaflet/Mapbox。
多个分类 / 城市 mixed 查询：用复合条件 WHERE city='北京' AND kind='cafe' AND ST_DWithin(...)，加部分索引 CREATE INDEX ... ON ch18_poi USING gist (geom) WHERE city='北京'。
GeoJSON 多边形（外卖配送范围）：用 ST_Contains(area, ST_MakePoint(...)) 判断用户是否在配送区。
超大数据集（亿级）：考虑按 geohash 前缀分区 + 每分区单独 GiST 索引。

加分项：能说出「为什么 geography 比 geometry 慢但准确」、能写出 H3 / S2 等 cell-based 方案与 PostGIS 的对比。

Q5：什么场景应该用 TimescaleDB？什么场景用 Prometheus / InfluxDB？

考察点：时序方案选型。

答案：

TimescaleDB 适合：

数据要做事务性 ACID —— 比如金融 tick 数据、订单事件流，不允许丢
要 JOIN 维度表 —— 设备表、用户表、产品表。Prometheus 完全没 JOIN
已经有大量 PG 业务 —— 时序数据混在同一个 PG 集群里，运维和权限统一
复杂报表 —— 带 CTE、窗口函数、自定义函数。SQL 能力远强于 InfluxQL/Flux
数据需要按业务字段做查询 —— 不止按时间

Prometheus / InfluxDB 适合：

纯监控指标，写多读少，不需要事务
高基数 label —— Prometheus 的 PromQL 对 label 维度有针对性优化
Pull 模式 + alertmanager 一体化方案 —— Prometheus 整套生态成熟
存储成本敏感 —— InfluxDB 的 TSI 压缩做得很激进
用户已经熟 PromQL —— 切换成本低

TimescaleDB 杀手锏特性：

time_bucket('1 hour', time)：自动按时间桶聚合
连续聚合 + 自动刷新：不用自己写 cron
自动压缩：7 天前数据压到 1/10
数据保留策略：add_retention_policy 自动删旧数据

反模式：用 Prometheus 长期存 1 年以上数据 → 体积爆炸；用 TimescaleDB 做高基数 prom-style metrics → 标签笛卡儿积撑爆 chunk。

加分项：能提到 Citus + TimescaleDB 组合（Citus 11+ 已合并部分时序能力）；提到 OpenTelemetry → ClickHouse / TimescaleDB 的现代选型。

Q6：pg_cron 比 OS crontab 强在哪？主备切换场景下怎么表现？

考察点：定时任务高可用。

答案：

任务定义与数据同库：cron.job 表存在 PG 里，跟着 pg_basebackup / 流复制一起走，主备永远一致。OS crontab 要自己 rsync / Ansible 分发。
主备切换自动接管：pg_cron 后台进程检测「自己是不是 primary」，只在 primary 上执行任务，备库不重复执行；故障切换后新 primary 自动接手。OS crontab 没法感知主备。
任务历史可查询：cron.job_run_details 视图记录每次运行的开始/结束时间、状态、返回信息，SQL 直接查。OS crontab 要解析 /var/log/cron 文本。
权限和 SQL 一体：任务直接是 SQL，不需要拼 psql -c，没有 shell 注入风险。
5 字段 cron 语法：与 Unix cron 一致，迁移无心智负担。
小限制：
- 任务只能跑 SQL（不能跑 shell 命令）
- 默认只能在「cron.database_name」配的库里建 / 调度任务，跨库要用 cron.schedule_in_database
- 失败重试要自己在 SQL 里实现（用事务包 + UPSERT 状态）
典型用例：
- 每日清理日志：DELETE FROM logs WHERE ts < now() - interval '30 days'
- 定期刷新物化视图：REFRESH MATERIALIZED VIEW CONCURRENTLY user_stats
- 配合 pg_partman：SELECT partman.run_maintenance(true) 自动建/删分区
- 定期做 ANALYZE：ANALYZE perf_orders

加分项：能说出在 RDS 上用 pg_cron 是首选（比 lambda + EventBridge 简单）；提到 pg_timetable 是另一个更复杂但能跑 shell 的替代品。

Q7：`pg_trgm` 是怎么让 `LIKE '%xxx%'` 走索引的？背后原理是什么？

考察点：trigram 算法、GIN 索引原理。

答案：

trigram = 连续 3 个字符：pg_trgm 把字符串切成所有连续 3 字符片段。例如 iPhone → {" i", " ip", "iph", "pho", "hon", "one", "ne ", "e "}（前后补空格各 2 个）。
GIN trigram 索引：在所有 trigram 上建倒排索引。每个 trigram 指向「包含它的所有行」的 ctid 集合。
LIKE '%xxx%' 转换：PG 把模式拆成 trigram 集合，然后对索引做交集。比如 LIKE '%iPhone%' 拆成 5 个 trigram，索引返回「同时包含这 5 个 trigram」的候选行，再做精确 LIKE 验证。
能加速的模式：
- LIKE '%xxx%'、LIKE '%xxx'、LIKE 'xxx%'、ILIKE、~（正则）都可以
- 模式至少要有一个完整 trigram（即 ≥ 3 字符），太短没法用索引

建索引方式：

sql

CREATE INDEX ON tbl USING gin (col gin_trgm_ops);   -- 推荐
CREATE INDEX ON tbl USING gist (col gist_trgm_ops); -- 数据量小时也行

额外能力：
- similarity(a, b) 返回 0~1 相似度
- a % b 阈值过滤（默认 0.3，可 SET 调整）
- a <-> b KNN 距离，配合 ORDER BY 做拼写纠错
中文场景：中文字符基本都是 3 字节 UTF-8，所以单个汉字也能形成 trigram。但中文不分词，建议输入 ≥ 3 个汉字；正经中文搜索还是叠加 zhparser 分词更准。
vs 其他方案：
- 简单关键词搜索：pg_trgm 足够
- 中文分词 + 词频权重：tsvector + zhparser
- 复杂相关性 / 多语言：zombodb 接 ES

加分项：能说出 gin_trgm_ops 和 gist_trgm_ops 的区别（GIN 写入慢查询快、GiST 反之）；知道 pg_trgm.similarity_threshold 是会话级 GUC。

Q8：怎么用 PG 实现一个「事件源 / CDC」管道？用到哪些扩展？

考察点：逻辑解码、wal2json、Debezium、生产 CDC 实践。

答案：

架构：

PG (wal_level=logical)
    │
    ├── Replication Slot ── wal2json/pgoutput plugin ── Debezium PG Connector
    │                                                          │
    │                                                          ▼
    │                                                       Kafka
    │                                                          │
    └── 业务 INSERT/UPDATE/DELETE                       下游消费者

步骤：

配置 PG：

ini

wal_level = logical
max_replication_slots = 10
max_wal_senders = 10
shared_preload_libraries = 'wal2json'   # 选用 wal2json 时

创建发布（pgoutput plugin，PG 10+ 推荐）：

sql

CREATE PUBLICATION cdc_pub FOR TABLE orders, payments;

创建 replication slot：

sql

SELECT pg_create_logical_replication_slot('cdc_slot', 'pgoutput');
-- 或 wal2json
SELECT pg_create_logical_replication_slot('cdc_slot', 'wal2json');

Debezium 接入：在 Kafka Connect 上启动 PG Connector，配 slot 名 + publication 名，Debezium 自动开始消费 WAL 流。
消费 JSON 变更事件：Kafka 里每条消息是一个 row 变更，包含 before / after / op (c/u/d/r) 等字段，下游写 ES、ClickHouse、数仓都行。

关键扩展：

wal2json：把 WAL 解码成 JSON（兼容性好，老版本必备）
pgoutput：PG 10+ 内置，效率更高（推荐）
pg_recvlogical：CLI 工具，能脱机消费 slot 验证
pglogical：早期逻辑复制方案，现在大多用 pgoutput

坑：

slot 不消费会撑爆磁盘：未确认的 WAL 不会被回收，slot 不消费 = WAL 永远不删。监控 pg_replication_slots.confirmed_flush_lsn。
大事务延迟：一个 100 万行 UPDATE 会变成 100 万条变更事件，下游延迟暴涨。控制单事务规模。
TOAST 大字段：默认只发 key，不发 toast 列内容，需要 REPLICA IDENTITY FULL 才能拿到完整 OLD/NEW。
DDL 不复制：逻辑复制不传 DDL，schema 变更要手动同步。
主备切换：原 primary 上的 slot 不会自动到新 primary，需要应用层重建 slot 并重做全量初始化（snapshot）。

加分项：能区分「物理复制（流复制，整盘字节流，必须 PG 同版本）」与「逻辑复制（按表/行级，跨大版本可行）」；能说出 PG 14+ 的 failover slot（订阅可跟随主备切换）。

🔗 延伸阅读

第 3 章数据类型：JSONB / 数组是 PostGIS / pgvector 之外原生的「半结构化」杀手锏，理解它们再看扩展更通透。
第 6 章索引与查询优化：本章用到的 GIN（pg_trgm）、GiST（PostGIS）、HNSW/IVFFlat（pgvector）都是索引扩展点，配合第 6 章的访问方法图谱看更系统。
第 19 章综合项目：把 pgvector + pg_stat_statements + pg_cron 串起来，做一个真实的语义搜索 + 定时归档系统。

至此，第 17、18 章完成。下一章「综合实战项目」会贯穿全教程的所有特性，做一个端到端的真实业务系统。

🎬 可视化演示

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

💻 示例代码

python

"""01_extensions_inventory.py —— 第 18 章配套代码 #1

用途
    列出当前 PG 实例上：
        ① 已经 CREATE EXTENSION 启用的扩展（pg_extension）
        ② 安装在文件系统、可以 CREATE EXTENSION 启用的扩展（pg_available_extensions）
        ③ 这些扩展中各自的版本与简介
    生产排查时第一步：先确认有哪些扩展可用，再决定方案。

运行
    python 01_extensions_inventory.py
"""

from __future__ import annotations

import sys

try:
    import psycopg
except ImportError:
    sys.exit('请先安装 psycopg v3：pip install "psycopg[binary]>=3.1"')


CONN_INFO = "host=127.0.0.1 port=5432 dbname=learn_pg user=postgres"


def banner(t: str) -> None:
    print()
    print("=" * 80)
    print(f" {t}")
    print("=" * 80)


def fmt_table(headers, rows, widths=None):
    if not rows:
        print("  (空)")
        return
    if widths is None:
        widths = [max(len(str(h)), max((len(str(r[i])) for r in rows), default=0))
                  for i, h in enumerate(headers)]
    print("  " + "  ".join(f"{h:<{widths[i]}}" for i, h in enumerate(headers)))
    print("  " + "  ".join("-" * w for w in widths))
    for r in rows:
        print("  " + "  ".join(f"{str(r[i]):<{widths[i]}}" for i in range(len(headers))))


# 一些常用扩展的「人话简介」
HUMAN_DESC = {
    "pg_stat_statements":  "聚合 SQL 执行统计（必装）",
    "pgcrypto":            "对称/非对称加密、UUID、HMAC",
    "pageinspect":         "查看页/元组的二进制结构",
    "pgstattuple":         "精确测量表/索引膨胀",
    "pg_trgm":             "三元组相似度，LIKE %x% 提速",
    "btree_gin":           "让 GIN 也支持普通类型",
    "btree_gist":          "让 GiST 也支持普通类型",
    "hstore":              "key-value 类型（JSONB 之前的方案）",
    "tablefunc":           "crosstab 交叉表 / normal_rand",
    "unaccent":            "去除重音字符",
    "intarray":            "整数数组高级操作符",
    "dblink":              "执行远程 PG 上的 SQL",
    "postgres_fdw":        "把远程 PG 表当本地表查",
    "file_fdw":            "把 CSV 当外部表查",
    "postgis":             "地理信息系统（重磅）",
    "vector":              "向量类型 + HNSW/IVFFlat 索引（pgvector）",
    "timescaledb":         "时序数据库扩展（hypertable + 压缩）",
    "pg_cron":             "在 PG 内跑 cron 定时任务",
    "pg_partman":          "自动管理时序分区",
    "citus":               "PG 的水平分布式扩展",
    "plpython3u":          "在 PG 中运行 Python 函数",
    "plv8":                "在 PG 中运行 JavaScript 函数",
    "age":                 "图数据库（OpenCypher 语法）",
    "auto_explain":        "慢查询自动 EXPLAIN",
    "wal2json":            "把 WAL 解码为 JSON（CDC）",
    "pg_buffercache":      "查看 shared_buffers 缓存了哪些页",
    "pgrouting":           "PostGIS 之上的路网寻路",
}


def main() -> None:
    try:
        conn = psycopg.connect(CONN_INFO, autocommit=True)
    except psycopg.OperationalError as exc:
        sys.exit(f"连接失败：{exc}")

    with conn, conn.cursor() as cur:
        banner("① 已启用扩展（pg_extension）")
        cur.execute("""
            SELECT extname,
                   extversion,
                   nspname AS schema
            FROM pg_extension e
            JOIN pg_namespace n ON n.oid = e.extnamespace
            ORDER BY extname
        """)
        rows = [(name, ver, sch, HUMAN_DESC.get(name, ""))
                for name, ver, sch in cur.fetchall()]
        fmt_table(["extname", "version", "schema", "中文说明"], rows)

        banner("② 已安装可用 / 但未启用的扩展（pg_available_extensions）")
        cur.execute("""
            SELECT name, default_version, installed_version, comment
            FROM pg_available_extensions
            WHERE installed_version IS NULL
            ORDER BY name
        """)
        rows = cur.fetchall()
        if rows:
            fmt_table(
                ["name", "default_ver", "installed_ver", "official comment"],
                [(n, dv, iv or "-", (c or "")[:60]) for n, dv, iv, c in rows],
            )
            print(f"\n  共 {len(rows)} 个扩展可用，启用方法：CREATE EXTENSION 名字;")
        else:
            print("  (无未启用扩展)")

        banner("③ 全教程关注的「明星扩展」是否就绪？")
        STARS = ["pg_stat_statements", "pg_trgm", "pgcrypto", "btree_gin",
                 "postgis", "vector", "timescaledb", "pg_cron",
                 "pg_partman", "citus", "auto_explain", "wal2json"]
        cur.execute("SELECT name FROM pg_available_extensions")
        avail = {r[0] for r in cur.fetchall()}
        cur.execute("SELECT extname FROM pg_extension")
        installed = {r[0] for r in cur.fetchall()}

        rows = []
        for ext in STARS:
            if ext in installed:
                status, hint = "✅ 已启用", "可直接使用"
            elif ext in avail:
                status, hint = "🟡 可启用", f"CREATE EXTENSION {ext};"
            else:
                status, hint = "❌ 未安装", "需在 OS 层装包：见对应章节"
            rows.append((ext, status, HUMAN_DESC.get(ext, ""), hint))
        fmt_table(["extension", "状态", "用途", "下一步"], rows)


if __name__ == "__main__":
    main()

python

"""02_postgis_nearby.py —— 第 18 章配套代码 #2

用途
    用 PostGIS 实现「附近 N 公里的商家」查询，并对比有/无 GiST 空间索引的耗时。
    场景：用户站在天安门 (116.397428, 39.90923)，找 3 公里以内的 POI。

依赖
    1. PostGIS：apt install postgresql-XX-postgis-3
    2. 已运行 init.sql 准备 ch18_poi 表（5000 行 POI）

预期输出
    [无索引] 全表扫            : XX ms, 找到 N 个
    [GiST 索引] ST_DWithin     : YY ms, 找到 N 个
    Top 5 最近 POI:
      1) POI #321  cafe        125.3 m
      ...
"""

from __future__ import annotations

import sys
import time

try:
    import psycopg
except ImportError:
    sys.exit('请先安装 psycopg v3：pip install "psycopg[binary]>=3.1"')


CONN_INFO = "host=127.0.0.1 port=5432 dbname=learn_pg user=postgres"
USER_LON, USER_LAT = 116.397428, 39.90923
RADIUS_M = 3000


def has_postgis(cur) -> bool:
    cur.execute("SELECT 1 FROM pg_extension WHERE extname = 'postgis'")
    return cur.fetchone() is not None


def main() -> None:
    try:
        conn = psycopg.connect(CONN_INFO, autocommit=True)
    except psycopg.OperationalError as exc:
        sys.exit(f"连接失败：{exc}")

    with conn, conn.cursor() as cur:
        if not has_postgis(cur):
            sys.exit(
                "❌ PostGIS 未启用。请先：\n"
                "   apt install postgresql-XX-postgis-3 (XX = PG 主版本号)\n"
                "   psql -U postgres -d learn_pg -c 'CREATE EXTENSION postgis;'\n"
                "   然后重跑 init.sql 生成 ch18_poi 表。"
            )

        cur.execute("SELECT count(*) FROM ch18_poi")
        total = cur.fetchone()[0]
        print(f"POI 总数：{total}，用户位置 ({USER_LON}, {USER_LAT})，搜索半径 {RADIUS_M} 米\n")

        # 1. 不用空间索引：用 ST_Distance > radius 强制不走索引
        print("[1] 暴力距离过滤（不会走索引，因为函数表达式两边都是变量）")
        cur.execute("DROP INDEX IF EXISTS idx_poi_geom_tmp")  # 确保不影响
        sql_no_idx = """
            SELECT id, name, kind,
                   ST_Distance(geom, ST_MakePoint(%s, %s)::geography) AS dist_m
            FROM ch18_poi
            WHERE ST_Distance(geom, ST_MakePoint(%s, %s)::geography) <= %s
            ORDER BY dist_m
        """
        t0 = time.perf_counter()
        cur.execute(sql_no_idx, (USER_LON, USER_LAT, USER_LON, USER_LAT, RADIUS_M))
        no_idx_rows = cur.fetchall()
        cost_no_idx = (time.perf_counter() - t0) * 1000
        print(f"  → {cost_no_idx:.1f} ms，找到 {len(no_idx_rows)} 个 POI")

        # 2. 用 GiST 索引 + ST_DWithin（PostGIS 推荐姿势）
        print("\n[2] ST_DWithin + GiST 索引（推荐姿势）")
        sql_idx = """
            SELECT id, name, kind,
                   ST_Distance(geom, ST_MakePoint(%s, %s)::geography) AS dist_m
            FROM ch18_poi
            WHERE ST_DWithin(geom, ST_MakePoint(%s, %s)::geography, %s)
            ORDER BY geom <-> ST_MakePoint(%s, %s)::geography
            LIMIT 50
        """
        t0 = time.perf_counter()
        cur.execute(sql_idx, (USER_LON, USER_LAT, USER_LON, USER_LAT, RADIUS_M, USER_LON, USER_LAT))
        idx_rows = cur.fetchall()
        cost_idx = (time.perf_counter() - t0) * 1000
        print(f"  → {cost_idx:.1f} ms，返回前 50 个最近 POI")

        # 3. 看 EXPLAIN
        print("\n[3] EXPLAIN ANALYZE（验证索引是否生效）")
        cur.execute(
            "EXPLAIN (ANALYZE, BUFFERS) " + sql_idx,
            (USER_LON, USER_LAT, USER_LON, USER_LAT, RADIUS_M, USER_LON, USER_LAT),
        )
        for row in cur.fetchall():
            print("  ", row[0])

        # 4. Top 5
        print("\n[4] Top 5 最近 POI")
        for i, (pid, name, kind, dist) in enumerate(idx_rows[:5], 1):
            print(f"   {i}) #{pid:<6} {kind:<10} {name:<30} {dist:>7.1f} m")

        # 5. 关键 PostGIS 操作符 / 函数小抄
        print("\n💡 PostGIS 关键 API 速查：")
        print("   · ST_MakePoint(lon, lat)              → POINT 几何")
        print("   · ::geography                          → 转地理类型，距离单位为米")
        print("   · ST_DWithin(g1, g2, dist)             → 「g1 到 g2 距离 ≤ dist」（走 GiST）")
        print("   · ST_Distance(g1, g2)                  → 精确距离")
        print("   · g1 <-> g2                            → 「KNN 距离操作符」，配合 ORDER BY 用")
        print("   · ST_AsGeoJSON(g)                      → 输出 GeoJSON 给前端")
        print("   · CREATE INDEX ... USING gist (geom)  → 必建空间索引")


if __name__ == "__main__":
    main()

python

"""03_pgvector_rag.py —— 第 18 章配套代码 #3

用途
    用 pgvector 做一个最小可运行的语义检索 demo（RAG 的「R」部分）：
        1. 用 sentence-transformers/all-MiniLM-L6-v2 把若干文档转成 384 维向量
        2. 写入 ch18_docs.embedding（vector 类型）
        3. 建 HNSW 索引
        4. 输入查询文本 → embedding → ORDER BY embedding <=> query_emb LIMIT 5
    输出最相似的 Top 5 文档及 cosine 距离。

依赖
    1. PG 端：pgvector 扩展（CREATE EXTENSION vector）+ 已建好 ch18_docs 表（init.sql 完成）
    2. Python 端：
           pip install "psycopg[binary]>=3.1" sentence-transformers numpy
       第一次运行会自动从 HuggingFace 下载模型（约 80MB）。

降级方案
    如果机器无法访问 HuggingFace，脚本会回退到「随机 + 关键词哈希」的伪 embedding，
    主要为了演示 SQL 流程，并不能体现真实的语义检索效果。
"""

from __future__ import annotations

import sys
import textwrap

try:
    import psycopg
    from psycopg.rows import tuple_row
except ImportError:
    sys.exit('请先安装 psycopg v3：pip install "psycopg[binary]>=3.1"')


CONN_INFO = "host=127.0.0.1 port=5432 dbname=learn_pg user=postgres"
DIM = 384

DOCS = [
    ("PostgreSQL 入门",
     "PostgreSQL 是一款开源的对象-关系数据库管理系统，强调标准 SQL 与可扩展性。"),
    ("MVCC 多版本并发控制",
     "PG 用元组的 xmin/xmax 字段实现多版本并发控制，读不阻塞写、写不阻塞读。"),
    ("WAL 与崩溃恢复",
     "事务先写 Write-Ahead Log 再改数据页，宕机后通过重放 WAL 把数据库带回一致状态。"),
    ("VACUUM 与死元组",
     "更新和删除会留下死元组，autovacuum 后台进程负责清理并防止 XID 回卷。"),
    ("索引种类",
     "B-Tree 适合等值与范围；GIN 适合 JSONB / 全文 / 数组；GiST 适合空间数据；BRIN 适合时序大表。"),
    ("PostGIS 地理信息",
     "PostGIS 是 PG 的空间扩展，提供 geometry/geography 类型，支持「附近商家」等空间查询。"),
    ("pgvector 向量检索",
     "pgvector 扩展提供 vector 类型，配合 HNSW 索引可在 PG 内做语义检索，是 RAG 的常见后端。"),
    ("Streaming Replication",
     "PG 通过流复制把主库 WAL 实时推送到备库，支持同步、异步与 quorum 模式。"),
    ("Logical Replication",
     "逻辑复制基于 WAL 解码，按表/行级别复制变更，可跨大版本、跨架构。"),
    ("PgBouncer",
     "PgBouncer 是 PG 的轻量级连接池，transaction pooling 能把上万客户端复用到几十个真实连接。"),
    ("Citus 分布式 PG",
     "Citus 让 PG 支持水平分片，把大表按 shard key 切到多个 worker 节点。"),
    ("电商订单设计",
     "订单系统通常包含 users / orders / order_items / payments 几张表，利用 RETURNING 与 ON CONFLICT 简化写入。"),
]


def encode_with_sbert(texts: list[str]):
    """优先用 sentence-transformers 生成真实 embedding。"""
    try:
        from sentence_transformers import SentenceTransformer
        model = SentenceTransformer("sentence-transformers/all-MiniLM-L6-v2")
        embs = model.encode(texts, normalize_embeddings=True)
        return embs.tolist()
    except Exception as exc:
        print(f"⚠️  sentence-transformers 不可用 ({exc.__class__.__name__})，"
              "fallback 到伪 embedding（仅演示 SQL 流程，不代表真实语义）")
        return None


def encode_fake(texts: list[str]):
    """兜底：把字符 hash 映射到 384 维归一化向量。"""
    import hashlib
    import math
    out = []
    for t in texts:
        vec = [0.0] * DIM
        for token in t.lower().replace('，',' ').replace('。',' ').split():
            h = int(hashlib.md5(token.encode('utf-8')).hexdigest(), 16)
            for i in range(8):
                idx = (h >> (i * 4)) & (DIM - 1)
                vec[idx] += 1.0
        norm = math.sqrt(sum(v*v for v in vec)) or 1.0
        out.append([v / norm for v in vec])
    return out


def to_pg_vector(vec) -> str:
    """把 list[float] 转 pgvector 文本字面量 [0.1,0.2,...]"""
    return "[" + ",".join(f"{v:.6f}" for v in vec) + "]"


def main() -> None:
    try:
        conn = psycopg.connect(CONN_INFO, autocommit=True)
    except psycopg.OperationalError as exc:
        sys.exit(f"连接失败：{exc}")

    with conn, conn.cursor(row_factory=tuple_row) as cur:
        cur.execute("SELECT 1 FROM pg_extension WHERE extname = 'vector'")
        if cur.fetchone() is None:
            sys.exit(
                "❌ pgvector 未启用。安装：参见 https://github.com/pgvector/pgvector\n"
                "   通常：\n"
                "     git clone --branch v0.7.4 https://github.com/pgvector/pgvector\n"
                "     cd pgvector && make && sudo make install\n"
                "   然后：CREATE EXTENSION vector;"
            )

        # 1. 编码
        texts = [t + "  " + d for t, d in DOCS]
        embs = encode_with_sbert(texts)
        if embs is None:
            embs = encode_fake(texts)
        else:
            assert len(embs[0]) == DIM, f"模型输出维度 {len(embs[0])} ≠ 表定义 {DIM}"

        # 2. 写入 ch18_docs（先清表）
        cur.execute("TRUNCATE ch18_docs RESTART IDENTITY")
        for (title, content), vec in zip(DOCS, embs):
            cur.execute(
                "INSERT INTO ch18_docs (title, content, embedding) VALUES (%s, %s, %s::vector)",
                (title, content, to_pg_vector(vec)),
            )
        print(f"✅ 已写入 {len(DOCS)} 条文档")

        # 3. 建 HNSW 索引（cosine 距离）
        cur.execute("DROP INDEX IF EXISTS ch18_idx_docs_embedding_hnsw")
        try:
            cur.execute(
                "CREATE INDEX ch18_idx_docs_embedding_hnsw "
                "ON ch18_docs USING hnsw (embedding vector_cosine_ops)"
            )
            print("✅ HNSW 索引已建（vector_cosine_ops）")
        except psycopg.errors.FeatureNotSupported:
            print("⚠️  当前 pgvector 版本不支持 HNSW，回退到 IVFFlat")
            cur.execute(
                "CREATE INDEX ch18_idx_docs_embedding_ivf "
                "ON ch18_docs USING ivfflat (embedding vector_cosine_ops) WITH (lists = 10)"
            )

        # 4. 检索
        queries = [
            "怎么处理 PG 里的死元组？",
            "如何在数据库里做语义搜索？",
            "PG 主从同步原理",
        ]
        for q in queries:
            qemb = encode_with_sbert([q]) or encode_fake([q])
            qvec = to_pg_vector(qemb[0])
            print(f"\n🔍 查询：{q}")
            cur.execute(
                """
                SELECT id, title, content,
                       embedding <=> %s::vector AS cos_dist
                FROM ch18_docs
                ORDER BY embedding <=> %s::vector
                LIMIT 3
                """,
                (qvec, qvec),
            )
            for i, (did, title, content, dist) in enumerate(cur.fetchall(), 1):
                print(f"  {i}) [{dist:.3f}] {title}")
                print(f"       {textwrap.shorten(content, width=70)}")

        # 5. 操作符速查
        print("\n💡 pgvector 距离操作符：")
        print("   · v1 <-> v2  → L2 欧氏距离")
        print("   · v1 <#> v2  → 负内积（值越小越相似）")
        print("   · v1 <=> v2  → cosine 距离（推荐用于文本 embedding）")
        print("   · 索引：HNSW（PG 0.5.0+，召回好且快） / IVFFlat（节省内存，要先 ANALYZE）")


if __name__ == "__main__":
    main()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182

python

"""04_pg_cron_demo.py —— 第 18 章配套代码 #4

用途
    演示 pg_cron：在 PG 内部调度定时任务，无需 OS 级 crontab。
    例子：
      · 每天凌晨 3 点清理 30 天前的日志表
      · 每 5 分钟把 raw_metrics 的数据聚合到 hourly_metrics
      · 每周日全量 VACUUM ANALYZE

前置条件
    pg_cron 必须装在专门的 cron schema 里，且 shared_preload_libraries 包含它：
        # postgresql.conf
        shared_preload_libraries = 'pg_cron'
        cron.database_name = 'learn_pg'      # 任务表所在库
    安装：
        Ubuntu/Debian:  apt install postgresql-XX-cron
        编译安装请参考  https://github.com/citusdata/pg_cron
    然后：
        psql -U postgres -c 'CREATE EXTENSION pg_cron;'
        重启 PG。

行为
    · 如果 pg_cron 已装：脚本注册 3 个示例任务，列出当前任务表
    · 如果未装：打印安装步骤后退出
"""

from __future__ import annotations

import sys

try:
    import psycopg
except ImportError:
    sys.exit('请先安装 psycopg v3：pip install "psycopg[binary]>=3.1"')


CONN_INFO = "host=127.0.0.1 port=5432 dbname=learn_pg user=postgres"


JOBS = [
    {
        "name": "demo_clean_logs",
        "schedule": "0 3 * * *",
        "command": "DELETE FROM ch17_perf_orders WHERE created_at < now() - interval '180 days'",
        "desc": "每天凌晨 3 点清理 180 天前的旧订单",
    },
    {
        "name": "demo_refresh_mv",
        "schedule": "*/5 * * * *",
        "command": "ANALYZE ch17_perf_orders",
        "desc": "每 5 分钟刷新 ch17_perf_orders 统计信息",
    },
    {
        "name": "demo_weekly_vacuum",
        "schedule": "0 4 * * 0",
        "command": "VACUUM (ANALYZE, VERBOSE) ch17_perf_orders",
        "desc": "每周日凌晨 4 点 VACUUM ANALYZE",
    },
]


def banner(t: str) -> None:
    print()
    print("=" * 80)
    print(f" {t}")
    print("=" * 80)


def main() -> None:
    try:
        conn = psycopg.connect(CONN_INFO, autocommit=True)
    except psycopg.OperationalError as exc:
        sys.exit(f"连接失败：{exc}")

    with conn, conn.cursor() as cur:
        cur.execute("SELECT 1 FROM pg_extension WHERE extname = 'pg_cron'")
        if cur.fetchone() is None:
            print(
                "❌ pg_cron 未启用。安装步骤：\n"
                "  1. apt install postgresql-XX-cron  （或源码编译）\n"
                "  2. 编辑 postgresql.conf：\n"
                "       shared_preload_libraries = 'pg_cron'\n"
                "       cron.database_name = 'learn_pg'\n"
                "  3. 重启 PG\n"
                "  4. psql -U postgres -d learn_pg -c 'CREATE EXTENSION pg_cron;'\n"
                "  5. 重新运行本脚本"
            )
            return

        banner("① 当前已注册任务（cron.job）")
        cur.execute("""
            SELECT jobid, jobname, schedule, command, active, username, database
            FROM cron.job
            ORDER BY jobid
        """)
        for r in cur.fetchall():
            print(f"  #{r[0]} {r[1]:<25} schedule={r[2]:<12} active={r[4]}  "
                  f"db={r[6]}  user={r[5]}\n      cmd={r[3]}")

        banner("② 注册三个示例任务")
        for job in JOBS:
            cur.execute(
                """
                SELECT cron.schedule_in_database(
                    %s,        -- jobname
                    %s,        -- schedule (5 字段 cron)
                    %s,        -- command
                    'learn_pg' -- database
                )
                """,
                (job["name"], job["schedule"], job["command"]),
            )
            jobid = cur.fetchone()[0]
            print(f"  ✅ #{jobid:<3} {job['name']:<25} schedule={job['schedule']:<12} "
                  f"  -- {job['desc']}")

        banner("③ 任务执行历史（cron.job_run_details，最近 10 条）")
        cur.execute("""
            SELECT jobid, status, start_time, end_time, return_message
            FROM cron.job_run_details
            ORDER BY start_time DESC
            LIMIT 10
        """)
        rows = cur.fetchall()
        if not rows:
            print("  (尚无执行记录，等待下一次调度时间到达)")
        else:
            for r in rows:
                print(f"  job={r[0]} status={r[1]} {r[2]} → {r[3]}  msg={r[4]}")

        banner("④ 卸载示例任务（防止污染）")
        for job in JOBS:
            cur.execute("SELECT cron.unschedule(%s)", (job["name"],))
            print(f"  🗑️  unschedule({job['name']!r})")

        print("\n💡 pg_cron 关键 API：")
        print("   · cron.schedule(name, sched, sql)       注册 / 替换任务（5 字段 cron）")
        print("   · cron.schedule_in_database(...)        指定目标库（推荐）")
        print("   · cron.unschedule(name)                  注销任务")
        print("   · cron.job / cron.job_run_details        任务定义 / 执行历史视图")


if __name__ == "__main__":
    main()

python

"""05_pg_trgm_search.py —— 第 18 章配套代码 #5

用途
    用 pg_trgm 三元组扩展实现「商品名称模糊搜索」：
        ① LIKE '%关键词%' 在大表上一般是全表扫；建 GIN trigram 索引后能秒级返回
        ② similarity(a, b) 返回 0~1 的相似度，可用于「拼写纠错」
        ③ % 操作符 = similarity 阈值过滤

依赖
    · pg_trgm 已启用（init.sql 里已 CREATE EXTENSION pg_trgm）
    · ch18_goods 表已建好 + GIN trigram 索引

预期
    [无索引等价查询] LIKE '%queryword%'  -> Seq Scan ~XX ms
    [GIN trgm 索引]  LIKE '%queryword%'  -> Bitmap Index Scan ~Y ms
    Top 5 相似词:
       1) 0.84  iPhone 15 Pro Max ...
       2) 0.71  ...
"""

from __future__ import annotations

import sys
import time

try:
    import psycopg
except ImportError:
    sys.exit('请先安装 psycopg v3：pip install "psycopg[binary]>=3.1"')


CONN_INFO = "host=127.0.0.1 port=5432 dbname=learn_pg user=postgres"


def banner(t: str) -> None:
    print()
    print("=" * 80)
    print(f" {t}")
    print("=" * 80)


def main() -> None:
    try:
        conn = psycopg.connect(CONN_INFO, autocommit=True)
    except psycopg.OperationalError as exc:
        sys.exit(f"连接失败：{exc}")

    with conn, conn.cursor() as cur:
        cur.execute("SELECT 1 FROM pg_extension WHERE extname = 'pg_trgm'")
        if cur.fetchone() is None:
            sys.exit("❌ pg_trgm 未启用：请先 CREATE EXTENSION pg_trgm（init.sql 已包含）")

        cur.execute("SELECT count(*) FROM ch18_goods")
        total = cur.fetchone()[0]
        print(f"ch18_goods 表行数：{total:,}")

        # 1. 直接 LIKE，看 EXPLAIN 是否走 trgm 索引
        kw = "iPhone"
        banner(f"① LIKE '%{kw}%' + GIN trigram 索引")
        sql = "SELECT id, name FROM ch18_goods WHERE name LIKE %s LIMIT 10"
        cur.execute("EXPLAIN ANALYZE " + sql, (f"%{kw}%",))
        for r in cur.fetchall():
            print("  ", r[0])
        t0 = time.perf_counter()
        cur.execute(sql, (f"%{kw}%",))
        rows = cur.fetchall()
        print(f"  → {(time.perf_counter()-t0)*1000:.2f} ms，前 10 条：")
        for i, (gid, name) in enumerate(rows, 1):
            print(f"     {i}) #{gid} {name}")

        # 2. 临时禁用索引看「裸跑」对比
        banner(f"② 对照：禁用 trgm 索引后跑同样 LIKE")
        cur.execute("SET LOCAL enable_seqscan = on")
        cur.execute("SET LOCAL enable_indexscan = off")
        cur.execute("SET LOCAL enable_bitmapscan = off")
        cur.execute("EXPLAIN ANALYZE " + sql, (f"%{kw}%",))
        for r in cur.fetchall():
            print("  ", r[0])
        cur.execute("RESET enable_seqscan")
        cur.execute("RESET enable_indexscan")
        cur.execute("RESET enable_bitmapscan")

        # 3. similarity 操作符：模糊匹配 + 拼写纠错
        misspell = "iPhne 15 Pr"
        banner(f"③ similarity 拼写纠错：用户输入「{misspell}」")
        cur.execute(
            """
            SELECT name, similarity(name, %s) AS sim
            FROM ch18_goods
            WHERE name %% %s            -- '%' 操作符等价于 similarity > pg_trgm.similarity_threshold
            ORDER BY sim DESC
            LIMIT 5
            """,
            (misspell, misspell),
        )
        for i, (name, sim) in enumerate(cur.fetchall(), 1):
            print(f"   {i}) sim={sim:.3f}  {name}")

        # 4. 当前 similarity_threshold
        cur.execute("SHOW pg_trgm.similarity_threshold")
        thresh = cur.fetchone()[0]
        print(f"\n  当前 pg_trgm.similarity_threshold = {thresh}")
        print("  调整：SET pg_trgm.similarity_threshold = 0.2;  // 默认 0.3")

        # 5. <-> 距离操作符 + ORDER BY KNN
        banner("④ <-> 距离操作符（最相似 Top N，无需 WHERE）")
        cur.execute(
            """
            SELECT name, name <-> %s AS distance
            FROM ch18_goods
            ORDER BY name <-> %s
            LIMIT 5
            """,
            (misspell, misspell),
        )
        for i, (name, dist) in enumerate(cur.fetchall(), 1):
            print(f"   {i}) dist={dist:.3f}  {name}")

        print("\n💡 pg_trgm 速查：")
        print("   · 字符串切成连续 3 字符片段（trigram）做集合相似度")
        print("   · LIKE '%xxx%' / ILIKE → GIN trigram 索引")
        print("   · % 操作符 → similarity 阈值过滤")
        print("   · <-> 操作符 → 距离 = 1 - similarity，ORDER BY 拿 Top N")
        print("   · 中文也能用，但建议长度 ≥ 3 个字符")


if __name__ == "__main__":
    main()

markdown

# 第 18 章 配套代码

本章演示 **PostGIS 地理查询**、**pgvector RAG 语义检索**、**pg_cron 定时任务**、**pg_trgm 模糊搜索**等扩展生态实战。所有自建表都加了 `ch18_` 前缀（`ch18_poi`、`ch18_docs`、`ch18_goods`），避免和其他章节冲突。

## 准备工作

1. 跑一次 init.sql 初始化基础扩展和测试表：

   ```bash
   psql -h 127.0.0.1 -U postgres -d learn_pg -f ../init.sql

会启用内置扩展（pg_trgm、pgcrypto、btree_gin、tablefunc、unaccent），并尝试启用 PostGIS 与 pgvector（装了才建对应的表）。

每个脚本依赖的扩展不同，请按需安装：

脚本	必需扩展	安装命令（Ubuntu）
`01_extensions_inventory.py`	无	（只读 `pg_extension`）
`02_postgis_nearby.py`	`postgis`	`apt install postgresql-XX-postgis-3` 后 `CREATE EXTENSION postgis;`
`03_pgvector_rag.py`	`vector`	源码编译 pgvector 后 `CREATE EXTENSION vector;`
`04_pg_cron_demo.py`	`pg_cron`	`apt install postgresql-XX-cron`，并把 `pg_cron` 加到 `shared_preload_libraries`，重启后 `CREATE EXTENSION pg_cron;`
`05_pg_trgm_search.py`	`pg_trgm`（内置）	`CREATE EXTENSION pg_trgm;`（init.sql 已做）

安装 Python 依赖：

bash

pip install "psycopg[binary]>=3.1"
# 03 脚本可选：用真实 sentence-transformers 嵌入模型（不装则用伪向量演示）
pip install sentence-transformers

（可选）通过环境变量覆盖默认连接：

bash

export PG_DSN="host=127.0.0.1 port=5432 dbname=learn_pg user=postgres password=postgres"

脚本一览（推荐运行顺序）

脚本	一句话说明	关键 PG 特性
`01_extensions_inventory.py`	列出当前库已装/可装扩展，看清楚生态全景	`pg_extension`、`pg_available_extensions`
`02_postgis_nearby.py`	用 `ch18_poi` 演示「天安门附近 1km 内有哪些咖啡店」	`geography`、`ST_DWithin`、GiST 索引
`03_pgvector_rag.py`	把若干文档转成 384 维向量灌进 `ch18_docs`，做语义检索	`vector(384)`、HNSW / IVFFlat、`<=>` 余弦距离
`04_pg_cron_demo.py`	注册 3 个示例 cron 任务（`ch17_perf_orders` 上的 VACUUM/ANALYZE/DELETE）	`cron.schedule_in_database`、`cron.job`
`05_pg_trgm_search.py`	用 `ch18_goods` 演示 `LIKE '%xxx%'` 走 GIN trigram 索引	`gin_trgm_ops`、`similarity()`、`%` 阈值

预期输出（节选 02）

🔍 距天安门 1km 内的 cafe（按距离升序）：
  · POI #1283   distance=  213 m
  · POI #4517   distance=  587 m
  · POI #2031   distance=  834 m
  ...

常见报错与依赖

connection refused → PG 未启动。
extension "postgis" is not available → 没装 PostGIS 系统包，参见上面的安装表。
extension "vector" is not available → 没装 pgvector，源码编译后再 CREATE EXTENSION vector;。
pg_cron must be loaded via shared_preload_libraries → 改 postgresql.conf，加上 shared_preload_libraries='pg_cron' 与 cron.database_name='learn_pg'，重启 PG。
relation "ch18_poi" does not exist → init.sql 跑的时候 PostGIS 没装，跳过了建表；装好 PostGIS 后重跑 init.sql。
03 用 sentence-transformers 时首次会下模型，需要 ≥ 1GB 显存或较长 CPU 等待；脚本提供「伪向量回退」分支。

01_extensions_inventory.py ↗ · 02_postgis_nearby.py ↗ · 03_pgvector_rag.py ↗ · 04_pg_cron_demo.py ↗ · 05_pg_trgm_search.py ↗ · README.md ↗

第 18 章 常用扩展生态：PG 真正的「瑞士军刀」 ​

0. 导读：扩展才是 PG 的「灵魂」 ​

1. 扩展机制概览 ​

1.1 一个扩展由什么组成？ ​

1.2 三个核心命令 ​

1.3 安装流程（OS 层面） ​

1.4 shared_preload_libraries：必须重启的扩展 ​

2. 必装的内置扩展（contrib） ​

2.1 pg_stat_statements：SQL 雷达 ​

2.2 pgcrypto：加密 + UUID + HMAC ​

2.3 pageinspect：看页/元组的二进制结构 ​

2.4 pgstattuple：精确膨胀测量 ​

2.5 pg_trgm：模糊搜索神器 ​

2.6 btree_gin / btree_gist：让 GIN/GiST 也能索引普通类型 ​

2.7 hstore：键值对类型 ​

2.8 tablefunc：交叉表 crosstab ​

2.9 unaccent、intarray、dblink、postgres_fdw、file_fdw ​

3. PostGIS（地理信息系统） ​

3.1 安装与启用 ​

3.2 几何类型与 SRID ​

3.3 基础类型 ​

3.4 实战：附近 3km 商家 ​

3.5 GeoJSON 端到端 ​

3.6 进阶能力 ​

4. pgvector（向量数据库） ​

4.1 安装 ​

4.2 类型与基础 ​

4.3 三种距离操作符 ​

4.4 索引：HNSW vs IVFFlat ​

4.5 实战：RAG 检索 ​

5. TimescaleDB（时序数据库） ​

5.1 核心概念：hypertable ​

5.2 连续聚合（continuous aggregate） ​

5.3 自动压缩 ​

5.4 vs InfluxDB / Prometheus ​

6. pg_cron（PG 内的定时任务） ​

6.1 安装 ​

6.2 注册任务 ​

6.3 vs OS crontab ​

7. pg_partman（自动分区管理） ​

8. Citus（分布式 PG） ​

9. PL/Python、PL/V8（多语言函数） ​

10. Apache AGE（图数据库） ​

11. ZomboDB（PG 索引接 Elasticsearch） ​

12. wal2json：CDC 神器 ​

13. 选型决策树速查 ​

14. 与 MySQL 的对比 ​

15. 小结 ​

🎮 配套演示 ​

16. 面试高频题 ​

Q1：PG 的「扩展（extension）」机制和 MySQL 的插件有什么本质区别？ ​

Q2：PostGIS 里 ST_DWithin 和 ST_Distance 有什么区别？为什么「附近搜索」要用前者？ ​

Q3：pgvector 的 HNSW 和 IVFFlat 索引有什么区别？怎么选？ ​

Q4：怎么用 PG 做「附近商家」的全套技术方案？请说出表结构、索引、SQL 和性能数量级。 ​

Q5：什么场景应该用 TimescaleDB？什么场景用 Prometheus / InfluxDB？ ​

Q6：pg_cron 比 OS crontab 强在哪？主备切换场景下怎么表现？ ​

Q7：pg_trgm 是怎么让 LIKE '%xxx%' 走索引的？背后原理是什么？ ​

Q8：怎么用 PG 实现一个「事件源 / CDC」管道？用到哪些扩展？ ​

🔗 延伸阅读 ​

🎬 可视化演示 ​

💻 示例代码 ​

脚本一览（推荐运行顺序） ​

预期输出（节选 02） ​

常见报错与依赖 ​

第 18 章常用扩展生态：PG 真正的「瑞士军刀」

0. 导读：扩展才是 PG 的「灵魂」

1. 扩展机制概览

1.1 一个扩展由什么组成？

1.2 三个核心命令

1.3 安装流程（OS 层面）

1.4 `shared_preload_libraries`：必须重启的扩展

2. 必装的内置扩展（contrib）

2.1 `pg_stat_statements`：SQL 雷达

2.2 `pgcrypto`：加密 + UUID + HMAC

2.3 `pageinspect`：看页/元组的二进制结构

2.4 `pgstattuple`：精确膨胀测量

2.5 `pg_trgm`：模糊搜索神器

2.6 `btree_gin` / `btree_gist`：让 GIN/GiST 也能索引普通类型

2.7 `hstore`：键值对类型

2.8 `tablefunc`：交叉表 crosstab

2.9 `unaccent`、`intarray`、`dblink`、`postgres_fdw`、`file_fdw`

3. PostGIS（地理信息系统）

3.1 安装与启用

3.2 几何类型与 SRID

3.3 基础类型

3.4 实战：附近 3km 商家

3.5 GeoJSON 端到端

3.6 进阶能力

4. pgvector（向量数据库）

4.1 安装

4.2 类型与基础

4.3 三种距离操作符

4.4 索引：HNSW vs IVFFlat

4.5 实战：RAG 检索

5. TimescaleDB（时序数据库）

5.1 核心概念：hypertable

5.2 连续聚合（continuous aggregate）

5.3 自动压缩

5.4 vs InfluxDB / Prometheus

6. pg_cron（PG 内的定时任务）

6.1 安装

6.2 注册任务

6.3 vs OS crontab

7. pg_partman（自动分区管理）

8. Citus（分布式 PG）

9. PL/Python、PL/V8（多语言函数）

10. Apache AGE（图数据库）

11. ZomboDB（PG 索引接 Elasticsearch）

12. wal2json：CDC 神器

13. 选型决策树速查

14. 与 MySQL 的对比

15. 小结

🎮 配套演示

16. 面试高频题

Q1：PG 的「扩展（extension）」机制和 MySQL 的插件有什么本质区别？

Q2：PostGIS 里 `ST_DWithin` 和 `ST_Distance` 有什么区别？为什么「附近搜索」要用前者？

Q3：pgvector 的 HNSW 和 IVFFlat 索引有什么区别？怎么选？

Q4：怎么用 PG 做「附近商家」的全套技术方案？请说出表结构、索引、SQL 和性能数量级。

Q5：什么场景应该用 TimescaleDB？什么场景用 Prometheus / InfluxDB？

Q6：pg_cron 比 OS crontab 强在哪？主备切换场景下怎么表现？

Q7：`pg_trgm` 是怎么让 `LIKE '%xxx%'` 走索引的？背后原理是什么？

Q8：怎么用 PG 实现一个「事件源 / CDC」管道？用到哪些扩展？

🔗 延伸阅读

🎬 可视化演示

💻 示例代码

脚本一览（推荐运行顺序）

预期输出（节选 02）

常见报错与依赖