如何在PL/pgSQL中动态构建查询生成主键JSON对象？

我正在构建一个PostgreSQL触发器函数，当父表发生插入、更新或删除操作时，向审计表插入一条记录。当前遇到的核心问题是：如何动态生成包含父表主键及其对应值的JSON对象（父表主键数量不固定，但必然存在），并将这个JSON存入审计表的primary_keys列，以此作为父表记录的唯一标识，方便按父表记录保留最近X次操作记录。

父表定义 #

CREATE TABLE parent_table (
    id int4 NOT NULL,
    textfield text NULL,
    id2 int4 NOT NULL,
    CONSTRAINT test_audit_table_pk PRIMARY KEY (id,id2)
);

CREATE TRIGGER t_insert_upate_delete AFTER INSERT OR DELETE OR UPDATE ON
    parent_table FOR EACH ROW EXECUTE FUNCTION audit.audit_trigger();

INSERT INTO parent_table(id, id2, textfield)
VALUES (9, 10, 'TODAY');

解决方案：修改后的完整触发器函数 #

CREATE OR REPLACE FUNCTION audit.audit_trigger()
    RETURNS trigger
    LANGUAGE plpgsql
    SECURITY DEFINER
AS $function$
DECLARE
    audit_table_schema VARCHAR = 'audit';
    audit_table_name VARCHAR = TG_TABLE_SCHEMA || '__' || TG_TABLE_NAME || '_history';
    stmt VARCHAR;
    pk_json jsonb;
    pk_columns text[];
    build_json_stmt VARCHAR;
BEGIN
    -- 自动创建审计表（如果不存在）
    EXECUTE format($$
        CREATE TABLE IF NOT EXISTS %1$I.%2$I (
            tabname text NULL,
            schemaname text NULL,
            operation text NULL,
            new_val jsonb NULL,
            old_val jsonb NULL,
            updated_cols text NULL,
            primary_keys jsonb NULL,
            etl_modified_timestamp timestamptz NULL
        )
    $$, audit_table_schema, audit_table_name);

    -- 动态获取当前表的主键列（按定义顺序）
    SELECT array_agg(kcu.column_name ORDER BY kcu.ordinal_position)
    INTO pk_columns
    FROM information_schema.table_constraints tc
    JOIN information_schema.key_column_usage kcu
        ON tc.constraint_name = kcu.constraint_name
        AND tc.table_schema = kcu.table_schema
    WHERE tc.constraint_type = 'PRIMARY KEY'
        AND tc.table_name = TG_TABLE_NAME
        AND tc.table_schema = TG_TABLE_SCHEMA;

    -- 构建json_build_object的动态参数，生成主键JSON
    build_json_stmt := 'SELECT json_build_object(' || string_agg(format('''%s'', $1.%I', col, col), ', ') || ')::jsonb';
    EXECUTE build_json_stmt INTO pk_json USING CASE TG_OP
        WHEN 'DELETE' THEN OLD
        ELSE NEW
    END;

    -- 根据操作类型插入审计记录
    IF TG_OP = 'INSERT' THEN
        stmt = format($$
            INSERT INTO %1$I.%2$I (
                tabname, schemaname, operation, new_val, primary_keys, etl_modified_timestamp
            )
            VALUES ($1, $2, $3, TO_JSONB($4), $5, CURRENT_TIMESTAMP)
        $$, audit_table_schema, audit_table_name);
        EXECUTE stmt USING TG_TABLE_NAME, TG_TABLE_SCHEMA, TG_OP, NEW, pk_json;

    ELSIF TG_OP = 'UPDATE' THEN
        -- 仅当字段有实际变更时插入
        IF hstore(NEW.*) - hstore(OLD.*) <> ''::hstore THEN
            stmt = format($$
                INSERT INTO %1$I.%2$I (
                    tabname, schemaname, operation, new_val, old_val, updated_cols, primary_keys, etl_modified_timestamp
                )
                VALUES ($1, $2, $3, TO_JSONB($4), TO_JSONB($5), $6, $7, CURRENT_TIMESTAMP)
            $$, audit_table_schema, audit_table_name);
            EXECUTE stmt USING 
                TG_TABLE_NAME, TG_TABLE_SCHEMA, TG_OP, 
                NEW, OLD, 
                array_to_string(akeys(hstore(NEW.*) - hstore(OLD.*)), ', '), 
                pk_json;
        END IF;

    ELSIF TG_OP = 'DELETE' THEN
        stmt = format($$
            INSERT INTO %1$I.%2$I (
                tabname, schemaname, operation, old_val, primary_keys, etl_modified_timestamp
            )
            VALUES ($1, $2, $3, TO_JSONB($4), $5, CURRENT_TIMESTAMP)
        $$, audit_table_schema, audit_table_name);
        EXECUTE stmt USING TG_TABLE_NAME, TG_TABLE_SCHEMA, TG_OP, OLD, pk_json;
    END IF;

    RETURN NULL;
END;
$function$;

关键实现说明 #

1. 动态获取主键列：通过information_schema查询当前表的主键列，按定义顺序存入数组，保证主键顺序和表结构一致。
2. 生成主键JSON：
   • 用string_agg和format动态构建json_build_object的参数列表，例如对id和id2会生成'id', $1.id, 'id2', $1.id2
   • 根据操作类型传入NEW（INSERT/UPDATE）或OLD（DELETE）记录，执行动态SQL得到最终的主键JSON对象。
3. 操作类型适配：
   • INSERT：仅存入新值和主键JSON
   • UPDATE：存入新旧值、变更字段列表和主键JSON，且仅在字段实际变更时插入
   • DELETE：存入旧值和主键JSON
4. 安全与语法保障：使用format函数处理标识符转义，结合USING子句参数化执行，避免SQL注入和语法错误。

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