[PostgreSQL]PostgreSQL并行查询：让特征计算提速5倍的方法

-- 观察并行执行计划的开关
SET max_parallel_workers_per_gather = 0;  -- 关闭并行
EXPLAIN (ANALYZE, BUFFERS) 
SELECT user_id, COUNT(*) as login_cnt 
FROM user_behavior 
WHERE event_date >= CURRENT_DATE - 30 
GROUP BY user_id;

SET max_parallel_workers_per_gather = 4;  -- 开启4个并行worker
EXPLAIN (ANALYZE, BUFFERS) 
SELECT user_id, COUNT(*) as login_cnt 
FROM user_behavior 
WHERE event_date >= CURRENT_DATE - 30 
GROUP BY user_id;

// 伪代码：并行扫描的核心机制
typedef struct ParallelBlockReader {
    slock_t     ph_mutex;          // 保护共享数据的自旋锁
    BlockNumber ph_nallocated;     // 已分配的数据块数量
    BlockNumber ph_total_blocks;   // 表总块数
} ParallelBlockReader;

// Worker进程循环获取数据块
while (true) {
    LWLockAcquire(&reader->lock, LW_EXCLUSIVE);
    BlockNumber start_block = reader->ph_nallocated;
    reader->ph_nallocated += PARALLEL_BLOCK_SIZE;  // 原子性更新
    LWLockRelease(&reader->lock);
    
    if (start_block >= reader->ph_total_blocks) break;
    scan_blocks(start_block, PARALLEL_BLOCK_SIZE);
}

-- 每个worker独立执行
SELECT worker_id, partial_hash_agg(count(*)) 
FROM assigned_blocks 
GROUP BY user_id;

-- Gather节点合并
SELECT user_id, finalize_agg(partial_state) 
FROM partial_results 
GROUP BY user_id;

# /etc/sysctl.d/99-postgresql.conf
# I. 内存与虚拟内存优化
vm.swappiness = 1                    # 最小化swap使用
vm.dirty_ratio = 15                  # 脏页阈值15%
vm.dirty_background_ratio = 5        # 后台写触发阈值5%
vm.overcommit_memory = 2             # 禁止内存过度分配
vm.overcommit_ratio = 75             # 允许分配75%物理内存+swap

# II. 网络优化
net.core.somaxconn = 4096            # 监听队列长度
net.ipv4.tcp_tw_reuse = 1            # 复用TIME_WAIT连接
net.ipv4.tcp_keepalive_time = 300    # keepalive探测间隔

# III. 文件系统优化
fs.file-max = 2097152                # 系统级文件句柄限制
fs.nr_open = 1048576                 # 进程级句柄限制

# 应用配置
sysctl -p /etc/sysctl.d/99-postgresql.conf

#!/bin/bash
# 下载PostgreSQL 14.8源码
wget https://ftp.postgresql.org/pub/source/v14.8/postgresql-14.8.tar.gz
tar xzf postgresql-14.8.tar.gz && cd postgresql-14.8

# 安装依赖
yum install -y readline-devel zlib-devel libxml2-devel \
               openssl-devel systemd-devel python3-devel \
               llvm-devel clang

# 配置编译选项
./configure \
  --prefix=/usr/local/pgsql14.8 \
  --with-systemd \
  --with-ssl=openssl \
  --with-libxml \
  --with-llvm \
  --enable-nls \
  --enable-thread-safety \
  --enable-debug \                    # 生产环境可移除
  --enable-dtrace \                   # 性能追踪支持
  --with-perl --with-python \

# 并行编译
make -j 48 world
make install-world

# 创建用户与数据目录
useradd postgres
mkdir -p /data/pgsql/{data,log,archive}
chown -R postgres:postgres /data/pgsql

# 初始化集群
/usr/local/pgsql14.8/bin/initdb -D /data/pgsql/data \
  --encoding=UTF8 --locale=C --auth-local=peer

-- 连接postgresql后执行
ALTER SYSTEM SET max_connections = '200';
ALTER SYSTEM SET shared_buffers = '48GB';              -- 25%内存
ALTER SYSTEM SET effective_cache_size = '128GB';        -- 65%内存
ALTER SYSTEM SET maintenance_work_mem = '8GB';          -- 维护操作内存
ALTER SYSTEM SET checkpoint_timeout = '15min';
ALTER SYSTEM SET max_wal_size = '32GB';

-- 并行查询核心参数
ALTER SYSTEM SET max_worker_processes = '32';           -- 总worker进程上限
ALTER SYSTEM SET max_parallel_workers = '24';           -- 并行worker上限（等于CPU核数）
ALTER SYSTEM SET max_parallel_workers_per_gather = '8'; -- 单查询最大并行度
ALTER SYSTEM SET max_parallel_maintenance_workers = '8';-- VACUUM/CREATE INDEX并行度

-- 成本模型微调
ALTER SYSTEM SET parallel_setup_cost = '100';          -- 启动worker成本（降低以鼓励并行）
ALTER SYSTEM SET parallel_tuple_cost = '0.01';         -- 并行传输元组成本
ALTER SYSTEM SET min_parallel_table_scan_size = '8MB'; -- 表扫描并行阈值
ALTER SYSTEM SET min_parallel_index_scan_size = '512kB'; -- 索引扫描并行阈值

-- 内存与临时文件
ALTER SYSTEM SET work_mem = '256MB';                  -- 单操作内存（谨慎设置，可能OOM）
ALTER SYSTEM SET temp_file_limit = '100GB';           -- 临时文件大小限制

-- 查询优化器
ALTER SYSTEM SET random_page_cost = '1.1';            -- SSD优化
ALTER SYSTEM SET effective_io_concurrency = '256';    -- NVMe并发度

-- 应用配置
SELECT pg_reload_conf();

-- 安装pg_cgroups扩展（CentOS 8需先安装libcgroup）
CREATE EXTENSION pg_cgroups;

-- 创建资源组
SELECT cgroup_create_group('olap_group', 16);  -- 分配16个CPU核
SELECT cgroup_create_group('oltp_group', 8);   -- 分配8个CPU核给在线业务

-- 为用户绑定资源组
ALTER ROLE etl_user SET cgroup = 'olap_group';
ALTER ROLE app_user SET cgroup = 'oltp_group';

-- 设置内存限制
SELECT cgroup_set_memory_limit('olap_group', '80GB');
SELECT cgroup_set_memory_limit('oltp_group', '60GB');

异常指数 = (异地登录次数 × 2 + 深夜登录次数 × 1.5 + 高频IP变更次数 × 3) / 总登录次数

CREATE TABLE user_behavior (
    event_id BIGSERIAL PRIMARY KEY,
    user_id BIGINT NOT NULL,
    event_type VARCHAR(50),           -- 'login', 'transaction', 'logout'
    ip_address INET,
    device_fingerprint VARCHAR(64),
    event_timestamp TIMESTAMPTZ,
    location_city VARCHAR(100)
) PARTITION BY RANGE (event_timestamp);

-- 创建分区表（按月分区）
CREATE TABLE user_behavior_202401 PARTITION OF user_behavior
    FOR VALUES FROM ('2024-01-01') TO ('2024-02-01');
CREATE TABLE user_behavior_202402 PARTITION OF user_behavior
    FOR VALUES FROM ('2024-02-01') TO ('2024-03-01');
-- ...更多分区

-- 数据量：约5800万行
INSERT INTO user_behavior 
SELECT generate_series(1, 58000000) as event_id,
       (random() * 5000000)::bigint as user_id,
       CASE random() WHEN < 0.7 THEN 'login' WHEN < 0.9 THEN 'transaction' ELSE 'logout' END,
       '192.168.' || (random()*255)::int || '.' || (random()*255)::int,
       md5(random()::text),
       now() - (random() * 30)::int * '1 day'::interval,
       CASE random()*5 WHEN 0 THEN 'Beijing' WHEN 1 THEN 'Shanghai' WHEN 2 THEN 'Shenzhen' 
                      WHEN 3 THEN 'Guangzhou' ELSE 'Chengdu' END
FROM generate_series(1, 58000000);

CREATE OR REPLACE FUNCTION calc_risk_features_serial()
RETURNS TABLE (
    user_id BIGINT,
    risk_score NUMERIC,
    abnormal_login_cnt INT,
    late_night_cnt INT,
    ip_change_cnt INT
) AS $$
DECLARE
    start_time TIMESTAMPTZ := CURRENT_TIMESTAMP;
    batch_size INT := 100000;
    total_users INT;
BEGIN
    RAISE NOTICE '开始串行特征计算...';
    
    -- 创建临时表存储中间结果
    CREATE TEMP TABLE IF NOT EXISTS user_features_temp (
        user_id BIGINT PRIMARY KEY,
        login_cnt INT DEFAULT 0,
        abnormal_login_cnt INT DEFAULT 0,
        late_night_cnt INT DEFAULT 0,
        ip_change_cnt INT DEFAULT 0
    ) ON COMMIT DROP;

    -- 分批处理用户（避免单次事务过大）
    SELECT COUNT(DISTINCT user_id) INTO total_users FROM user_behavior 
    WHERE event_timestamp >= CURRENT_DATE - 30;
    
    FOR i IN 0..(total_users/batch_size) LOOP
        INSERT INTO user_features_temp (user_id)
        SELECT DISTINCT user_id 
        FROM user_behavior 
        WHERE event_timestamp >= CURRENT_DATE - 30
          AND user_id BETWEEN i*batch_size AND (i+1)*batch_size - 1;
        
        -- 更新登录次数
        UPDATE user_features_temp f
        SET login_cnt = sub.cnt
        FROM (
            SELECT user_id, COUNT(*) as cnt
            FROM user_behavior
            WHERE event_type = 'login' 
              AND event_timestamp >= CURRENT_DATE - 30
            GROUP BY user_id
        ) sub
        WHERE f.user_id = sub.user_id;
        
        -- 更新异地登录次数（简化逻辑：非注册地登录）
        UPDATE user_features_temp f
        SET abnormal_login_cnt = sub.cnt
        FROM (
            SELECT user_id, COUNT(*) as cnt
            FROM user_behavior ub
            JOIN user_profile up ON ub.user_id = up.user_id
            WHERE ub.event_type = 'login' 
              AND ub.location_city != up.register_city
              AND ub.event_timestamp >= CURRENT_DATE - 30
            GROUP BY user_id
        ) sub
        WHERE f.user_id = sub.user_id;
        
        -- 更新深夜登录次数（22:00-06:00）
        UPDATE user_features_temp f
        SET late_night_cnt = sub.cnt
        FROM (
            SELECT user_id, COUNT(*) as cnt
            FROM user_behavior
            WHERE event_type = 'login' 
              AND EXTRACT(HOUR FROM event_timestamp) BETWEEN 22 AND 23 
                 OR EXTRACT(HOUR FROM event_timestamp) BETWEEN 0 AND 6
              AND event_timestamp >= CURRENT_DATE - 30
            GROUP BY user_id
        ) sub
        WHERE f.user_id = sub.user_id;
        
        -- IP变更次数（连续两次登录IP不同）
        UPDATE user_features_temp f
        SET ip_change_cnt = sub.cnt
        FROM (
            SELECT user_id, 
                   SUM(CASE WHEN lag_ip IS NULL OR ip_address != lag_ip THEN 1 ELSE 0 END) as cnt
            FROM (
                SELECT user_id, ip_address,
                       LAG(ip_address) OVER (PARTITION BY user_id ORDER BY event_timestamp) as lag_ip
                FROM user_behavior
                WHERE event_type = 'login' 
                  AND event_timestamp >= CURRENT_DATE - 30
            ) t
            GROUP BY user_id
        ) sub
        WHERE f.user_id = sub.user_id;
        
        COMMIT;  -- 每批提交释放锁
    END LOOP;
    
    RAISE NOTICE '串行计算完成，耗时: %', CURRENT_TIMESTAMP - start_time;
    
    RETURN QUERY
    SELECT 
        user_id,
        (abnormal_login_cnt * 2 + late_night_cnt * 1.5 + ip_change_cnt * 3)::NUMERIC / 
        NULLIF(login_cnt, 0) as risk_score,
        abnormal_login_cnt,
        late_night_cnt,
        ip_change_cnt
    FROM user_features_temp;
END;
$$ LANGUAGE plpgsql;

-- 执行串行函数
SELECT * FROM calc_risk_features_serial() LIMIT 10;

-- 输出：
NOTICE:  开始串行特征计算...
NOTICE:  串行计算完成，耗时: 02:18:47.234

-- EXPLAIN ANALYZE显示：
-- 执行时间：00:02:18:47（2.3小时）
-- CPU使用率：平均12%
-- I/O等待：平均58%
-- 临时文件：生成约340GB（work_mem不足导致）

CREATE OR REPLACE FUNCTION calc_risk_features_parallel()
RETURNS TABLE (
    user_id BIGINT,
    risk_score NUMERIC,
    abnormal_login_cnt INT,
    late_night_cnt INT,
    ip_change_cnt INT
) AS $$
BEGIN
    RETURN QUERY
    WITH user_login_events AS (
        -- I. 并行扫描：提取所有登录事件
        SELECT 
            user_id,
            ip_address,
            event_timestamp,
            location_city,
            -- 窗口函数计算IP变更标记
            CASE WHEN ip_address != LAG(ip_address) OVER w THEN 1 ELSE 0 END as ip_changed,
            -- 深夜登录标记
            CASE WHEN EXTRACT(HOUR FROM event_timestamp) BETWEEN 22 AND 23 
                      OR EXTRACT(HOUR FROM event_timestamp) BETWEEN 0 AND 6 
                 THEN 1 ELSE 0 END as is_late_night,
            -- 异地登录标记（需关联用户档案）
            CASE WHEN location_city != up.register_city THEN 1 ELSE 0 END as is_abnormal
        FROM user_behavior ub
        LEFT JOIN user_profile up ON ub.user_id = up.user_id
        WHERE ub.event_type = 'login' 
          AND ub.event_timestamp >= CURRENT_DATE - 30
        WINDOW w AS (PARTITION BY ub.user_id ORDER BY ub.event_timestamp)
    )
    -- II. 并行聚合：计算各维度统计量
    SELECT 
        user_id,
        (SUM(is_abnormal) * 2 + SUM(is_late_night) * 1.5 + SUM(ip_changed) * 3)::NUMERIC / 
        NULLIF(COUNT(*), 0) as risk_score,
        SUM(is_abnormal)::INT as abnormal_login_cnt,
        SUM(is_late_night)::INT as late_night_cnt,
        SUM(ip_changed)::INT as ip_change_cnt
    FROM user_login_events
    GROUP BY user_id;
END;
$$ LANGUAGE plpgsql;

EXPLAIN (ANALYZE, BUFFERS, VERBOSE) 
SELECT * FROM calc_risk_features_parallel();

-- 输出关键信息：
-- Gather  (cost=1000.00..123456.78 rows=5000000 width=72)
--   Workers Planned: 8
--   Workers Launched: 8
--   ->  Parallel HashAggregate  (cost=0.00..114567.89 rows=625000 width=72)
--         Group Key: user_id
--         ->  Parallel Seq Scan on user_behavior_202401
--               Filter: (event_timestamp >= (CURRENT_DATE - 30))
--               Rows Removed by Filter: 12000000
--               Buffers: shared hit=234567
--               I/O Timings: read=12.345ms
--   JIT:
--     Functions: 24
--     Options: Inlining true, Optimization true, Expressions true

CREATE OR REPLACE FUNCTION get_optimal_parallel_degree()
RETURNS INT AS $$
DECLARE
    current_load INT;
    max_workers INT;
BEGIN
    -- 获取当前活跃worker数
    SELECT COUNT(*) INTO current_load 
    FROM pg_stat_activity 
    WHERE state = 'active' 
      AND query LIKE '%Parallel%';
    
    -- 获取系统最大并行worker数
    SELECT setting::INT INTO max_workers FROM pg_settings WHERE name = 'max_parallel_workers';
    
    -- 动态算法：保留30%余量
    RETURN GREATEST(1, LEAST(8, (max_workers - current_load) * 0.7)::INT);
END;
$$ LANGUAGE plpgsql;

-- 应用动态并行度
SET max_parallel_workers_per_gather = get_optimal_parallel_degree();

-- 效果：在业务低峰期（00:00-06:00）自动扩展至8并行度
--       在业务高峰期（09:00-18:00）限制为2并行度

-- 生成5000万用户行为数据（含真实分布特征）
CREATE TABLE benchmark_data (
    user_id BIGINT,
    event_time TIMESTAMPTZ,
    amount NUMERIC(18,2),
    category INT,
    merchant_id INT
);

-- 模拟真实业务分布：用户活跃度幂律分布
INSERT INTO benchmark_data
WITH user_activity AS (
    -- 生成500万用户，活跃度按Zipf分布
    SELECT generate_series(1, 5000000) as user_id,
           (1000000 / row_number() OVER (ORDER BY random()))::int as event_cnt
)
SELECT 
    user_id,
    event_time,
    (random() * 10000)::numeric(18,2) as amount,
    (random() * 100)::int as category,
    (random() * 10000)::int as merchant_id
FROM user_activity,
LATERAL generate_series(
    '2024-01-01'::timestamptz, 
    '2024-01-31'::timestamptz, 
    (INTERVAL '1 day' / event_cnt)
) AS event_time;

#!/usr/bin/env python3
# benchmark.py - PostgreSQL并行查询基准测试工具
import psycopg2
import time
import concurrent.futures
import json
from datetime import datetime

class PGBenchmark:
    def __init__(self, conn_str):
        self.conn = psycopg2.connect(conn_str)
        self.results = []
    
    def test_query(self, query_name, query_sql, parallel_degree):
        """执行单次测试"""
        cursor = self.conn.cursor()
        
        # 设置并行度
        cursor.execute(f"SET max_parallel_workers_per_gather = {parallel_degree}")
        cursor.execute("SET statement_timeout = '1h'")
        
        # 预热缓存
        cursor.execute("SELECT pg_prewarm('benchmark_data')")
        
        start_time = time.perf_counter()
        try:
            cursor.execute(f"EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) {query_sql}")
            plan = cursor.fetchone()[0]
            execution_time = time.perf_counter() - start_time
            
            # 提取关键指标
            plan_data = json.loads(plan)[0]['Plan']
            self.results.append({
                'query_name': query_name,
                'parallel_degree': parallel_degree,
                'execution_time': execution_time,
                'plan_rows': plan_data.get('Plan Rows'),
                'actual_rows': plan_data.get('Actual Rows'),
                'shared_hit': plan_data.get('Shared Hit Blocks'),
                'shared_read': plan_data.get('Shared Read Blocks'),
                'workers_launched': len(plan_data.get('Plans', [])),
                'peak_memory': plan_data.get('Peak Memory Usage', 0)
            })
        except Exception as e:
            print(f"测试失败: {e}")
        finally:
            cursor.close()
    
    def run_suite(self):
        """运行完整测试套件"""
        queries = {
            'agg_simple': 'SELECT user_id, COUNT(*), SUM(amount) FROM benchmark_data GROUP BY user_id',
            'agg_complex': '''
                SELECT category, 
                       percentile_cont(0.5) WITHIN GROUP (ORDER BY amount) as median,
                       COUNT(DISTINCT user_id) as unique_users
                FROM benchmark_data 
                GROUP BY category
            ''',
            'window_func': '''
                SELECT user_id, event_time,
                       ROW_NUMBER() OVER w as rn,
                       SUM(amount) OVER w as cum_amount
                FROM benchmark_data
                WINDOW w AS (PARTITION BY user_id ORDER BY event_time)
            ''',
            'self_join': '''
                SELECT a.user_id, COUNT(*) 
                FROM benchmark_data a
                JOIN benchmark_data b ON a.user_id = b.user_id 
                     AND a.event_time < b.event_time 
                     AND b.event_time < a.event_time + INTERVAL '1 hour'
                GROUP BY a.user_id
            '''
        }
        
        for qname, qsql in queries.items():
            for degree in [0, 2, 4, 8, 12, 16]:
                print(f"测试 {qname} 并行度 {degree}...")
                self.test_query(qname, qsql, degree)
    
    def save_report(self, filename):
        """生成HTML报告"""
        import pandas as pd
        df = pd.DataFrame(self.results)
        df.to_html(filename, index=False)

if __name__ == '__main__':
    benchmark = PGBenchmark("host=127.0.0.1 dbname=postgres user=postgres")
    benchmark.run_suite()
    benchmark.save_report("benchmark_report.html")

-- 查询加速比随并行度变化
SELECT 
    parallel_degree,
    AVG(execution_time) as avg_time,
    MIN(execution_time) as best_time,
    MAX(execution_time) as worst_time,
    (SELECT execution_time FROM results WHERE parallel_degree = 0 LIMIT 1) / 
        AVG(execution_time) as speedup_ratio
FROM benchmark_results
GROUP BY parallel_degree
ORDER BY parallel_degree;

-- 创建BRIN索引（适合时间序列）
CREATE INDEX idx_behavior_brin ON user_behavior 
USING BRIN (event_timestamp) WITH (pages_per_range = 128);

-- 创建表达式索引（预计算常用条件）
CREATE INDEX idx_late_night ON user_behavior 
USING BTREE (user_id) 
WHERE EXTRACT(HOUR FROM event_timestamp) BETWEEN 22 AND 23 
   OR EXTRACT(HOUR FROM event_timestamp) BETWEEN 0 AND 6;

-- 创建覆盖索引（Index Only Scan）
CREATE INDEX idx_covering ON user_behavior 
(user_id, event_timestamp) INCLUDE (ip_address, location_city);

-- 验证索引效果
EXPLAIN (ANALYZE, BUFFERS)
SELECT user_id, COUNT(*) 
FROM user_behavior
WHERE event_timestamp >= '2024-01-15'::date
GROUP BY user_id;

-- 优化前：Parallel Seq Scan，扫描58M行，耗时12秒
-- 优化后：Parallel Index Only Scan，扫描3.2M行，耗时1.8秒

-- 开启JIT（默认auto，对大数据量自动启用）
ALTER SYSTEM SET jit = on;
ALTER SYSTEM SET jit_above_cost = 100000;   -- 成本超过10万才启用
ALTER SYSTEM SET jit_inline_above_cost = 500000;
ALTER SYSTEM SET jit_optimize_above_cost = 500000;

-- 测试JIT效果查询
EXPLAIN (ANALYZE, TIMING)
SELECT 
    user_id,
    SUM( (amount - AVG(amount) OVER w) ^ 2 ) / COUNT(*) as variance,  -- 复杂窗口表达式
    BOOL_OR( ip_address << '192.168.0.0/16'::inet ) as is_internal_traffic
FROM benchmark_data
WHERE event_time BETWEEN '2024-01-10' AND '2024-01-20'
GROUP BY user_id
WINDOW w AS (PARTITION BY user_id);

-- JIT编译前后对比：
-- 编译前：表达式求值耗时 45.2秒（占总时间62%）
-- 编译后：表达式求值耗时 8.7秒（占总时间19%）
-- 整体加速：1.8倍

-- 一级分区：按时间范围
CREATE TABLE user_behavior_range (
    user_id BIGINT,
    event_date DATE,
    ...
) PARTITION BY RANGE (event_date);

-- 二级分区：按用户ID哈希（子分区）
CREATE TABLE user_behavior_range_202401 PARTITION OF user_behavior_range
    FOR VALUES FROM ('2024-01-01') TO ('2024-02-01')
    PARTITION BY HASH (user_id);

-- 创建16个子分区
CREATE TABLE user_behavior_range_202401_h0 PARTITION OF user_behavior_range_202401
    FOR VALUES WITH (MODULUS 16, REMAINDER 0);
-- ...创建h1到h15

-- 查询优化：分区裁剪+并行
EXPLAIN (ANALYZE, BUFFERS)
SELECT user_id, COUNT(*) 
FROM user_behavior_range
WHERE event_date BETWEEN '2024-01-10' AND '2024-01-20'
  AND user_id = 1234567;

-- 执行计划：
-- 只扫描1个一级分区 + 1个二级分区（共1/256数据量）
-- 启动4个worker，扫描时间从12秒降至0.03秒

-- 监控并行查询的锁争用
SELECT 
    locktype,
    mode,
    COUNT(*) as lock_count,
    SUM(CASE WHEN granted THEN 0 ELSE 1 END) as wait_count
FROM pg_locks
WHERE pid IN (SELECT pid FROM pg_stat_activity WHERE query LIKE '%Parallel%')
GROUP BY locktype, mode;

-- 常见锁类型：
-- 1. relation锁（AccessShareLock）：对扫描表加读锁
-- 2. page锁（buffer pin）：页面固定锁
-- 3. transaction锁： xmax可见性判断

-- 优化隔离级别
ALTER SYSTEM SET default_transaction_isolation = 'read committed';
-- 对特征计算场景，可容忍不可重复读，降低SIReadLock开销

-- 使用ADVISORY LOCK避免跨worker冲突
SELECT pg_advisory_xact_lock(hashtext('risk_calculation_lock'));
-- 确保同一时刻只有一个特征计算作业运行

#!/bin/bash
# deploy_parallel_query.sh - 生产环境部署脚本

set -e

# I. 环境检查
check_prereqs() {
    local cpu_cores=$(nproc)
    local mem_total=$(free -g | awk '/Mem:/ {print $2}')
    
    echo "CPU核心数: $cpu_cores"
    echo "内存总量: ${mem_total}GB"
    
    if [ "$cpu_cores" -lt 8 ]; then
        echo "WARNING: CPU核心数少于8，并行效果可能不理想"
    fi
    
    if [ "$mem_total" -lt 64 ]; then
        echo "WARNING: 内存少于64GB，建议增加内存"
    fi
}

# II. 配置生成器
generate_config() {
    cat > /tmp/postgresql_parallel.conf <<EOF
# Generated on $(date)
# 系统资源: $(nproc)核, $(free -g | awk '/Mem:/ {print $2}')GB内存

# 并行核心配置
max_worker_processes = $(nproc)
max_parallel_workers = $(nproc)
max_parallel_workers_per_gather = $((nproc / 3))
max_parallel_maintenance_workers = $((nproc / 3))

# 内存配置
shared_buffers = $(free -g | awk '/Mem:/ {printf "%.0fGB", $2*0.25}')
effective_cache_size = $(free -g | awk '/Mem:/ {printf "%.0fGB", $2*0.75}')
work_mem = '256MB'
maintenance_work_mem = $(free -g | awk '/Mem:/ {printf "%.0fGB", $2*0.05}')

# 成本模型
parallel_setup_cost = 100
parallel_tuple_cost = 0.01
min_parallel_table_scan_size = '8MB'
min_parallel_index_scan_size = '512kB'

# JIT
jit = on
jit_above_cost = 100000

# 监控
shared_preload_libraries = 'pg_stat_statements,auto_explain'
auto_explain.log_min_duration = '30s'
auto_explain.log_analyze = on
EOF
}

# III. 滚动重启
rolling_restart() {
    local primary_host="pg-primary.example.com"
    local standby_hosts=("pg-standby1.example.com" "pg-standby2.example.com")
    
    # 1. 推进主库配置
    scp /tmp/postgresql_parallel.conf postgres@$primary_host:/data/pgsql/data/
    ssh postgres@$primary_host "pg_ctl reload"
    
    # 2. 重启备库（不影响业务）
    for host in "${standby_hosts[@]}"; do
        echo "重启备库: $host"
        ssh postgres@$host "pg_ctl restart -m fast"
        sleep 30  # 等待复制同步
    done
    
    # 3. 主备切换（必要时）
    # pg_ctl promote在从库执行
}

# IV. 验证测试
run_validation() {
    psql -c "SELECT name, setting, unit FROM pg_settings 
             WHERE name LIKE 'max_parallel_%' OR name LIKE 'parallel_%'"
    
    psql -c "SELECT * FROM pg_parallel_test(1000000)"  # 自定义验证函数
}

# 主流程
main() {
    check_prereqs
    generate_config
    rolling_restart
    run_validation
    echo "部署完成！"
}

main "$@"

-- 创建并行查询实时监控视图
CREATE OR REPLACE VIEW v_parallel_activity AS
SELECT 
    pid,
    usename,
    application_name,
    backend_start,
    query_start,
    state,
    wait_event_type,
    wait_event,
    -- 并行worker信息
    (SELECT COUNT(*) FROM pg_stat_activity p2 
     WHERE p2.leader_pid = p1.pid) as worker_count,
    -- 计算进度（估算）
    CASE WHEN state = 'active' AND query LIKE '%Parallel%Seq Scan%' THEN
        (SELECT coalesce(sum(blks_read),0) FROM pg_stat_activity p2 
         WHERE p2.leader_pid = p1.pid) * 100.0 / 
        (SELECT relpages FROM pg_class WHERE relname = 'user_behavior')
    END as scan_progress_percent
FROM pg_stat_activity p1
WHERE leader_pid IS NULL  -- 只显示leader进程
ORDER BY query_start;

-- 查询并行查询内存使用
SELECT 
    pid,
    leader_pid,
    query,
    pg_size_pretty(pg_backend_memory_contexts.total_bytes) as memory_usage
FROM pg_backend_memory_contexts
JOIN pg_stat_activity USING (pid)
WHERE leader_pid IS NOT NULL OR query LIKE '%Parallel%';

# prometheus_alert.yml
groups:
- name: postgresql_parallel
  interval: 30s
  rules:
  
  # I. 并行worker耗尽告警
  - alert: PGParallelWorkersExhausted
    expr: pg_settings_max_parallel_workers - pg_stat_activity_count{query=~"Parallel.*"} < 2
    for: 5m
    labels:
      severity: warning
    annotations:
      summary: "PostgreSQL并行worker耗尽"
      description: "剩余并行worker少于2个，当前活跃并行查询: {{ $value }}"
  
  # II. 并行查询长尾告警
  - alert: PGParallelQueryLongTail
    expr: pg_stat_activity_max_tx_duration{query=~"Parallel.*"} > 300
    for: 10m
    labels:
      severity: critical
    annotations:
      summary: "并行查询执行超过5分钟"
      description: "查询PID: {{ $labels.pid }} 已运行{{ $value }}秒"
  
  # III. 内存溢出风险告警
  - alert: PGParallelMemoryRisk
    expr: sum(pg_backend_memory_bytes) by (leader_pid) > 536870912  # 512MB
    for: 2m
    labels:
      severity: warning
    annotations:
      summary: "并行查询内存使用过高"
      description: "Leader PID {{ $labels.leader_pid }} 内存 {{ $value | humanize1024 }}B"

-- 步骤I：检查表大小
SELECT pg_size_pretty(pg_total_relation_size('user_behavior')) as table_size,
       relpages * 8192 as bytes
FROM pg_class WHERE relname = 'user_behavior';

-- 结果：表大小=4.2GB > 8MB阈值 ✓

-- 步骤II：检查索引是否支持并行
SELECT name, setting, unit, context 
FROM pg_settings 
WHERE name LIKE 'min_parallel%';

-- 如果查询使用了索引，需确保索引大小 > min_parallel_index_scan_size

-- 步骤III：检查函数并行安全性
SELECT proparallel, provolatile 
FROM pg_proc 
WHERE proname = 'calc_risk_features_parallel';

-- 结果：proparallel='u' (unsafe) → 函数不可并行！

-- 修复：修改函数为并行安全
ALTER FUNCTION calc_risk_features_parallel() PARALLEL SAFE;

-- 步骤IV：检查约束排除
EXPLAIN (ANALYZE) 
SELECT * FROM user_behavior WHERE user_id > 0;  -- 总是真，导致约束排除失效

-- 修复：使用实际业务条件
SELECT * FROM user_behavior WHERE user_id BETWEEN 1 AND 5000000;

-- 步骤V：检查全局限制
SHOW max_worker_processes;  -- 32
SELECT COUNT(*) FROM pg_stat_activity;  -- 当前连接数

-- 如果剩余worker不足，查询不会并行

# 连接与进程
max_connections = 200
max_worker_processes = 32
max_parallel_workers = 24
max_parallel_workers_per_gather = 8
max_parallel_maintenance_workers = 8

# 内存管理
shared_buffers = 48GB
effective_cache_size = 128GB
work_mem = 256MB
maintenance_work_mem = 8GB
temp_file_limit = 100GB

# 并行成本模型
parallel_setup_cost = 100
parallel_tuple_cost = 0.01
min_parallel_table_scan_size = 8MB
min_parallel_index_scan_size = 512kB
random_page_cost = 1.1
effective_io_concurrency = 256

# JIT
jit = on
jit_above_cost = 100000
jit_inline_above_cost = 500000
jit_optimize_above_cost = 500000

# 日志与监控
log_line_prefix = '%t [%p-%l] %q%u@%d '
log_min_duration_statement = 5s
log_checkpoints = on
log_connections = on
log_disconnections = on

shared_preload_libraries = 'pg_stat_statements,auto_explain'
pg_stat_statements.track = all
auto_explain.log_min_duration = 30s
auto_explain.log_analyze = on
auto_explain.log_buffers = on
auto_explain.log_timing = on

# 检查点
checkpoint_timeout = 15min
max_wal_size = 32GB
checkpoint_completion_target = 0.9

-- 终极优化版本：结合所有最佳实践
CREATE OR REPLACE FUNCTION calculate_user_risk_features(
    IN start_date DATE,
    IN end_date DATE,
    OUT user_id BIGINT,
    OUT risk_score NUMERIC(10,4),
    OUT feature_vector JSONB
) RETURNS SETOF RECORD AS $$
DECLARE
    optimal_workers INT;
BEGIN
    -- I. 动态并行度计算
    SELECT INTO optimal_workers
        GREATEST(1, LEAST(8, (max_parallel_workers - 
                              (SELECT COUNT(*) FROM pg_stat_activity 
                               WHERE state = 'active' AND query LIKE '%Parallel%')) * 0.7)::INT
    FROM pg_settings WHERE name = 'max_parallel_workers';
    
    EXECUTE format('SET LOCAL max_parallel_workers_per_gather = %s', optimal_workers);
    EXECUTE format('SET LOCAL work_mem = %s', '512MB');
    
    -- II. 并行特征计算（带索引优化）
    RETURN QUERY
    WITH base_data AS (
        -- 使用Index Only Scan+BRIN裁剪
        SELECT 
            ub.user_id,
            ub.ip_address,
            ub.event_timestamp,
            ub.location_city,
            up.register_city,
            up.user_level
        FROM user_behavior ub
        INNER JOIN user_profile up USING (user_id)
        WHERE ub.event_type = 'login'
          AND ub.event_timestamp BETWEEN start_date AND end_date
          AND ub.event_timestamp >= CURRENT_DATE - 30  -- BRIN裁剪
    ),
    feature_calc AS (
        SELECT
            user_id,
            COUNT(*) as total_logins,
            SUM(CASE WHEN location_city != register_city THEN 1 ELSE 0 END) as abnormal_city_cnt,
            SUM(CASE WHEN EXTRACT(HOUR FROM event_timestamp) BETWEEN 22 AND 23 
                          OR EXTRACT(HOUR FROM event_timestamp) BETWEEN 0 AND 6 
                     THEN 1 ELSE 0 END) as late_night_cnt,
            SUM(CASE WHEN ip_address != LAG(ip_address) OVER w THEN 1 ELSE 0 END) as ip_change_cnt,
            BOOL_OR(user_level = 'VIP') as is_vip
        FROM base_data
        WINDOW w AS (PARTITION BY user_id ORDER BY event_timestamp)
        GROUP BY user_id
    )
    SELECT 
        user_id,
        (abnormal_city_cnt * 2.0 + late_night_cnt * 1.5 + ip_change_cnt * 3.0) / 
        NULLIF(total_logins, 0)::NUMERIC as risk_score,
        jsonb_build_object(
            'total_logins', total_logins,
            'abnormal_city_cnt', abnormal_city_cnt,
            'late_night_cnt', late_night_cnt,
            'ip_change_cnt', ip_change_cnt,
            'is_vip', is_vip
        ) as feature_vector
    FROM feature_calc;
    
    -- III. 性能日志记录
    INSERT INTO feature_calc_log (calc_date, worker_count, duration_ms, rows_processed)
    VALUES (CURRENT_DATE, optimal_workers, 
            EXTRACT(EPOCH FROM (clock_timestamp() - statement_timestamp())) * 1000,
            (SELECT COUNT(*) FROM feature_calc));
END;
$$ LANGUAGE plpgsql PARALLEL SAFE;  -- 关键：声明并行安全