PostgreSQL物化视图增量刷新优化方案咨询

问题背景 #

你创建的public.kwh_tag物化视图用于统计每日电表读数及当日耗电量，但因smartmeter表每秒/每2秒写入数据，全量刷新耗时长达1分半以上，且随数据增长持续增加；使用CONCURRENTLY选项耗时更长。由于历史数据无变更，仅需增量计算最新日期数据。

物化视图定义及数据示例 #

-- 物化视图结构
                                   Materialized view "public.kwh_tag"
   Column    |       Type       | Collation | Nullable | Default | Storage | Stats target | Description
-------------+------------------+-----------+----------+---------+---------+--------------+-------------
 timest      | date             |           |          |         | plain   |              |
 Zählerstand | double precision |           |          |         | plain   |              |
 kWh         | double precision |           |          |         | plain   |              |
Indexes:
    "kwh_tag_timest_idx" UNIQUE, btree (timest)

-- 视图定义
 SELECT s.timest::date AS timest,
    s.kwh_180 AS "Zählerstand",
    s.kwh_180 - lag(s.kwh_180, 1) OVER (ORDER BY s.timest) AS "kWh"
   FROM ( SELECT DISTINCT ON ((smartmeter.timest::date)) smartmeter.timest,
            smartmeter.kwh_180
           FROM smartmeter
          ORDER BY (smartmeter.timest::date) DESC, smartmeter.timest DESC) s;

数据输出示例：

2023-06-10 |  35965.0285 | 22.8133000000016
 2023-06-11 |   35985.919 | 20.8905000000013
 2023-06-12 |  36012.7307 | 26.8116999999984
 2023-06-13 |  36030.2164 | 17.4856999999975

内置解决方案 #

1. 手动增量更新脚本 #

利用物化视图的唯一索引kwh_tag_timest_idx，仅计算并插入/更新物化视图中未包含的最新日期数据：

WITH latest_mv AS (
    -- 获取物化视图中最新的记录（用于计算新增日期的耗电量）
    SELECT timest AS last_date, "Zählerstand" AS last_kwh 
    FROM kwh_tag 
    ORDER BY timest DESC LIMIT 1
),
new_daily_data AS (
    -- 从smartmeter中获取新增日期的每日最后一条读数
    SELECT DISTINCT ON (sm.timest::date)
        sm.timest::date AS timest,
        sm.kwh_180 AS current_kwh
    FROM smartmeter sm
    CROSS JOIN latest_mv lm
    WHERE sm.timest::date > lm.last_date
    ORDER BY sm.timest::date DESC, sm.timest DESC
)
-- 插入/更新到物化视图
INSERT INTO kwh_tag (timest, "Zählerstand", "kWh")
SELECT 
    nd.timest,
    nd.current_kwh,
    -- 计算当日耗电量：当日读数 - 前一日读数（若为第一个新增日期则用物化视图最新读数）
    nd.current_kwh - COALESCE(
        (SELECT "Zählerstand" FROM kwh_tag WHERE timest = (nd.timest - INTERVAL '1 day')::date),
        lm.last_kwh
    ) AS "kWh"
FROM new_daily_data nd
CROSS JOIN latest_mv lm
-- 处理可能的重复数据（如中途手动刷新过当日数据）
ON CONFLICT (timest) DO UPDATE 
SET "Zählerstand" = EXCLUDED."Zählerstand",
    "kWh" = EXCLUDED."Zählerstand" - COALESCE(
        (SELECT "Zählerstand" FROM kwh_tag WHERE timest = (EXCLUDED.timest - INTERVAL '1 day')::date),
        (SELECT last_kwh FROM latest_mv)
    );

2. 定时自动增量更新（推荐） #

通过pg_cron扩展定时执行上述增量脚本，避免手动操作，推荐每日凌晨执行（此时业务负载低）：

-- 先安装pg_cron扩展（若未安装）
CREATE EXTENSION IF NOT EXISTS pg_cron;

-- 每天凌晨2点执行增量更新
SELECT cron.schedule(
    'daily-refresh-kwh-tag',
    '0 2 * * *',
    $$
    WITH latest_mv AS (
        SELECT timest AS last_date, "Zählerstand" AS last_kwh FROM kwh_tag ORDER BY timest DESC LIMIT 1
    ),
    new_daily_data AS (
        SELECT DISTINCT ON (sm.timest::date)
            sm.timest::date AS timest,
            sm.kwh_180 AS current_kwh
        FROM smartmeter sm
        CROSS JOIN latest_mv lm
        WHERE sm.timest::date > lm.last_date
        ORDER BY sm.timest::date DESC, sm.timest DESC
    )
    INSERT INTO kwh_tag (timest, "Zählerstand", "kWh")
    SELECT 
        nd.timest,
        nd.current_kwh,
        nd.current_kwh - COALESCE(
            (SELECT "Zählerstand" FROM kwh_tag WHERE timest = (nd.timest - INTERVAL '1 day')::date),
            lm.last_kwh
        ) AS "kWh"
    FROM new_daily_data nd
    CROSS JOIN latest_mv lm
    ON CONFLICT (timest) DO UPDATE 
    SET "Zählerstand" = EXCLUDED."Zählerstand",
        "kWh" = EXCLUDED."Zählerstand" - COALESCE(
            (SELECT "Zählerstand" FROM kwh_tag WHERE timest = (EXCLUDED.timest - INTERVAL '1 day')::date),
            (SELECT last_kwh FROM latest_mv)
        );
    $$
);

3. 性能优化辅助 #

• 为smartmeter表创建timest字段的降序索引，加速每日最新记录查询：

CREATE INDEX idx_smartmeter_timest_desc ON smartmeter (timest DESC);

• 首次初始化时需全量刷新一次物化视图，之后再使用增量更新：

REFRESH MATERIALIZED VIEW kwh_tag;

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