如何获取Postgres 10所有现有表的Relation Id并触发Relation消息

嘿，我来帮你搞定这个问题！你在Go里用PostgreSQL 10逻辑复制时遇到的Relation Id获取问题，其实有两种靠谱的解决方式：一种是直接查系统表拿到所有表的Relation Id，另一种是通过触发逻辑复制的Relation消息来获取，我给你详细说说：

PostgreSQL的系统表pg_class里存储了所有关系（包括表、索引等）的元数据，其中oid字段就是逻辑复制里提到的Relation Id。你可以用SQL查询直接拿到所有用户表的Relation Id：

-- 查询public schema下的所有普通表的Relation Id和表名
SELECT oid AS relation_id, relname AS table_name
FROM pg_class
WHERE relkind = 'r'  -- 只筛选普通表（r代表regular table）
  AND relnamespace = (SELECT oid FROM pg_namespace WHERE nspname = 'public');

如果你的表分布在其他schema下，只需要修改nspname的值就行。这个方法简单直接，不需要依赖逻辑复制的消息流，适合提前获取所有表的Relation Id。

逻辑复制里的Relation消息是用来传递表结构元数据的，通常在初始快照阶段会自动发送所有订阅表的Relation消息。如果你想在Go代码中触发它，只需要在启动复制时请求初始快照即可。

下面是用pgx库（PostgreSQL的Go常用驱动）实现的示例代码：

package main

import (
    "context"
    "fmt"
    "github.com/jackc/pgx/v5"
    "github.com/jackc/pgx/v5/pgproto3"
)

func main() {
    // 1. 建立数据库连接
    connStr := "postgres://your_user:your_password@your_host:5432/your_db"
    conn, err := pgx.Connect(context.Background(), connStr)
    if err != nil {
        panic(fmt.Sprintf("连接失败: %v", err))
    }
    defer conn.Close()

    // 2. 获取复制连接
    replConn, err := conn.Replication()
    if err != nil {
        panic(fmt.Sprintf("获取复制连接失败: %v", err))
    }

    // 3. 启动逻辑复制，指定initial快照触发Relation消息
    // 替换成你的复制槽名和发布名
    slotName := "your_replication_slot"
    publicationName := "your_publication"
    err = replConn.StartReplication(
        context.Background(),
        slotName,
        0,  // 起始WAL位置，0表示从当前开始
        -1, // 终止WAL位置，-1表示持续复制
        map[string]string{
            "proto_version":      "1",              // 逻辑复制协议版本
            "publication_names":  publicationName,  // 订阅的发布名称
            "snapshot":           "initial",        // 请求初始快照，触发Relation消息
        },
    )
    if err != nil {
        panic(fmt.Sprintf("启动复制失败: %v", err))
    }

    // 4. 循环读取复制消息，处理Relation类型
    fmt.Println("开始接收复制消息...")
    for {
        msg, err := replConn.WaitForReplicationMessage(context.Background())
        if err != nil {
            panic(fmt.Sprintf("读取消息失败: %v", err))
        }

        switch msg := msg.(type) {
        case *pgproto3.RelationMessage:
            // 这里拿到了Relation消息，包含Relation ID和表结构
            fmt.Printf("收到Relation消息：Relation ID = %d，表名 = %s\n", msg.RelationID, msg.RelationName)
            // 可以在这里把Relation ID和表名映射保存下来
        case *pgproto3.InsertMessage:
            // 处理插入消息，此时可以用之前保存的Relation ID映射找到对应表
            fmt.Printf("收到Insert消息，对应Relation ID = %d\n", msg.RelationID)
        // 其他消息类型（Update、Delete、Commit等）可以按需处理
        default:
            fmt.Printf("收到未知类型消息: %T\n", msg)
        }
    }
}

关键说明： #

• 启动复制时指定"snapshot": "initial"是触发Relation消息的核心，PostgreSQL会把发布中所有表的元数据以Relation消息的形式发送过来。
• 如果你已经有一个正在运行的复制槽，想要重新获取Relation消息，只需要停止当前复制，重新执行START_REPLICATION并带上initial快照参数即可。
• 注意：Relation消息只会在初始快照阶段发送一次，后续的WAL日志中只有数据变更消息（Insert/Update/Delete等），所以如果需要后续新增表的Relation消息，你需要重新请求快照或者确保发布包含新增表后重新启动复制。

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