如何基于审计表编写SQL将指定br_id的业务数据回滚到目标时间点

核心逻辑是：对每一张关联br_id的业务表，先找到目标审计时间点之前、该br_id对应的最新一条审计记录，用这条记录的字段值覆盖业务表中的对应数据；如果目标时间点之前没有该br_id的审计记录，说明该时间点该br_id还未创建，直接删除业务表中该br_id的对应数据即可。

• 确认所有审计表结构符合要求：都包含br_id字段、audit_dtme时间戳字段、operation字段（一般为INSERT/UPDATE/DELETE），以及和对应业务表完全一致的业务字段
• 操作前必须先备份当前所有业务表中目标br_id的全量数据，避免回滚错误导致数据丢失
• 建议先在测试环境用测试数据验证逻辑正确性，再到生产环境执行

步骤1：封装单表回滚通用函数 #

可以写一个PL/pgSQL函数，传入业务表名、审计表名、目标br_id、目标审计时间，自动完成单表的回滚操作，示例代码如下：

CREATE OR REPLACE FUNCTION rollback_br_table(
    p_br_id INT,
    p_target_time TIMESTAMPTZ,
    p_biz_table TEXT,
    p_audit_table TEXT
) RETURNS VOID AS $$
DECLARE
    v_has_record BOOLEAN;
    v_sql TEXT;
BEGIN
    -- 先判断目标时间点之前是否存在该br_id的审计记录
    v_sql := format('SELECT EXISTS(SELECT 1 FROM %I WHERE br_id = $1 AND audit_dtme <= $2)', p_audit_table);
    EXECUTE v_sql INTO v_has_record USING p_br_id, p_target_time;

    IF NOT v_has_record THEN
        -- 目标时间点该br_id还未创建，删除业务表中对应数据
        v_sql := format('DELETE FROM %I WHERE br_id = $1', p_biz_table);
        EXECUTE v_sql USING p_br_id;
        RETURN;
    END IF;

    -- 找到目标时间点前最新的审计记录，更新业务表（仅当最新操作不是删除时）
    v_sql := format('
        UPDATE %I b
        SET (col1, col2, col3) = (a.col1, a.col2, a.col3) -- 此处替换为对应业务表的所有业务字段
        FROM (
            SELECT DISTINCT ON (br_id) * 
            FROM %I 
            WHERE br_id = $1 AND audit_dtme <= $2
            ORDER BY br_id, audit_dtme DESC
        ) a
        WHERE b.br_id = a.br_id
        AND a.operation != ''DELETE'';
    ', p_biz_table, p_audit_table);
    EXECUTE v_sql USING p_br_id, p_target_time;

    -- 如果最新审计记录是删除操作，删除业务表对应数据
    v_sql := format('
        DELETE FROM %I b
        USING (
            SELECT DISTINCT ON (br_id) operation
            FROM %I 
            WHERE br_id = $1 AND audit_dtme <= $2
            ORDER BY br_id, audit_dtme DESC
        ) a
        WHERE b.br_id = $1 AND a.operation = ''DELETE'';
    ', p_biz_table, p_audit_table);
    EXECUTE v_sql USING p_br_id, p_target_time;
END;
$$ LANGUAGE plpgsql;

提示：也可以通过查询Postgres的information_schema.columns系统表自动拼接业务字段，实现完全通用的无需手动改字段的函数。

步骤2：批量执行所有相关表的回滚 #

把所有业务表和对应的审计表配对，调用上面的函数即可，所有操作建议放在事务中执行：

BEGIN;
-- 先备份当前目标br_id的所有数据，示例br_id=123
CREATE TABLE br_tbl_1_backup_20240520 AS SELECT * FROM br_tbl_1 WHERE br_id = 123;
CREATE TABLE br_tbl_2_backup_20240520 AS SELECT * FROM br_tbl_2 WHERE br_id = 123;

-- 执行回滚，示例目标时间为2024-05-01 12:00:00东八区时间
SELECT rollback_br_table(123, '2024-05-01 12:00:00+08'::TIMESTAMPTZ, 'br_tbl_1', 'br_tbl_1_audit');
SELECT rollback_br_table(123, '2024-05-01 12:00:00+08'::TIMESTAMPTZ, 'br_tbl_2', 'br_tbl_2_audit');
-- 其他关联表依次调用即可

-- 校验回滚结果无误后再提交，有问题直接执行ROLLBACK撤销操作
-- COMMIT;

通用SQL兼容方案（不依赖Postgres特有语法） #

如果需要适配其他数据库，把上述取最新审计记录的逻辑替换为窗口函数即可：

SELECT * FROM (
    SELECT *,
        ROW_NUMBER() OVER (PARTITION BY br_id ORDER BY audit_dtme DESC) AS rn
    FROM br_tbl_1_audit
    WHERE br_id = 123 AND audit_dtme <= '2024-05-01 12:00:00'
) t WHERE rn = 1

• 如果关联表的关联键不是直接存储br_id，需要调整查询逻辑，通过主表关联找到对应数据行的主键，再到对应审计表匹配历史版本
• 若审计表配置了过期数据清理策略，需要先确认目标时间点的审计数据仍完整留存

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