2026 年了，直接用 PostgreSQL 吧

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

-- 使用 BM25 的全文搜索
CREATE EXTENSION pg_textsearch;

-- AI 向量搜索
CREATE EXTENSION vector;
CREATE EXTENSION vectorscale;

-- AI 嵌入和 RAG 工作流
CREATE EXTENSION ai;

-- 时序数据
CREATE EXTENSION timescaledb;

-- 消息队列
CREATE EXTENSION pgmq;

-- 定时任务
CREATE EXTENSION pg_cron;

-- 地理空间
CREATE EXTENSION postgis;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

-- 创建表
CREATE TABLE articles (
  id SERIAL PRIMARY KEY,
  title TEXT,
  content TEXT
);

-- 创建 BM25 索引
CREATE INDEX idx_articles_bm25 ON articles 
USING bm25(content) WITH (text_config = 'english');

-- 使用 BM25 评分搜索
SELECT title, -(content <@> 'database optimization') as score
FROM articles
ORDER BY content <@> 'database optimization'
LIMIT 10;

1
2
3
4
5
6
7
8
9

SELECT 
  title,
  -(content <@> 'database optimization') as bm25_score,
  embedding <=> query_embedding as vector_distance,
  0.7 * (-(content <@> 'database optimization')) + 
  0.3 * (1 - (embedding <=> query_embedding)) as hybrid_score
FROM articles
ORDER BY hybrid_score DESC
LIMIT 10;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

-- 启用扩展
CREATE EXTENSION vector;
CREATE EXTENSION vectorscale CASCADE;

-- 带嵌入的表
CREATE TABLE documents (
  id SERIAL PRIMARY KEY,
  content TEXT,
  embedding vector(1536)
);

-- 高性能索引（DiskANN）
CREATE INDEX idx_docs_embedding ON documents 
USING diskann(embedding);

-- 查找相似文档
SELECT content, embedding <=> '[0.1, 0.2, ...]'::vector as distance
FROM documents
ORDER BY embedding <=> '[0.1, 0.2, ...]'::vector
LIMIT 10;

1
2
3
4
5
6
7
8

SELECT ai.create_vectorizer(
  'documents'::regclass,
  loading => ai.loading_column(column_name=>'content'),
  embedding => ai.embedding_openai(
    model=>'text-embedding-3-small', 
    dimensions=>'1536'
  )
);

 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

-- 启用 TimescaleDB
CREATE EXTENSION timescaledb;

-- 创建表
CREATE TABLE metrics (
  time TIMESTAMPTZ NOT NULL,
  device_id TEXT,
  temperature DOUBLE PRECISION
);

-- 转换为超表
SELECT create_hypertable('metrics', 'time');

-- 使用时间桶查询
SELECT 
  time_bucket('1 hour', time) as hour,
  AVG(temperature)
FROM metrics
WHERE time > NOW() - INTERVAL '24 hours'
GROUP BY hour;

-- 自动删除旧数据
SELECT add_retention_policy('metrics', INTERVAL '30 days');

-- 压缩（减少 90% 存储）
ALTER TABLE metrics SET (timescaledb.compress);
SELECT add_compression_policy('metrics', INTERVAL '7 days');

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

-- UNLOGGED = 无 WAL 开销，更快的写入
CREATE UNLOGGED TABLE cache (
  key TEXT PRIMARY KEY,
  value JSONB,
  expires_at TIMESTAMPTZ
);

-- 设置过期时间
INSERT INTO cache (key, value, expires_at)
VALUES ('user:123', '{"name": "Alice"}', NOW() + INTERVAL '1 hour')
ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value;

-- 获取
SELECT value FROM cache 
WHERE key = 'user:123' AND expires_at > NOW();

-- 清理（使用 pg_cron 调度）
DELETE FROM cache WHERE expires_at < NOW();

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

CREATE EXTENSION pgmq;
SELECT pgmq.create('my_queue');

-- 发送
SELECT pgmq.send('my_queue', '{"event": "signup", "user_id": 123}');

-- 接收（带可见性超时）
SELECT * FROM pgmq.read('my_queue', 30, 5);

-- 处理后删除
SELECT pgmq.delete('my_queue', msg_id);

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

CREATE TABLE jobs (
  id SERIAL PRIMARY KEY,
  payload JSONB,
  status TEXT DEFAULT 'pending'
);

-- Worker 原子性地声明任务
UPDATE jobs SET status = 'processing'
WHERE id = (
  SELECT id FROM jobs WHERE status = 'pending'
  FOR UPDATE SKIP LOCKED LIMIT 1
) RETURNING *;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

CREATE TABLE users (
  id SERIAL PRIMARY KEY,
  data JSONB
);

-- 插入嵌套文档
INSERT INTO users (data) VALUES ('{
  "name": "Alice",
  "profile": {
    "bio": "Developer", 
    "links": ["github.com/alice"]
  }
}');

-- 查询嵌套字段
SELECT 
  data->>'name', 
  data->'profile'->>'bio'
FROM users
WHERE data->'profile'->>'bio' LIKE '%Developer%';

-- 索引 JSON 字段
CREATE INDEX idx_users_email ON users ((data->>'email'));

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

CREATE EXTENSION postgis;

CREATE TABLE stores (
  id SERIAL PRIMARY KEY,
  name TEXT,
  location GEOGRAPHY(POINT, 4326)
);

-- 查找 5 公里内的商店
SELECT 
  name, 
  ST_Distance(
    location, 
    ST_MakePoint(-122.4, 37.78)::geography
  ) as meters
FROM stores
WHERE ST_DWithin(
  location, 
  ST_MakePoint(-122.4, 37.78)::geography, 
  5000
);

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

CREATE EXTENSION pg_cron;

-- 每小时运行
SELECT cron.schedule('cleanup', '0 * * * *', 
  $$DELETE FROM cache WHERE expires_at < NOW()$$
);

-- 每晚汇总
SELECT cron.schedule('rollup', '0 2 * * *',
  $$REFRESH MATERIALIZED VIEW CONCURRENTLY daily_stats$$
);

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

-- 倒数排名融合：结合关键词 + 语义搜索
WITH bm25 AS (
  SELECT id, ROW_NUMBER() OVER (ORDER BY content <@> $1) as rank
  FROM documents LIMIT 20
),
vectors AS (
  SELECT id, ROW_NUMBER() OVER (ORDER BY embedding <=> $2) as rank  
  FROM documents LIMIT 20
)
SELECT 
  d.*, 
  1.0/(60 + COALESCE(b.rank, 1000)) + 
  1.0/(60 + COALESCE(v.rank, 1000)) as score
FROM documents d
LEFT JOIN bm25 b ON d.id = b.id
LEFT JOIN vectors v ON d.id = v.id
WHERE b.id IS NOT NULL OR v.id IS NOT NULL
ORDER BY score DESC 
LIMIT 10;

1
2
3
4
5
6
7
8
9

CREATE EXTENSION pg_trgm;

CREATE INDEX idx_name_trgm ON products 
USING GIN (name gin_trgm_ops);

-- 即使有拼写错误也能找到 "PostgreSQL"
SELECT name FROM products
WHERE name % 'posgresql'
ORDER BY similarity(name, 'posgresql') DESC;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

-- 查找经理下的所有下属（组织架构）
WITH RECURSIVE org_tree AS (
  SELECT id, name, manager_id, 1 as depth
  FROM employees WHERE id = 42
  
  UNION ALL
  
  SELECT e.id, e.name, e.manager_id, t.depth + 1
  FROM employees e
  JOIN org_tree t ON e.manager_id = t.id
  WHERE t.depth < 10
)
SELECT * FROM org_tree;

1
2
3
4
5
6
7
8
9

-- 使用 pgBouncer 进行连接池
# pgbouncer.ini
[databases]
mydb = host=localhost port=5432 dbname=mydb

[pgbouncer]
pool_mode = transaction
max_client_conn = 1000
default_pool_size = 25

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

-- 分析查询性能
EXPLAIN (ANALYZE, BUFFERS) SELECT ...;

-- 创建适当的索引
CREATE INDEX CONCURRENTLY idx_users_email ON users(email);

-- 使用部分索引
CREATE INDEX idx_active_users ON users(email) 
WHERE status = 'active';

-- 表达式索引
CREATE INDEX idx_lower_email ON users(LOWER(email));

 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

-- 慢查询监控
SELECT 
  query,
  calls,
  total_exec_time,
  mean_exec_time,
  max_exec_time
FROM pg_stat_statements
ORDER BY mean_exec_time DESC
LIMIT 10;

-- 表膨胀检查
SELECT 
  schemaname,
  tablename,
  pg_size_pretty(pg_total_relation_size(schemaname||'.'||tablename)) AS size
FROM pg_tables
ORDER BY pg_total_relation_size(schemaname||'.'||tablename) DESC;

-- 连接数监控
SELECT 
  datname,
  count(*) as connections,
  max_conn - count(*) as available
FROM pg_stat_activity, 
  (SELECT setting::int as max_conn FROM pg_settings WHERE name='max_connections') mc
GROUP BY datname, max_conn;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

# 逻辑备份
pg_dump -Fc mydb > backup.dump

# 恢复
pg_restore -d mydb backup.dump

# 物理备份（使用 pg_basebackup）
pg_basebackup -D /backup/dir -Ft -z -P

# 时间点恢复（PITR）配置
# postgresql.conf
wal_level = replica
archive_mode = on
archive_command = 'cp %p /archive/%f'

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

-- Primary 配置
# postgresql.conf
wal_level = replica
max_wal_senders = 3
wal_keep_size = 1GB

# pg_hba.conf
host replication replicator standby_ip/32 md5

-- Standby 配置
# standby.signal 文件
standby_mode = 'on'
primary_conninfo = 'host=primary_ip port=5432 user=replicator'

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

-- 创建只读用户
CREATE ROLE readonly_user LOGIN PASSWORD 'strong_password';
GRANT CONNECT ON DATABASE mydb TO readonly_user;
GRANT USAGE ON SCHEMA public TO readonly_user;
GRANT SELECT ON ALL TABLES IN SCHEMA public TO readonly_user;

-- 行级安全（RLS）
CREATE POLICY user_data_policy ON users
  FOR SELECT
  USING (user_id = current_user_id());

ALTER TABLE users ENABLE ROW LEVEL SECURITY;

-- SSL 连接
# postgresql.conf
ssl = on
ssl_cert_file = '/path/to/server.crt'
ssl_key_file = '/path/to/server.key'

 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

-- 1. 安装扩展
CREATE EXTENSION pg_textsearch;

-- 2. 创建等效模式
CREATE TABLE products (
  id SERIAL PRIMARY KEY,
  name TEXT,
  description TEXT,
  price NUMERIC
);

-- 3. 创建搜索索引
CREATE INDEX idx_products_search ON products 
USING bm25(description) WITH (text_config = 'english');

-- 4. 数据迁移
-- 使用 Elasticsearch 的 scroll API 导出数据
-- 然后批量导入到 Postgres

-- 5. 更新应用代码
-- 之前: ES query DSL
-- 现在: 简单的 SQL
SELECT * FROM products
WHERE description @@ 'laptop'
ORDER BY -(description <@> 'laptop')
LIMIT 10;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

-- 1. 创建带 JSONB 的表
CREATE TABLE orders (
  id SERIAL PRIMARY KEY,
  data JSONB,
  created_at TIMESTAMPTZ DEFAULT NOW()
);

-- 2. 创建索引
CREATE INDEX idx_orders_user ON orders ((data->>'user_id'));
CREATE INDEX idx_orders_status ON orders ((data->>'status'));

-- 3. 数据导出和导入
-- mongoexport --db mydb --collection orders --out orders.json
-- 然后使用 Python/Go 脚本转换并导入

-- 4. 查询模式转换
-- MongoDB: db.orders.find({"user_id": "123", "status": "pending"})
-- Postgres:
SELECT * FROM orders
WHERE data->>'user_id' = '123' 
  AND data->>'status' = 'pending';

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

-- 1. 创建缓存表
CREATE UNLOGGED TABLE cache (
  key TEXT PRIMARY KEY,
  value JSONB,
  expires_at TIMESTAMPTZ
);

-- 2. 创建索引
CREATE INDEX idx_cache_expires ON cache(expires_at) 
WHERE expires_at IS NOT NULL;

-- 3. 设置自动清理
SELECT cron.schedule('cache_cleanup', '*/5 * * * *',
  $$DELETE FROM cache WHERE expires_at < NOW()$$
);

-- 4. 应用代码转换
-- Redis: redis.setex('user:123', 3600, json_data)
-- Postgres:
INSERT INTO cache (key, value, expires_at)
VALUES ('user:123', $1, NOW() + INTERVAL '1 hour')
ON CONFLICT (key) DO UPDATE SET 
  value = EXCLUDED.value,
  expires_at = EXCLUDED.expires_at;