[PostgreSQL]PostgreSQL算法场景5大性能陷阱与排查清单

-- 步骤1：创建测试表（模拟算法特征表）
DROP TABLE IF EXISTS user_behavior CASCADE;
CREATE TABLE user_behavior (
    behavior_id BIGSERIAL PRIMARY KEY,
    user_id INT NOT NULL,
    item_id INT NOT NULL,
    behavior_type VARCHAR(20) NOT NULL,
    behavior_value DOUBLE PRECISION,
    created_at TIMESTAMP NOT NULL,
    feature_vector VECTOR(768)  -- pgvector扩展，存储embedding
) PARTITION BY RANGE (created_at);

-- 步骤2：创建分区（模拟时间分区表）
CREATE TABLE user_behavior_2024_01 PARTITION OF user_behavior
    FOR VALUES FROM ('2024-01-01') TO ('2024-02-01');
CREATE TABLE user_behavior_2024_02 PARTITION OF user_behavior
    FOR VALUES FROM ('2024-02-01') TO ('2024-03-01');
-- ... 创建更多分区以模拟真实场景

-- 步骤3：生成1000万条测试数据（模拟生产数据量）
INSERT INTO user_behavior (user_id, item_id, behavior_type, behavior_value, created_at, feature_vector)
SELECT 
    (random() * 1000000)::INT as user_id,
    (random() * 500000)::INT as item_id,
    CASE (random() * 3)::INT 
        WHEN 0 THEN 'click'
        WHEN 1 THEN 'purchase'
        WHEN 2 THEN 'favorite'
        ELSE 'share'
    END as behavior_type,
    random() * 100 as behavior_value,
    '2024-01-01'::TIMESTAMP + (random() * 30 || ' days')::INTERVAL as created_at,
    array_fill(random(), ARRAY[768])::VECTOR as feature_vector
FROM generate_series(1, 10000000);

-- 步骤4：创建基本索引（但未创建时间索引）
CREATE INDEX idx_user_behavior_user_id ON user_behavior(user_id);
CREATE INDEX idx_user_behavior_item_id ON user_behavior(item_id);

-- 步骤5：执行典型的算法查询（提取最近7天行为）
EXPLAIN (ANALYZE, BUFFERS, VERBOSE, TIMING)
SELECT user_id, 
       array_agg(item_id ORDER BY created_at) as recent_items,
       avg(behavior_value) as avg_value,
       count(*) as behavior_count
FROM user_behavior
WHERE created_at > NOW() - INTERVAL '7 days'
GROUP BY user_id
HAVING count(*) >= 5
ORDER BY behavior_count DESC
LIMIT 100;

Limit  (cost=123456.78..123457.28 rows=100 width=56) (actual time=28452.123..28452.456 rows=100 loops=1)
  Buffers: shared hit=12345 read=234567
  ->  Sort  (cost=123456.78..123458.78 rows=800 width=56) (actual time=28452.120..28452.345 rows=100 loops=1)
        Sort Key: (count(*)) DESC
        Sort Method: external merge  Disk: 3456kB
        Buffers: shared hit=12345 read=234567
        ->  HashAggregate  (cost=123456.00..123456.50 rows=800 width=56) (actual time=28450.890..28451.234 rows=1234 loops=1)
              Group Key: user_id
              Filter: (count(*) >= 5)
              Buffers: shared hit=12345 read=234567
              ->  Seq Scan on user_behavior_2024_01  (cost=0.00..112345.00 rows=1234567 width=24) (actual time=0.123..12345.678 rows=1234567 loops=1)
                    Filter: (created_at > (now() - '7 days'::interval))
                    Rows Removed by Filter: 8765433
                    Buffers: shared hit=12345 read=234567
Planning Time: 12.345 ms
Execution Time: 28453.789 ms

-- 步骤1：为时间字段创建BRIN索引（适合时序数据）
DROP INDEX IF EXISTS idx_user_behavior_created_at_brin;
CREATE INDEX idx_user_behavior_created_at_brin 
ON user_behavior USING BRIN (created_at) 
WITH (pages_per_range = 128);

-- 步骤2：创建复合索引覆盖查询（索引覆盖扫描）
DROP INDEX IF EXISTS idx_user_behavior_covering;
CREATE INDEX idx_user_behavior_covering 
ON user_behavior (created_at, user_id, item_id, behavior_value);

-- 步骤3：更新统计信息确保优化器选择正确
ANALYZE user_behavior;

-- 步骤4：重新执行查询验证优化效果
EXPLAIN (ANALYZE, BUFFERS, VERBOSE, TIMING)
SELECT user_id, 
       array_agg(item_id ORDER BY created_at) as recent_items,
       avg(behavior_value) as avg_value,
       count(*) as behavior_count
FROM user_behavior
WHERE created_at > NOW() - INTERVAL '7 days'
GROUP BY user_id
HAVING count(*) >= 5
ORDER BY behavior_count DESC
LIMIT 100;

-- 步骤1：创建模拟用户画像表
DROP TABLE IF EXISTS user_profile CASCADE;
CREATE TABLE user_profile (
    user_id BIGSERIAL PRIMARY KEY,
    email VARCHAR(255) NOT NULL,
    phone VARCHAR(20),
    tags JSONB NOT NULL,  -- 存储用户标签，算法常用
    feature_vector VECTOR(128),
    created_at TIMESTAMP DEFAULT NOW()
);

-- 步骤2：插入500万条测试数据
INSERT INTO user_profile (email, phone, tags, feature_vector)
SELECT 
    'user' || i || '@example.COM',  -- 大小写混合
    '138' || LPAD(i::TEXT, 8, '0'),
    jsonb_build_object(
        'age_group', CASE WHEN i % 5 = 0 THEN 'young' ELSE 'adult' END,
        'interest', ARRAY['sports', 'music', 'tech'][1 + (i % 3)]
    ),
    array_fill(random(), ARRAY[128])::VECTOR
FROM generate_series(1, 5000000) i;

-- 步骤3：创建常规索引
CREATE INDEX idx_user_profile_email ON user_profile(email);
CREATE INDEX idx_user_profile_tags_gin ON user_profile USING GIN(tags);

-- 步骤4：算法场景下的问题查询（大小写不敏感匹配）
EXPLAIN (ANALYZE, BUFFERS)
SELECT user_id, email, tags
FROM user_profile
WHERE LOWER(email) = 'user12345@example.com';  -- 索引失效！

-- 步骤5：JSONB查询中的函数问题
EXPLAIN (ANALYZE, BUFFERS)
SELECT user_id, tags
FROM user_profile
WHERE tags->>'age_group' = 'young';  -- GIN索引无法直接支持

-- 步骤6：向量相似度查询中的类型转换问题
EXPLAIN (ANALYZE, BUFFERS)
SELECT user_id, feature_vector <-> array_fill(0.5, ARRAY[128])::VECTOR as distance
FROM user_profile
WHERE (feature_vector <-> array_fill(0.5, ARRAY[128])::VECTOR)::FLOAT < 10.0;  -- 强制转换导致索引失效

-- 查询1：函数索引失效
Seq Scan on user_profile  (cost=0.00..123456.78 rows=1 width=72) 
  (actual time=0.456..1534.123 rows=1 loops=1)
  Filter: (lower(email) = 'user12345@example.com'::text)
  Rows Removed by Filter: 4999999
  Buffers: shared hit=12345 read=43210

-- 查询2：JSONB操作符问题
Seq Scan on user_profile  (cost=0.00..123456.78 rows=500000 width=68)
  (actual time=0.234..2845.678 rows=1000000 loops=1)
  Filter: ((tags ->> 'age_group'::text) = 'young'::text)
  Rows Removed by Filter: 4000000

-- 优化方案1：创建表达式索引（针对大小写问题）
DROP INDEX IF EXISTS idx_user_profile_lower_email;
CREATE INDEX idx_user_profile_lower_email 
ON user_profile (LOWER(email));

-- 验证优化效果
EXPLAIN (ANALYZE, BUFFERS)
SELECT user_id, email, tags
FROM user_profile
WHERE LOWER(email) = 'user12345@example.com';  -- 现在使用索引！

-- 优化方案2：使用JSONB路径索引（针对标签查询）
DROP INDEX IF EXISTS idx_user_profile_tags_path;
CREATE INDEX idx_user_profile_tags_path 
ON user_profile USING GIN ((tags -> 'age_group'));

-- 更优方案：创建覆盖索引
CREATE INDEX idx_user_profile_tags_covering 
ON user_profile USING GIN (tags jsonb_path_ops);

-- 查询改写
EXPLAIN (ANALYZE, BUFFERS)
SELECT user_id, tags
FROM user_profile
WHERE tags @> '{"age_group": "young"}';  -- 使用@>操作符，索引有效

-- 优化方案3：向量查询优化（避免类型转换）
-- 原始问题：WHERE (vector <-> query)::FLOAT < 10.0
-- 优化后：直接比较
EXPLAIN (ANALYZE, BUFFERS)
SELECT user_id, feature_vector <-> array_fill(0.5, ARRAY[128])::VECTOR as distance
FROM user_profile
WHERE feature_vector <-> array_fill(0.5, ARRAY[128])::VECTOR < 10.0;  -- 移除FLOAT转换

-- 优化方案4：使用覆盖索引减少回表
CREATE INDEX idx_user_profile_email_covering 
ON user_profile (LOWER(email)) INCLUDE (user_id, tags);

-- 统计信息更新
ANALYZE user_profile;

-- 步骤1：创建用户画像大表
DROP TABLE IF EXISTS user_profile_big;
CREATE TABLE user_profile_big (
    user_id BIGSERIAL PRIMARY KEY,
    user_vector VECTOR(256),
    preference JSONB,
    update_time TIMESTAMP
);

-- 步骤2：创建商品特征小表
DROP TABLE IF EXISTS item_feature;
CREATE TABLE item_feature (
    item_id SERIAL PRIMARY KEY,
    category_id INT,
    price_band INT,
    item_vector VECTOR(256),
    update_time TIMESTAMP
);

-- 步骤3：填充数据（模拟真实比例）
INSERT INTO user_profile_big (user_vector, preference)
SELECT array_fill(random(), ARRAY[256])::VECTOR,
       jsonb_build_object('category', (random() * 100)::INT)
FROM generate_series(1, 20000000);  -- 2000万用户

INSERT INTO item_feature (category_id, price_band, item_vector)
SELECT (random() * 100)::INT,
       (random() * 10)::INT,
       array_fill(random(), ARRAY[256])::VECTOR
FROM generate_series(1, 500000);  -- 50万商品

-- 步骤4：创建必要的索引
CREATE INDEX idx_user_profile_vector ON user_profile_big USING ivfflat (user_vector);
CREATE INDEX idx_item_feature_category ON item_feature(category_id);
CREATE INDEX idx_item_feature_price ON item_feature(price_band);

-- 步骤5：模拟统计信息失真（复现生产问题）
ALTER TABLE item_feature ALTER COLUMN category_id SET STATISTICS 10;
UPDATE pg_statistic SET stanullfrac = 0.5 WHERE starelid = 'item_feature'::regclass;

-- 步骤6：执行算法推荐查询（问题版本）
EXPLAIN (ANALYZE, BUFFERS, TIMING)
SELECT 
    u.user_id,
    i.item_id,
    (u.user_vector <-> i.item_vector) as similarity,
    (u.preference->>'category')::INT as user_cat
FROM user_profile_big u
JOIN item_feature i ON (u.preference->>'category')::INT = i.category_id
WHERE i.price_band BETWEEN 3 AND 7
LIMIT 100;

-- 步骤7：模拟优化器误判，强制Nested Loop
SET enable_hashjoin = off;
SET enable_mergejoin = off;
EXPLAIN (ANALYZE, BUFFERS, TIMING)
SELECT 
    u.user_id,
    i.item_id,
    (u.user_vector <-> i.item_vector) as similarity
FROM user_profile_big u
JOIN item_feature i ON (u.preference->>'category')::INT = i.category_id
WHERE i.price_band BETWEEN 3 AND 7
LIMIT 100;

Nested Loop  (cost=0.00..123456789.00 rows=100 width=40)
  (actual time=12.345..3245.678 rows=100 loops=1)
  Buffers: shared hit=2345678 read=1234567
  ->  Seq Scan on item_feature i  (cost=0.00..1234.00 rows=100 width=24)
        Filter: ((price_band >= 3) AND (price_band <= 7))
        Rows Removed by Filter: 499900
  ->  Index Scan idx_user_profile_pref on user_profile_big u
        (cost=0.00..1234.56 rows=1 width=32)
        Index Cond: ((preference->>'category'::text)::integer = i.category_id)
        Buffers: shared hit=2345678

-- 关键指标

-- 优化方案1：修正统计信息
ANALYZE item_feature;
ALTER TABLE item_feature ALTER COLUMN category_id SET STATISTICS 1000;

-- 方案2：收集扩展统计信息（针对表达式）
CREATE STATISTICS st_item_feature_category (ndistinct) 
ON category_id FROM item_feature;

ANALYZE item_feature;

-- 优化方案3：调整JOIN相关参数
ALTER DATABASE your_db SET join_collapse_limit = 12;  -- 允许更多重排
ALTER DATABASE your_db SET from_collapse_limit = 12;
ALTER DATABASE your_db SET random_page_cost = 1.1;    -- SSD环境
ALTER DATABASE your_db SET effective_cache_size = '32GB';

-- 方案4：创建更适合JOIN的索引结构
DROP INDEX IF EXISTS idx_user_profile_category_expr;
CREATE INDEX idx_user_profile_category_expr 
ON user_profile_big (((preference->>'category')::INT)) 
INCLUDE (user_vector);

-- 方案5：查询改写，强制优化器选择Hash Join
EXPLAIN (ANALYZE, BUFFERS, TIMING)
SELECT 
    u.user_id,
    i.item_id,
    (u.user_vector <-> i.item_vector) as similarity
FROM user_profile_big u
JOIN item_feature i ON i.category_id = (u.preference->>'category')::INT
WHERE i.price_band BETWEEN 3 AND 7
  AND i.category_id IS NOT NULL  -- 添加谓词帮助优化器估算
LIMIT 100;

-- 方案6：使用CTE物化小表（终极方案）
EXPLAIN (ANALYZE, BUFFERS, TIMING)
WITH filtered_items AS MATERIALIZED (
    SELECT item_id, category_id, item_vector
    FROM item_feature
    WHERE price_band BETWEEN 3 AND 7
)
SELECT 
    u.user_id,
    fi.item_id,
    (u.user_vector <-> fi.item_vector) as similarity
FROM user_profile_big u
JOIN filtered_items fi ON fi.category_id = (u.preference->>'category')::INT
LIMIT 100;

Hash Join  (cost=12345.00..34567.00 rows=100 width=40)
  (actual time=23.456..45.678 rows=100 loops=1)
  Hash Cond: ((u.preference->>'category'::text)::integer = i.category_id)
  Buffers: shared hit=12345
  ->  Seq Scan on user_profile_big u  (cost=0.00..234567.00 rows=20000000 width=32)
  ->  Hash  (cost=123.00..123.00 rows=1000 width=24)
        ->  Bitmap Heap Scan on item_feature i
              Recheck Cond: ((price_band >= 3) AND (price_band <= 7))
              Buffers: shared hit=234

-- 步骤1：创建交易流水表
DROP TABLE IF EXISTS transaction_log CASCADE;
CREATE TABLE transaction_log (
    trans_id BIGSERIAL PRIMARY KEY,
    user_id INT NOT NULL,
    trans_amount NUMERIC(18,2) NOT NULL,
    trans_time TIMESTAMP NOT NULL,
    risk_features JSONB NOT NULL
) PARTITION BY RANGE (trans_time);

-- 步骤2：创建分区
CREATE TABLE transaction_log_2024_01 PARTITION OF transaction_log
    FOR VALUES FROM ('2024-01-01') TO ('2024-02-01');
CREATE TABLE transaction_log_2024_02 PARTITION OF transaction_log
    FOR VALUES FROM ('2024-02-01') TO ('2024-03-01');

-- 步骤3：插入3亿条模拟数据
INSERT INTO transaction_log (user_id, trans_amount, trans_time, risk_features)
SELECT 
    (random() * 5000000)::INT,
    random() * 10000,
    '2024-01-01'::TIMESTAMP + (random() * 60 || ' days')::INTERVAL,
    jsonb_build_object(
        'merchant_id', (random() * 100000)::INT,
        'device_score', random(),
        'location_risk', CASE WHEN random() > 0.9 THEN 'high' ELSE 'low' END
    )
FROM generate_series(1, 300000000);

-- 步骤4：创建基础索引
CREATE INDEX idx_trans_user_id ON transaction_log(user_id);
CREATE INDEX idx_trans_time ON transaction_log(trans_time);

-- 步骤5：模拟低work_mem环境
SET work_mem = '16MB';
SET shared_buffers = '512MB';  -- 模拟配置错误

-- 步骤6：执行风控聚合查询（问题复现）
EXPLAIN (ANALYZE, BUFFERS, TIMING, VERBOSE)
SELECT 
    user_id,
    COUNT(*) as trans_count,
    SUM(trans_amount) as total_amount,
    AVG((risk_features->>'device_score')::FLOAT) as avg_risk_score,
    PERCENTILE_CONT(0.95) WITHIN GROUP (ORDER BY trans_amount) as p95_amount
FROM transaction_log
WHERE trans_time BETWEEN '2024-01-15' AND '2024-01-16'
GROUP BY user_id
HAVING COUNT(*) > 10
ORDER BY total_amount DESC
LIMIT 1000;

-- 步骤7：监控临时文件产生
SELECT datname, temp_files, temp_bytes 
FROM pg_stat_database 
WHERE datname = current_database();

-- 步骤8：查看操作系统级I/O
-- Linux命令：iostat -x 1 观察磁盘写入

Limit  (cost=1234567.89..1234568.89 rows=1000 width=48)
  (actual time=2345678.123..2345679.456 rows=1000 loops=1)
  Buffers: shared hit=12345 read=2345678, temp read=1234567 written=1234567
  ->  Sort  (cost=1234567.89..1234568.89 rows=1000 width=48)
        Sort Key: (sum(trans_amount)) DESC
        Sort Method: external merge  Disk: 12345kB
        Buffers: temp read=1234567 written=1234567
        ->  HashAggregate  (cost=1234500.00..1234567.00 rows=500000 width=48)
              Group Key: user_id
              Filter: (count(*) > 10)
              Buffers: temp read=1234567 written=1234567
              ->  Bitmap Heap Scan on transaction_log_2024_01
                    Recheck Cond: (trans_time BETWEEN '2024-01-15' AND '2024-01-16')
                    Buffers: shared hit=12345 read=2345678

-- 步骤1：查看当前内存配置
SHOW shared_buffers;
SHOW work_mem;
SHOW maintenance_work_mem;
SHOW effective_cache_size;

-- 步骤2：计算推荐值（假设服务器128GB内存）
-- shared_buffers = 32GB (25%)
-- work_mem = 256MB (复杂查询)
-- maintenance_work_mem = 8GB (维护操作)
-- effective_cache_size = 96GB (OS缓存)

-- 步骤3：修改postgresql.conf（需重启）
-- 推荐生产配置：
shared_buffers = 32GB
work_mem = 256MB
maintenance_work_mem = 8GB
effective_cache_size = 96GB
temp_file_limit = 10GB  -- 防止磁盘溢出
max_connections = 200   -- 控制总内存

-- 步骤4：会话级动态调整（针对特定大查询）
SET LOCAL work_mem = '1GB';  -- 仅此事务有效
SET LOCAL temp_file_limit = '5GB';

-- 步骤5：优化查询以减少内存占用
EXPLAIN (ANALYZE, BUFFERS, TIMING)
SELECT user_id, SUM(trans_amount) as total
FROM transaction_log
WHERE trans_time BETWEEN '2024-01-15' AND '2024-01-16'
GROUP BY user_id
HAVING SUM(trans_amount) > 1000
ORDER BY total DESC
LIMIT 1000;

-- 步骤6：创建增量物化视图预聚合
CREATE MATERIALIZED VIEW mv_user_daily_trans AS
SELECT 
    user_id,
    DATE(trans_time) as trans_date,
    COUNT(*) as daily_count,
    SUM(trans_amount) as daily_amount
FROM transaction_log
GROUP BY user_id, DATE(trans_time);

-- 步骤7：创建并发刷新任务
CREATE OR REPLACE FUNCTION refresh_mv_concurrently()
RETURNS void AS $$
BEGIN
    REFRESH MATERIALIZED VIEW CONCURRENTLY mv_user_daily_trans;
END;
$$ LANGUAGE plpgsql;

-- 步骤8：使用分区聚合减少单次计算量
CREATE INDEX idx_trans_user_time ON transaction_log(user_id, trans_time);

SELECT user_id, SUM(daily_amount) as total_amount
FROM mv_user_daily_trans
WHERE trans_date BETWEEN '2024-01-15' AND '2024-01-16'
GROUP BY user_id
HAVING SUM(daily_count) > 10
ORDER BY total_amount DESC
LIMIT 1000;

-- 步骤1：创建实时预测结果表
DROP TABLE IF EXISTS model_predictions CASCADE;
CREATE TABLE model_predictions (
    prediction_id BIGSERIAL PRIMARY KEY,
    request_id UUID NOT NULL UNIQUE,
    model_version VARCHAR(50) NOT NULL,
    user_id INT NOT NULL,
    prediction_result JSONB NOT NULL,
    confidence FLOAT NOT NULL,
    created_at TIMESTAMP DEFAULT NOW(),
    status VARCHAR(20) DEFAULT 'pending'
) PARTITION BY HASH (user_id);

-- 步骤2：创建16个哈希分区
DO $$
BEGIN
    FOR i IN 0..15 LOOP
        EXECUTE format('CREATE TABLE model_predictions_p%s PARTITION OF model_predictions FOR VALUES WITH (MODULUS 16, REMAINDER %s);', i, i);
    END LOOP;
END $$;

-- 步骤3：创建索引（包含唯一约束导致的锁）
CREATE INDEX idx_model_pred_user ON model_predictions(user_id);
CREATE INDEX idx_model_pred_status ON model_predictions(status) WHERE status = 'pending';

-- 步骤4：模拟高并发写入（使用pgbench）
-- 创建测试脚本 write_pred.sql
/*
\set request_id gen_random_uuid()
\set user_id random(1, 1000000)
INSERT INTO model_predictions (request_id, model_version, user_id, prediction_result, confidence)
VALUES (:request_id, 'v2.1', :user_id, '{"label": "fraud", "score": 0.92}', 0.92)
ON CONFLICT (request_id) 
DO UPDATE SET 
    prediction_result = EXCLUDED.prediction_result,
    confidence = EXCLUDED.confidence,
    status = 'updated';
*/

-- 步骤5：执行pgbench压测
-- pgbench -c 50 -j 10 -t 10000 -f write_pred.sql postgres://user:pass@host/db

-- 步骤6：监控锁等待情况（会话1执行长事务）
BEGIN;
UPDATE model_predictions 
SET status = 'processing' 
WHERE user_id = 12345 
  AND status = 'pending';

-- 步骤7：会话2尝试更新相同数据（被阻塞）
-- 在另一个psql会话执行：
BEGIN;
UPDATE model_predictions 
SET status = 'failed' 
WHERE user_id = 12345 
  AND status = 'pending';  -- 将被阻塞

-- 步骤8：监控锁状态
SELECT 
    blocked_locks.pid AS blocked_pid,
    blocked_activity.usename AS blocked_user,
    blocking_locks.pid AS blocking_pid,
    blocking_activity.usename AS blocking_user,
    blocked_activity.query AS blocked_statement,
    blocking_activity.query AS blocking_statement
FROM pg_catalog.pg_locks blocked_locks
JOIN pg_catalog.pg_stat_activity blocked_activity ON blocked_activity.pid = blocked_locks.pid
JOIN pg_catalog.pg_locks blocking_locks 
    ON blocking_locks.locktype = blocked_locks.locktype
    AND blocking_locks.database IS NOT DISTINCT FROM blocked_locks.database
    AND blocking_locks.relation IS NOT DISTINCT FROM blocked_locks.relation
JOIN pg_catalog.pg_stat_activity blocking_activity ON blocking_activity.pid = blocking_locks.pid
WHERE NOT blocked_locks.granted;

-- 步骤9：监控死锁发生
SELECT * FROM pg_stat_database WHERE datname = current_database();

-- 步骤10：模拟死锁场景（两个会话交叉更新）
-- 会话A:
BEGIN;
UPDATE model_predictions SET status = 'A' WHERE user_id = 1;
-- 会话B:
BEGIN;
UPDATE model_predictions SET status = 'B' WHERE user_id = 2;
-- 会话A:
UPDATE model_predictions SET status = 'A' WHERE user_id = 2;  -- 等待B
-- 会话B:
UPDATE model_predictions SET status = 'B' WHERE user_id = 1;  -- 死锁！

-- 优化方案1：使用SKIP LOCKED避免锁等待
CREATE OR REPLACE FUNCTION process_pending_predictions(batch_size INT)
RETURNS TABLE (pid BIGINT, uid INT, result JSONB) AS $$
DECLARE
    rec RECORD;
BEGIN
    RETURN QUERY
    UPDATE model_predictions mp
    SET status = 'processing'
    WHERE mp.prediction_id IN (
        SELECT prediction_id 
        FROM model_predictions 
        WHERE status = 'pending'
        ORDER BY created_at 
        LIMIT batch_size
        FOR UPDATE SKIP LOCKED  -- 关键：跳过已被锁定的行
    )
    RETURNING mp.prediction_id, mp.user_id, mp.prediction_result;
END;
$$ LANGUAGE plpgsql;

-- 优化方案2：使用 advisory lock实现应用层锁
CREATE OR REPLACE FUNCTION acquire_model_lock(user_id INT)
RETURNS BOOLEAN AS $$
DECLARE
    lock_id BIGINT;
BEGIN
    lock_id := hashtext('model_lock_' || user_id);
    PERFORM pg_advisory_lock(lock_id);
    RETURN TRUE;
EXCEPTION
    WHEN lock_not_available THEN
        RETURN FALSE;
END;
$$ LANGUAGE plpgsql;

-- 优化方案3：分区级锁粒度控制
ALTER TABLE model_predictions ENABLE ROW LEVEL SECURITY;

CREATE POLICY policy_partition_isolation ON model_predictions
FOR UPDATE
USING (true)
WITH CHECK (true);

-- 优化方案4：批量操作改为COPY（避免锁）
-- 写入临时表再原子交换
CREATE TEMP TABLE temp_predictions (LIKE model_predictions);
COPY temp_predictions FROM '/data/batch_pred.csv' CSV;
INSERT INTO model_predictions 
SELECT * FROM temp_predictions
ON CONFLICT (request_id) DO UPDATE ...;

-- 优化方案5：调整隔离级别和死锁参数
ALTER DATABASE your_db SET default_transaction_isolation = 'read committed';
ALTER SYSTEM SET deadlock_timeout = '2s';
ALTER SYSTEM SET log_lock_waits = on;
ALTER SYSTEM SET lock_timeout = '10s';

-- 优化方案6：连接池配置优化
-- 在pgbouncer中配置：
-- max_client_conn = 1000
-- default_pool_size = 25
-- reserve_pool_size = 5
-- reserve_pool_timeout = 3

-- 优化方案7：监控与自动告警
CREATE OR REPLACE VIEW v_lock_monitor AS
SELECT 
    now() as snapshot_time,
    (SELECT count(*) FROM pg_locks WHERE NOT granted) as waiting_locks,
    (SELECT count(*) FROM pg_stat_activity WHERE state = 'idle in transaction') as idle_in_txn,
    (SELECT count(*) FROM pg_stat_database_conflicts WHERE datname = current_database()) as conflicts;

-- 创建监控任务
CREATE EXTENSION IF NOT EXISTS pg_cron;
SELECT cron.schedule('lock-monitor', '*/1 * * * *', 
    'INSERT INTO lock_history SELECT * FROM v_lock_monitor');

--监控SQL1：全表扫描告警
CREATE OR REPLACE VIEW v_seq_scan_alert AS
SELECT 
    schemaname, tablename, 
    seq_scan, seq_tup_read, 
    seq_scan > 10000 AND seq_tup_read > 1000000 as is_alert
FROM pg_stat_user_tables
WHERE seq_scan > 0;

--监控SQL2：索引失效检测
CREATE OR REPLACE VIEW v_index_invalid_alert AS
SELECT 
    schemaname, tablename, indexname, 
    idx_scan, pg_size_pretty(pg_relation_size(indexrelid)) as index_size
FROM pg_stat_user_indexes
WHERE idx_scan = 0 AND indexname NOT LIKE 'pg_toast%'
ORDER BY pg_relation_size(indexrelid) DESC
LIMIT 10;

--监控SQL3：临时文件监控
CREATE OR REPLACE VIEW v_tempfile_alert AS
SELECT 
    datname, temp_files, pg_size_pretty(temp_bytes) as temp_size,
    temp_files > 10 OR temp_bytes > 1073741824 as is_alert
FROM pg_stat_database
WHERE datname NOT IN ('template0', 'template1', 'postgres');

--监控SQL4：锁等待监控
CREATE OR REPLACE VIEW v_lock_wait_alert AS
SELECT 
    (SELECT count(*) FROM pg_locks WHERE NOT granted) as waiting_locks,
    (SELECT max(now() - xact_start) FROM pg_stat_activity WHERE waiting) as max_wait_time,
    (SELECT count(*) FROM pg_stat_activity WHERE state = 'active' AND wait_event_type = 'Lock') as waiting_sessions;

--监控SQL5：综合性能视图
CREATE OR REPLACE VIEW v_algorithm_performance_dashboard AS
SELECT 
    '全表扫描' as metric_type,
    count(*) as alert_count,
    string_agg(tablename::TEXT, ', ') as affected_tables
FROM v_seq_scan_alert WHERE is_alert
UNION ALL
SELECT 
    '索引失效',
    count(*),
    string_agg(indexname::TEXT, ', ')
FROM v_index_invalid_alert
UNION ALL
SELECT 
    '临时文件',
    sum(temp_files),
    string_agg(datname::TEXT, ', ')
FROM v_tempfile_alert WHERE is_alert
UNION ALL
SELECT 
    '锁等待',
    waiting_locks,
    'Max wait: ' || max_wait_time::TEXT
FROM v_lock_wait_alert WHERE waiting_locks > 5;

#!/bin/bash
# PostgreSQL算法场景生产部署脚本
# 适用于CentOS 7/8, PostgreSQL 15+, 128GB内存, 32核服务器

set -e

echo "=== 步骤1：系统调优 ==="
# 内核参数优化（/etc/sysctl.conf）
cat >> /etc/sysctl.conf <<EOF
# PostgreSQL算法场景优化
kernel.shmmax = 137438953472
kernel.shmall = 33554432
kernel.sem = 250 32000 100 128
fs.file-max = 7672460
net.ipv4.ip_local_port_range = 9000 65500
vm.swappiness = 1
vm.dirty_background_ratio = 5
vm.dirty_ratio = 40
EOF

sysctl -p

# 资源限制（/etc/security/limits.conf）
cat >> /etc/security/limits.conf <<EOF
postgres soft nofile 65536
postgres hard nofile 65536
postgres soft nproc 65536
postgres hard nproc 65536
postgres soft memlock unlimited
postgres hard memlock unlimited
EOF

echo "=== 步骤2：PostgreSQL安装与配置 ==="
# 安装PGDG源
yum install -y https://download.postgresql.org/pub/repos/yum/reporpms/EL-8-x86_64/pgdg-redhat-repo-latest.noarch.rpm
yum -y install postgresql15-server postgresql15-contrib postgresql15-devel

# 初始化
/usr/pgsql-15/bin/postgresql-15-setup initdb
systemctl enable postgresql-15

# 配置文件修改（/var/lib/pgsql/15/data/postgresql.conf）
cat > /tmp/pg_conf_changes.txt <<EOF
# 算法场景专用配置
shared_buffers = 32GB
work_mem = 256MB
maintenance_work_mem = 8GB
effective_cache_size = 96GB
max_connections = 200
temp_file_limit = 10GB
random_page_cost = 1.1
checkpoint_timeout = 15min
max_wal_size = 16GB
min_wal_size = 4GB
effective_io_concurrency = 200
max_worker_processes = 32
max_parallel_workers_per_gather = 16
max_parallel_workers = 32
max_parallel_maintenance_workers = 8
wal_buffers = 16MB
default_statistics_target = 1000
autovacuum_max_workers = 8
autovacuum_naptime = 30s
track_io_timing = on
shared_preload_libraries = 'pg_stat_statements,pg_prewarm,pg_cron'
pg_stat_statements.track = all
pg_stat_statements.max = 10000
deadlock_timeout = 2s
log_lock_waits = on
lock_timeout = 10s
statement_timeout = 30s
log_min_duration_statement = 1000
log_checkpoints = on
log_autovacuum_min_duration = 0
EOF

# 应用配置
cat /tmp/pg_conf_changes.txt >> /var/lib/pgsql/15/data/postgresql.conf

# pg_hba.conf配置
echo "host    all             all             0.0.0.0/0               scram-sha-256" >> /var/lib/pgsql/15/data/pg_hba.conf

systemctl start postgresql-15

echo "=== 步骤3：创建算法专用数据库 ==="
sudo -u postgres psql <<EOF
CREATE DATABASE algo_platform WITH OWNER postgres ENCODING 'UTF8' LC_COLLATE='en_US.UTF-8';
\c algo_platform

-- 创建扩展
CREATE EXTENSION IF NOT EXISTS pg_stat_statements;
CREATE EXTENSION IF NOT EXISTS pg_cron;
CREATE EXTENSION IF NOT EXISTS vector;  -- pgvector
CREATE EXTENSION IF NOT EXISTS postgis;  -- 空间数据

-- 创建schema
CREATE SCHEMA feature ENGINE;
CREATE SCHEMA model_output;
CREATE SCHEMA monitoring;
EOF

echo "=== 步骤4：表结构设计 ==="
sudo -u postgres psql -d algo_platform <<EOF
-- 用户特征表（分区+向量索引）
CREATE TABLE feature_engine.user_features (
    user_id BIGINT PRIMARY KEY,
    feature_version INT NOT NULL,
    feature_vector VECTOR(768) NOT NULL,
    update_time TIMESTAMP DEFAULT NOW()
) PARTITION BY HASH (user_id);

-- 创建64个分区
DO \$\$
BEGIN
    FOR i IN 0..63 LOOP
        EXECUTE format('CREATE TABLE feature_engine.user_features_p%s PARTITION OF feature_engine.user_features FOR VALUES WITH (MODULUS 64, REMAINDER %s);', i, i);
    END LOOP;
END \$\$;

CREATE INDEX idx_user_features_vector 
ON feature_engine.user_features USING ivfflat (feature_vector) WITH (lists = 1000);

CREATE INDEX idx_user_features_version 
ON feature_engine.user_features (feature_version);

-- 模型预测结果表（时间分区）
CREATE TABLE model_output.predictions (
    prediction_id BIGSERIAL,
    model_version VARCHAR(50) NOT NULL,
    user_id BIGINT NOT NULL,
    item_id BIGINT NOT NULL,
    score FLOAT NOT NULL,
    created_at TIMESTAMP DEFAULT NOW()
) PARTITION BY RANGE (created_at);

-- 预创建分区
CREATE TABLE model_output.predictions_2024_01 PARTITION OF model_output.predictions
    FOR VALUES FROM ('2024-01-01') TO ('2024-02-01');

-- 监控表
CREATE TABLE monitoring.performance_logs (
    log_id BIGSERIAL PRIMARY KEY,
    query_tag VARCHAR(100),
    execution_time_ms INT,
    temp_files INT,
    shared_blks_read INT,
    shared_blks_hit INT,
    created_at TIMESTAMP DEFAULT NOW()
);
EOF

echo "=== 步骤5：部署监控函数 ==="
sudo -u postgres psql -d algo_platform <<EOF
-- 自动性能日志记录
CREATE OR REPLACE FUNCTION monitoring.log_performance(
    p_query_tag TEXT,
    p_execution_time_ms INT,
    p_plan JSONB
)
RETURNS void AS $$
BEGIN
    INSERT INTO monitoring.performance_logs (
        query_tag, execution_time_ms, 
        temp_files, shared_blks_read, shared_blks_hit
    )
    SELECT 
        p_query_tag,
        p_execution_time_ms,
        (p_plan->>'Temp Files Used')::INT,
        (p_plan->>'Shared Blocks Read')::INT,
        (p_plan->>'Shared Blocks Hit')::INT;
END;
$$ LANGUAGE plpgsql;

-- 慢查询自动分析
CREATE OR REPLACE FUNCTION monitoring.analyze_slow_queries()
RETURNS TABLE(query TEXT, plan JSONB) AS $$
BEGIN
    RETURN QUERY
    SELECT query, 
           jsonb_build_object(
               'Temp Files Used', temp_files,
               'Shared Blocks Read', shared_blks_read,
               'Shared Blocks Hit', shared_blks_hit
           ) as plan
    FROM pg_stat_statements
    WHERE mean_time > 1000  -- 平均超过1秒
    ORDER BY total_time DESC
    LIMIT 10;
END;
$$ LANGUAGE plpgsql;
EOF

echo "=== 步骤6：压测验证 ==="
# 准备测试数据
sudo -u postgres psql -d algo_platform <<EOF
INSERT INTO feature_engine.user_features (user_id, feature_version, feature_vector)
SELECT 
    i,
    (random() * 10)::INT,
    array_fill(random(), ARRAY[768])::VECTOR
FROM generate_series(1, 1000000) i;
EOF

# 使用pgbench压测
cat > /tmp/algo_benchmark.sql <<EOF
\set uid random(1, 1000000)
-- 向量相似度查询
SELECT user_id, feature_vector <-> array_fill(0.5, ARRAY[768])::VECTOR as distance
FROM feature_engine.user_features
WHERE feature_version = 5
ORDER BY distance
LIMIT 100;

-- 批量写入预测结果
INSERT INTO model_output.predictions (model_version, user_id, item_id, score)
VALUES ('v2.1', :uid, random()*100000, random())
ON CONFLICT (prediction_id) DO NOTHING;
EOF

# 执行压测
pgbench -c 100 -j 32 -T 300 -f /tmp/algo_benchmark.sql postgres://postgres:password@localhost/algo_platform

echo "=== 部署完成！访问监控看板 ==="
# Grafana导入JSON配置...