AI 时代首选数据库：PostgreSQL 如何以向量扩展与 JSON 能力重塑智能应用

CREATE EXTENSION vector;

CREATE TABLE products (
    id          BIGSERIAL PRIMARY KEY,
    name        TEXT,
    description TEXT,
    embedding   VECTOR(768)  -- 显式指定维度
);

-- 插入示例数据（为演示缩短为 3 维）
INSERT INTO products (name, description, embedding) VALUES
('无线蓝牙耳机', '降噪耳机，续航 30 小时', '[0.12, -0.34, 0.56]'::VECTOR),
('智能运动手表', '心率监测，GPS 定位', '[-0.21, 0.45, -0.67]'::VECTOR),
('4K 无人机', '高清摄像，飞行 40 分钟', '[0.78, -0.12, 0.33]'::VECTOR);

SELECT id, name, 
       1 - (embedding <=> '[0.15, -0.30, 0.50]'::VECTOR) AS cosine_similarity
FROM products
ORDER BY embedding <=> '[0.15, -0.30, 0.50]'::VECTOR
LIMIT 5;

CREATE INDEX products_embedding_hnsw_idx 
ON products 
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 200);

SET hnsw.ef_search = 40;  -- 默认 40，可调至 100~200 提高召回

-- 按 ID 范围分区（示例）
CREATE TABLE products_partitioned (
    LIKE products INCLUDING ALL
) PARTITION BY RANGE (id);

CREATE TABLE products_p1 PARTITION OF products_partitioned
    FOR VALUES FROM (1) TO (10000000);
-- 在每个分区上独立创建 HNSW 索引

-- 并行查询会自动下推
SET max_parallel_workers_per_gather = 4;
EXPLAIN (COSTS OFF) 
SELECT * FROM products_partitioned 
ORDER BY embedding <=> '[0.1,0.2,0.3]'::VECTOR LIMIT 10;

ALTER TABLE products ADD COLUMN attributes JSONB;

UPDATE products SET attributes = '{
    "price": 299.99,
    "stock": 150,
    "tags": ["electronics", "audio"],
    "release_date": "2026-01-15"
}' WHERE id = 1;

SELECT id, name, attributes->>'price' AS price,
       1 - (embedding <=> '[0.12, -0.34, 0.56]'::VECTOR) AS sim
FROM products
WHERE (attributes->>'price')::NUMERIC < 500
ORDER BY embedding <=> '[0.12, -0.34, 0.56]'::VECTOR
LIMIT 10;

CREATE INDEX idx_products_price ON products (( (attributes->>'price')::NUMERIC ));
CREATE INDEX idx_products_tags ON products USING GIN ((attributes->'tags'));

ALTER TABLE products ADD COLUMN text_search TSVECTOR 
GENERATED ALWAYS AS (to_tsvector('english', description)) STORED;

CREATE INDEX idx_products_tsv ON products USING GIN (text_search);

SELECT id, name, description,
       ts_rank(text_search, to_tsquery('english', '降噪')) AS rank,
       1 - (embedding <=> '[0.12, -0.34, 0.56]'::VECTOR) AS sim
FROM products
WHERE text_search @@ to_tsquery('english', '降噪')
ORDER BY (1 - (embedding <=> '[0.12, -0.34, 0.56]'::VECTOR)) DESC
LIMIT 20;

SELECT id, name,
       0.7 * (1 - (embedding <=> '[0.12, -0.34, 0.56]'::VECTOR)) 
       + 0.3 * ts_rank(text_search, to_tsquery('english', '降噪')) AS hybrid_score
FROM products
WHERE text_search @@ to_tsquery('english', '降噪')  -- 可放宽为 OR 条件
ORDER BY hybrid_score DESC
LIMIT 10;

CREATE EXTENSION file_fdw;

CREATE SERVER csv_server FOREIGN DATA WRAPPER file_fdw;

CREATE FOREIGN TABLE training_samples (
    sample_id INTEGER,
    feature_vector VECTOR(128),
    label TEXT
) SERVER csv_server
OPTIONS ( filename '/data/train.csv', format 'csv' );

CREATE EXTENSION postgres_fdw;

CREATE SERVER remote_pg FOREIGN DATA WRAPPER postgres_fdw
OPTIONS (host 'xxx.tencentcdb.com', port '5432', dbname 'analytics');

CREATE USER MAPPING FOR CURRENT_USER SERVER remote_pg
OPTIONS (user 'user', password 'pass');

CREATE FOREIGN TABLE remote_embeddings (
    id BIGINT,
    emb VECTOR(768)
) SERVER remote_pg OPTIONS (schema_name 'public', table_name 'embeddings');

-- 本地 JOIN 远程向量
SELECT local.*, remote.emb
FROM local_table local
JOIN remote_embeddings remote ON local.id = remote.id
ORDER BY remote.emb <=> '[0.1, ...]'::VECTOR LIMIT 10;

-- 增加共享缓冲区（建议系统内存的 25%）
ALTER SYSTEM SET shared_buffers = '8GB';

-- 工作内存（影响排序、哈希、索引构建）
ALTER SYSTEM SET work_mem = '64MB';

-- 并行查询相关
ALTER SYSTEM SET max_parallel_workers = 16;
ALTER SYSTEM SET parallel_tuple_cost = 0.001;
ALTER SYSTEM SET parallel_setup_cost = 5.0;

-- 对于 HNSW 索引构建，临时提高 maintenance_work_mem
SET maintenance_work_mem = '4GB';

CREATE INDEX products_embedding_ivfflat_idx 
ON products 
USING ivfflat (embedding vector_cosine_ops)
WITH (lists = 100);  -- 聚类中心数，约为 sqrt(行数)

-- 查询时调整 probes（默认 1）
SET ivfflat.probes = 10;  -- 增加召回率

EXPLAIN (ANALYZE, BUFFERS) 
SELECT * FROM products 
ORDER BY embedding <=> '[0.12, -0.34, 0.56]'::VECTOR 
LIMIT 10;

-- 假设已获取用户查询向量 :query_vec TEXT
-- 价格范围 :min_price, :max_price
WITH filtered AS (
    SELECT id, name, description, price,
           1 - (embedding <=> :query_vec::VECTOR) AS sim
    FROM products
    WHERE price BETWEEN :min_price AND :max_price
      AND attributes->>'brand' = 'Nike'  -- 精确过滤
)
SELECT * FROM filtered
WHERE sim > 0.75   -- 相似度阈值
ORDER BY sim DESC
LIMIT 20;

CREATE EXTENSION pg_cron;

-- 每 5 分钟刷新热门结果
SELECT cron.schedule('refresh-hot', '*/5 * * * *', 
    'REFRESH MATERIALIZED VIEW CONCURRENTLY hot_recommendations');