PostgreSQL实时同步多基础表至汇总表的实现方案问询

AI 小北

2026-7-5

实现方案

一、核心思路

通过状态跟踪表+通用触发器实现多基础表的时间戳同步校验，完成数据汇总后，再通过触发器发送NOTIFY通知Python端处理。视图无法直接使用AFTER INSERT触发器（仅支持物理表），因此放弃视图方案，改用物理表+触发器的组合逻辑。

二、具体实现步骤

1. 创建同步状态跟踪表

用于记录每个时间戳在各基础表的存在情况，判断是否满足"所有表都有相同Timestamp"的条件：

CREATE TABLE sync_status (
    timestamp_val VARCHAR(24) NOT NULL,
    table_name VARCHAR(64) NOT NULL,
    is_present BOOLEAN DEFAULT TRUE,
    -- 联合索引加速查询与冲突判断
    PRIMARY KEY (timestamp_val, table_name)
);

2. 创建通用触发器函数

处理基础表的插入事件，更新状态表并判断是否触发汇总同步：

CREATE OR REPLACE FUNCTION sync_to_final_table()
RETURNS TRIGGER AS $$
DECLARE
    total_base_tables INT;
    ready_table_count INT;
    insert_fields TEXT;
    insert_values TEXT;
BEGIN
    -- 1. 更新状态表：存在则标记为已就绪，不存在则插入记录
    INSERT INTO sync_status (timestamp_val, table_name)
    VALUES (NEW.Timestamp, TG_TABLE_NAME)
    ON CONFLICT (timestamp_val, table_name) DO UPDATE SET is_present = TRUE;

    -- 2. 获取所有基础表的总数（需匹配你的表命名规则，这里假设是table1/table2...）
    SELECT COUNT(*) INTO total_base_tables
    FROM information_schema.tables
    WHERE table_name LIKE 'table%' AND table_schema = 'public';

    -- 3. 检查当前时间戳已就绪的表数量
    SELECT COUNT(*) INTO ready_table_count
    FROM sync_status
    WHERE timestamp_val = NEW.Timestamp AND is_present = TRUE;

    -- 4. 所有表都就绪时，同步数据到final_table
    IF ready_table_count = total_base_tables THEN
        -- 动态拼接插入字段与值（避免硬编码100个表）
        SELECT 
            string_agg(table_name || '_Value', ', '),
            string_agg(format('(SELECT Value FROM %I WHERE Timestamp = %L)', table_name, NEW.Timestamp), ', ')
        INTO insert_fields, insert_values
        FROM information_schema.tables
        WHERE table_name LIKE 'table%' AND table_schema = 'public';

        -- 执行插入
        EXECUTE format(
            'INSERT INTO final_table (Timestamp, %s) VALUES (%L, %s)',
            insert_fields,
            NEW.Timestamp,
            insert_values
        );

        -- 标记该时间戳已处理，避免重复同步
        UPDATE sync_status
        SET is_present = FALSE
        WHERE timestamp_val = NEW.Timestamp;
    END IF;

    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

3. 批量给所有基础表绑定触发器

避免手动创建100个触发器：

DO $$
DECLARE
    table_record RECORD;
BEGIN
    FOR table_record IN
        SELECT table_name
        FROM information_schema.tables
        WHERE table_name LIKE 'table%' AND table_schema = 'public'
    LOOP
        EXECUTE format(
            'CREATE TRIGGER trigger_sync_%s
             AFTER INSERT ON %I
             FOR EACH ROW
             EXECUTE FUNCTION sync_to_final_table();',
            table_record.table_name,
            table_record.table_name
        );
    END LOOP;
END $$;

4. 创建final_table的NOTIFY触发器

用于向Python端发送数据更新通知：

CREATE OR REPLACE FUNCTION notify_final_table_update()
RETURNS TRIGGER AS $$
BEGIN
    -- 发送通知，携带时间戳方便Python定位数据
    PERFORM pg_notify('final_table_data_ready', NEW.Timestamp);
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER trigger_notify_final_table
AFTER INSERT ON final_table
FOR EACH ROW
EXECUTE FUNCTION notify_final_table_update();

5. Python端监听通知并处理数据

使用psycopg2实现监听逻辑：

import psycopg2
from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT

# 连接数据库（替换为你的连接参数）
conn = psycopg2.connect(
    dbname="your_db",
    user="your_user",
    password="your_password",
    host="your_host"
)
conn.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT)
cur = conn.cursor()

# 订阅通知频道
cur.execute("LISTEN final_table_data_ready;")
print("等待final_table数据更新通知...")

while True:
    conn.poll()
    # 处理收到的通知
    while conn.notifies:
        notify = conn.notifies.pop(0)
        timestamp = notify.payload
        print(f"收到新数据通知，时间戳：{timestamp}")
        
        # 查询对应时间戳的汇总数据
        cur.execute("SELECT * FROM final_table WHERE Timestamp = %s;", (timestamp,))
        data = cur.fetchone()
        
        # 这里添加你的数据处理逻辑
        print(f"处理数据：{data}")

三、关键注意事项

时间戳格式一致性：确保所有基础表的Timestamp字段格式完全统一（建议改用TIMESTAMP类型替代VARCHAR(24)，避免字符串匹配误差）。
性能优化：sync_status表的联合主键已自带索引，若数据量较大，可给final_table的Timestamp字段单独建索引。
异常处理：可在触发器函数中添加异常捕获逻辑，避免单个基础表的插入错误影响整体同步流程。

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