PostgreSQL每日存储过程性能优化：如何跳过前日已处理内容

作为PostgreSQL老玩家，我来给你分享几个实用的思路，帮你跳过已处理内容、大幅提升每日汇总的性能～

1. 基于时间戳/更新标记的增量过滤（最易上手） #

这是新手最容易落地的方案，核心思路是给源表加一个记录数据更新时间的字段，然后每次只处理上次汇总之后新增/修改的数据。

步骤拆解： #

• 首先给所有需要汇总的源表添加last_updated字段（如果还没有的话）：

  ALTER TABLE your_source_table ADD COLUMN last_updated TIMESTAMPTZ DEFAULT NOW();
  

  再建个触发器，确保数据更新时自动刷新这个字段：

  CREATE OR REPLACE FUNCTION update_last_updated()
  RETURNS TRIGGER AS $$
  BEGIN
      NEW.last_updated = NOW();
      RETURN NEW;
  END;
  $$ LANGUAGE plpgsql;
  
  CREATE TRIGGER trigger_your_source_table_update
  BEFORE INSERT OR UPDATE ON your_source_table
  FOR EACH ROW EXECUTE FUNCTION update_last_updated();
  

• 然后创建一个控制表来记录上次汇总的时间，避免每次硬编码或者依赖外部脚本：

  CREATE TABLE etl_job_control (
      job_name VARCHAR(100) PRIMARY KEY,
      last_run_time TIMESTAMPTZ NOT NULL
  );
  -- 初始化你的汇总任务记录
  INSERT INTO etl_job_control (job_name, last_run_time)
  VALUES ('daily_data_summary', '1970-01-01 00:00:00+00');
  

• 最后修改你的存储过程，每次运行前先读取上次汇总时间，只处理增量数据，运行后更新控制表：

  CREATE OR REPLACE PROCEDURE run_daily_summary()
  LANGUAGE plpgsql
  AS $$
  DECLARE
      v_last_run TIMESTAMPTZ;
  BEGIN
      -- 获取上次汇总时间
      SELECT last_run_time INTO v_last_run
      FROM etl_job_control
      WHERE job_name = 'daily_data_summary';
  
      -- 只处理上次汇总之后更新的数据
      INSERT INTO your_summary_table (col1, col2, summary_value)
      SELECT col1, col2, SUM(amount)
      FROM your_source_table
      WHERE last_updated > v_last_run
      GROUP BY col1, col2;
  
      -- 更新最后一次汇总时间为当前时间
      UPDATE etl_job_control
      SET last_run_time = NOW()
      WHERE job_name = 'daily_data_summary';
  END;
  $$;
  

⚠️ 关键提醒：一定要给last_updated字段加索引，不然过滤数据时还是会全表扫描，起不到优化作用：

CREATE INDEX idx_your_source_last_updated ON your_source_table(last_updated);

2. 用触发器维护待处理队列（适合高频更新场景） #

如果你的源表数据更新非常频繁，每次按时间戳过滤还是要扫描大量数据，可以试试触发器+待处理队列表的方案：

• 创建一个待处理队列表，用来记录需要汇总的源表主键：

  CREATE TABLE pending_summary_records (
      record_id INT PRIMARY KEY, -- 对应源表的主键
      source_table VARCHAR(100) NOT NULL, -- 标记来自哪个源表
      is_processed BOOLEAN DEFAULT FALSE,
      created_at TIMESTAMPTZ DEFAULT NOW()
  );
  

• 给每个源表加触发器，当数据插入/更新时，自动把主键加入队列：

  CREATE OR REPLACE FUNCTION add_to_pending_queue()
  RETURNS TRIGGER AS $$
  BEGIN
      -- 先删除已存在的记录（避免重复处理同一条更新）
      DELETE FROM pending_summary_records
      WHERE record_id = NEW.id AND source_table = 'your_source_table';
      -- 插入新的待处理记录
      INSERT INTO pending_summary_records (record_id, source_table)
      VALUES (NEW.id, 'your_source_table');
      RETURN NEW;
  END;
  $$ LANGUAGE plpgsql;
  
  CREATE TRIGGER trigger_source_table_queue
  AFTER INSERT OR UPDATE ON your_source_table
  FOR EACH ROW EXECUTE FUNCTION add_to_pending_queue();
  

• 修改存储过程，直接处理队列中的未处理记录，处理完标记为已处理：

  CREATE OR REPLACE PROCEDURE run_daily_summary()
  LANGUAGE plpgsql
  AS $$
  DECLARE
      v_record RECORD;
  BEGIN
      -- 遍历待处理记录
      FOR v_record IN
          SELECT record_id, source_table
          FROM pending_summary_records
          WHERE is_processed = FALSE
      LOOP
          -- 根据源表处理对应记录的汇总逻辑
          IF v_record.source_table = 'your_source_table' THEN
              -- 这里写你的汇总逻辑，比如更新汇总表或者插入新记录
              INSERT INTO your_summary_table (col1, col2, summary_value)
              SELECT col1, col2, SUM(amount)
              FROM your_source_table
              WHERE id = v_record.record_id
              ON CONFLICT (col1, col2) DO UPDATE SET summary_value = EXCLUDED.summary_value;
          END IF;
  
          -- 标记为已处理
          UPDATE pending_summary_records
          SET is_processed = TRUE
          WHERE record_id = v_record.record_id AND source_table = v_record.source_table;
      END LOOP;
  
      -- 可选：清理已处理的旧记录，避免队列表过大
      DELETE FROM pending_summary_records
      WHERE is_processed = TRUE AND created_at < NOW() - INTERVAL '7 days';
  END;
  $$;
  

3. 增量汇总到中间表再合并（适合复杂汇总逻辑） #

如果你的汇总逻辑很复杂，涉及多张表关联，可以先把每日增量数据提取到中间表，再基于中间表做汇总，这样能减少大表关联的次数：

• 创建中间表存储每日增量数据：

  CREATE TABLE daily_increment_data (
      id SERIAL PRIMARY KEY,
      col1 VARCHAR(50),
      col2 INT,
      amount NUMERIC,
      source_table VARCHAR(100),
      created_at TIMESTAMPTZ DEFAULT NOW()
  );
  

• 存储过程第一步先把增量数据同步到中间表：

  -- 先同步增量数据到中间表
  INSERT INTO daily_increment_data (col1, col2, amount, source_table)
  SELECT col1, col2, amount, 'your_source_table'
  FROM your_source_table
  WHERE last_updated > v_last_run;
  

• 然后基于中间表做汇总，再合并到最终汇总表：

  -- 基于中间表做汇总
  INSERT INTO your_summary_table (col1, col2, summary_value)
  SELECT col1, col2, SUM(amount)
  FROM daily_increment_data
  WHERE created_at > v_last_run
  GROUP BY col1, col2
  ON CONFLICT (col1, col2) DO UPDATE SET summary_value = your_summary_table.summary_value + EXCLUDED.summary_value;
  

额外小技巧 #

• 如果你的汇总任务是每日固定时间运行，可以直接用NOW() - INTERVAL '1 day'作为过滤条件，但要注意时区问题，避免跨天数据漏处理。
• 定期清理汇总表的历史数据（如果不需要长期保留），或者用分区表（比如按日期分区）来提升查询和汇总效率。

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