PL/pgSQL跨schema执行动态查询 实现未知列数结果存入临时表

PostgreSQL的PL/pgSQL函数默认要求返回结构静态定义，针对你需要的动态返回列的场景，有两种可行实现方案：

方案1：返回动态记录集（调用时指定结构） #

该方案直接返回结果集，调用时需要手动声明返回列的结构：

CREATE OR REPLACE FUNCTION pg_temp.select_all(query text)
RETURNS SETOF RECORD AS $$
DECLARE
    v_schema text;
    v_union_query text := '';
BEGIN
    -- 拼接所有schema的UNION ALL语句
    FOR v_schema IN (
        SELECT schema_name
        FROM information_schema.schemata
        WHERE schema_name IN (SELECT login FROM cdu.nc_tenant)
    ) LOOP
        IF v_union_query <> '' THEN
            v_union_query := v_union_query || ' UNION ALL ';
        END IF;
        v_union_query := v_union_query || format('SELECT %L AS schema, * FROM (%s) AS t', v_schema, query);
    END LOOP;

    -- 返回动态结果集
    RETURN QUERY EXECUTE v_union_query;
END; $$
LANGUAGE plpgsql;

使用示例： #

-- 统计sku数量的查询
SELECT * FROM pg_temp.select_all('SELECT count(1) FROM sku') 
AS t(schema text, count bigint);

-- 查询租户变量的查询
SELECT * FROM pg_temp.select_all('SELECT variable, value FROM cdu.nc_tenant_variables where variable = ''theme''')
AS t(schema text, variable text, value text);

方案2：自动生成临时表存储结果（更易用，无需提前指定结构） #

该方案会自动创建临时表存储合并后的结果，调用后直接查询临时表即可，不需要提前声明返回结构，更符合你的使用习惯：

CREATE OR REPLACE FUNCTION pg_temp.select_all(query text)
RETURNS VOID AS $$
DECLARE
    v_schema text;
    v_first_schema boolean := true;
    v_temp_table text := 'cross_schema_result';
BEGIN
    -- 先清理已存在的临时表
    EXECUTE format('DROP TABLE IF EXISTS %I', v_temp_table);

    -- 遍历所有符合条件的schema
    FOR v_schema IN (
        SELECT schema_name
        FROM information_schema.schemata
        WHERE schema_name IN (SELECT login FROM cdu.nc_tenant)
    ) LOOP
        IF v_first_schema THEN
            -- 第一个schema的执行结果用来创建临时表结构
            EXECUTE format(
                'CREATE TEMP TABLE %I AS SELECT %L AS schema, * FROM (%s) AS t',
                v_temp_table,
                v_schema,
                query
            );
            v_first_schema := false;
        ELSE
            -- 后续schema的结果直接插入临时表
            EXECUTE format(
                'INSERT INTO %I SELECT %L AS schema, * FROM (%s) AS t',
                v_temp_table,
                v_schema,
                query
            );
        END IF;
    END LOOP;
END; $$
LANGUAGE plpgsql;

使用示例： #

-- 执行任意查询
SELECT pg_temp.select_all('SELECT count(1) FROM sku');
-- 直接查询临时表获取合并结果
SELECT * FROM pg_temp.cross_schema_result;

注意事项 #

• 所有目标schema执行传入的查询返回的列数量、列类型、列名必须完全一致，否则会执行报错
• 传入的查询语句中不要包含名为schema的列，避免和首列冲突
• 如果需要避免SQL注入风险，可以对传入的query参数做额外的合法性校验

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