PL/pgSQL长时循环任务进度监控优化方案咨询

我有运行时长可达数小时的PL/pgSQL函数和存储过程，需要实现类似Python tqdm或R progress的进度监控功能，确认任务在正常推进。

现有实现的问题 #

我已经有一个可用的基础版本，功能是读取example.task_list表中的整数，通过Upsert将其两倍值存入第二列。但这个实现有两个明显不足：

• 重复执行相同SQL查询：一次统计待完成任务数，一次用于循环遍历
• 业务逻辑与计时、进度通知代码混杂，可读性差

现有粗糙实现代码 #

CREATE TABLE example.task_list (
  intt int NOT NULL PRIMARY KEY,
  double_intt int
);

INSERT INTO example.task_list VALUES
  (1, NULL), (2, NULL), (3, NULL), (4, NULL)
RETURNING *;

CREATE OR REPLACE function example.doloop()
  returns void
  LANGUAGE plpgsql AS
$func$
DECLARE
   begin_time timestamptz = clock_timestamp();
   num_tasks_done int := 0;
   average_time_per_task interval := null;
   tasks_remaining int;
   f record;
   time_to_do interval;
BEGIN
   SELECT count(*) INTO tasks_remaining FROM (SELECT DISTINCT intt from example.task_list where double_intt is null);
   raise notice '%: Starting. We have  [%] tasks to do.', clock_timestamp(), tasks_remaining;
   FOR f IN (SELECT DISTINCT intt from example.task_list where double_intt is null)
      LOOP
      begin_time = clock_timestamp();
      raise notice '%: Putting in [%]', begin_time, f.intt;
      raise notice '%:   We have done [%] jobs and have [%] jobs remaining. It will take %', clock_timestamp(), num_tasks_done, tasks_remaining, average_time_per_task * tasks_remaining;
      INSERT INTO example.task_list (intt, double_intt) values (f.intt, 2 * f.intt)
      ON CONFLICT("intt") DO UPDATE
      SET double_intt = EXCLUDED.double_intt;
      time_to_do = clock_timestamp() - begin_time;
      raise notice '%:        Finished. It took : %', clock_timestamp(), time_to_do;
      /* Do calculations for time remaining */
      average_time_per_task = ((coalesce(average_time_per_task,'0 minutes') * num_tasks_done) + (time_to_do)) / (num_tasks_done+1);
      num_tasks_done = num_tasks_done + 1;
      tasks_remaining = tasks_remaining - 1;
      END LOOP;
  END;
$func$;
select example.doloop();

优化方案：复用任务列表 + 封装进度逻辑 #

1. 复用任务列表：避免重复查询 #

先将待处理的任务结果存入游标或临时表，只执行一次查询，既可以获取总任务数，又可以遍历处理。

2. 封装进度监控逻辑：分离业务与监控 #

创建自定义类型存储进度状态，编写通用的进度跟踪函数负责计时、计算剩余时间和输出通知，让业务逻辑专注于核心操作。

最终实现代码 #

第一步：创建进度跟踪的自定义类型和通用函数

-- 定义进度跟踪的状态类型
CREATE TYPE example.progress_state AS (
    total_tasks int,
    completed_tasks int,
    avg_time_per_task interval,
    start_time timestamptz
);

-- 初始化进度状态的函数
CREATE OR REPLACE FUNCTION example.init_progress(total_tasks int)
RETURNS example.progress_state AS $$
BEGIN
    RETURN (total_tasks, 0, '0 seconds'::interval, clock_timestamp())::example.progress_state;
END;
$$ LANGUAGE plpgsql;

-- 更新进度并输出通知的函数
CREATE OR REPLACE FUNCTION example.update_progress(state INOUT example.progress_state, task_id text DEFAULT '')
RETURNS example.progress_state AS $$
DECLARE
    elapsed_time interval;
    remaining_tasks int;
    estimated_remaining interval;
BEGIN
    -- 计算当前任务耗时
    elapsed_time := clock_timestamp() - state.start_time;
    
    -- 更新平均耗时
    state.avg_time_per_task := (state.avg_time_per_task * state.completed_tasks + elapsed_time) / (state.completed_tasks + 1);
    
    -- 更新完成数与剩余数
    state.completed_tasks := state.completed_tasks + 1;
    remaining_tasks := state.total_tasks - state.completed_tasks;
    
    -- 计算预计剩余时间
    estimated_remaining := state.avg_time_per_task * remaining_tasks;
    
    -- 输出进度通知
    RAISE NOTICE '[%] 进度: 已完成 %/% 任务 | 当前任务耗时: % | 预计剩余时间: %',
        clock_timestamp(),
        state.completed_tasks,
        state.total_tasks,
        elapsed_time,
        estimated_remaining;
    
    -- 重置下一个任务的开始时间
    state.start_time := clock_timestamp();
END;
$$ LANGUAGE plpgsql;

第二步：优化后的业务函数

CREATE OR REPLACE FUNCTION example.do_loop_with_progress()
RETURNS void AS $$
DECLARE
    -- 定义游标存储待处理任务（仅执行一次查询）
    temp_tasks CURSOR FOR SELECT DISTINCT intt FROM example.task_list WHERE double_intt IS NULL;
    task_record record;
    progress example.progress_state;
    total_tasks int;
BEGIN
    -- 获取总任务数
    SELECT count(*) INTO total_tasks FROM (SELECT DISTINCT intt FROM example.task_list WHERE double_intt IS NULL) t;
    
    -- 初始化进度状态
    progress := example.init_progress(total_tasks);
    RAISE NOTICE '%: 任务启动，共 % 个待处理任务', clock_timestamp(), total_tasks;
    
    -- 遍历任务并执行业务逻辑
    FOR task_record IN temp_tasks LOOP
        -- 核心业务操作：Upsert两倍值
        INSERT INTO example.task_list (intt, double_intt)
        VALUES (task_record.intt, 2 * task_record.intt)
        ON CONFLICT(intt) DO UPDATE
        SET double_intt = EXCLUDED.double_intt;
        
        -- 调用进度更新函数
        PERFORM example.update_progress(progress, task_record.intt::text);
    END LOOP;
    
    RAISE NOTICE '%: 所有任务完成，总耗时: %', clock_timestamp(), clock_timestamp() - (progress).start_time;
END;
$$ LANGUAGE plpgsql;

调用优化后的函数

SELECT example.do_loop_with_progress();

方案优势 #

• 无重复查询：仅执行一次任务列表查询，同时获取总数和遍历数据
• 逻辑解耦：业务代码专注核心操作，进度监控逻辑封装为通用组件，可读性与可维护性提升
• 可复用性：进度跟踪的类型和函数可直接用于其他需要监控的PL/pgSQL函数

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