在PostgreSQL存储过程中实现支持查询并行的结果获取与处理

要满足不阻碍查询并行性、不用游标、不创建临时表的要求，最优方案是将查询结果一次性加载为数组类型（行数组或单列数组），直接在内存中遍历处理。这种方式既保留了查询的并行执行能力，又避免了游标或临时表的开销。

核心原理 #

PostgreSQL的ARRAY()构造函数或array_agg()聚合函数可以将整个查询结果集打包成一个数组，底层的SELECT查询完全支持并行执行（只要符合PostgreSQL并行查询的条件，比如表数据量足够大、无并行阻碍构造）。数组会被一次性加载到存储过程的内存变量中，后续直接遍历数组即可处理结果。

具体实现示例 #

1. 处理多行多列结果（行类型数组） #

如果需要处理完整的行数据，可以直接使用表的行类型定义数组，或自定义行类型：

CREATE OR REPLACE PROCEDURE process_active_users()
LANGUAGE plpgsql
AS $$
DECLARE
    -- 使用目标表的行类型定义数组
    active_users users[];
    current_user users;
BEGIN
    -- 一次性加载所有符合条件的行到数组，底层查询可并行执行
    SELECT ARRAY(SELECT u FROM users u WHERE active = true) INTO active_users;

    -- 遍历数组处理每一行
    FOREACH current_user IN ARRAY active_users LOOP
        -- 自定义处理逻辑：示例为更新用户的最后处理时间
        UPDATE users 
        SET last_processed = NOW() 
        WHERE id = current_user.id;
        
        RAISE NOTICE '已处理用户：ID=%，名称=%', current_user.id, current_user.name;
    END LOOP;
END;
$$;

如果需要更灵活的行结构（比如只取部分列），可以先自定义行类型：

-- 自定义行类型
CREATE TYPE user_summary AS (
    user_id integer,
    user_name text,
    email text
);

CREATE OR REPLACE PROCEDURE process_user_summaries()
LANGUAGE plpgsql
AS $$
DECLARE
    user_summaries user_summary[];
    current_summary user_summary;
BEGIN
    SELECT ARRAY(
        SELECT (id, name, email)::user_summary 
        FROM users 
        WHERE active = true
    ) INTO user_summaries;

    FOREACH current_summary IN ARRAY user_summaries LOOP
        -- 处理逻辑示例
        RAISE NOTICE '用户摘要：ID=%，名称=%，邮箱=%', 
            current_summary.user_id, 
            current_summary.user_name, 
            current_summary.email;
    END LOOP;
END;
$$;

2. 处理单列结果（基础类型数组） #

如果只需要处理某一列的数据，使用array_agg()更简洁：

CREATE OR REPLACE PROCEDURE process_user_ids()
LANGUAGE plpgsql
AS $$
DECLARE
    user_ids integer[];
    current_id integer;
BEGIN
    -- 加载所有符合条件的ID到整数数组
    SELECT array_agg(id) INTO user_ids 
    FROM users 
    WHERE active = true;

    -- 遍历数组处理每个ID
    FOREACH current_id IN ARRAY user_ids LOOP
        RAISE NOTICE '正在处理用户ID：%', current_id;
        -- 这里添加具体的处理逻辑
    END LOOP;
END;
$$;

注意事项 #

• 内存限制：如果结果集极大（比如百万级以上行），加载到数组会占用大量内存，可能导致OOM。此时需要评估结果集大小，或考虑分批次处理（但分批次可能会影响并行性，需权衡）。
• 并行查询条件：确保底层SELECT查询符合PostgreSQL并行执行的要求，比如避免使用FOR UPDATE锁、不要调用非并行安全的函数等。
• 性能优势：一次性加载数组减少了数据库与存储过程的交互次数，且并行查询能提升结果获取的速度，整体效率远高于游标逐行处理。

内容的提问来源于stack exchange，提问作者lujop