如何通过PostgreSQL CDC捕获修改操作的数据库用户并过滤特定事件

一、无需新增表字段，即可将数据库用户信息纳入Kafka事件 #

完全不需要新增业务表字段，主流PostgreSQL CDC工具（如Debezium、PostgreSQL逻辑复制+Kafka Connect）都能自动捕获执行变更的数据库用户，并将其包含在发送到Kafka的事件元数据中。

以Debezium PostgreSQL连接器为例，它默认会在事件的source元数据节点中加入db.user字段，记录执行该变更的数据库用户名。典型的事件结构片段如下：

{
  "payload": {
    "before": null,
    "after": {"id": 1, "name": "test"},
    "source": {
      "db": "mydb",
      "user": "app_user_1", // 记录执行变更的数据库用户
      "table": "test_table"
    }
  }
}

只要CDC工具配置正确，这个信息会自动带上，不需要对业务表做任何修改。

二、过滤特定用户变更的多种方案 #

除了在消费端基于事件中的用户信息过滤外，还有几个更高效的前置过滤方式：

1. CDC工具层面直接过滤 #

以Debezium为例，可以通过配置**SMT（Single Message Transform）**在事件发送到Kafka前直接过滤掉特定用户的变更。示例配置如下：

connector.class=io.debezium.connector.postgresql.PostgresConnector
transforms=filterUser
transforms.filterUser.type=io.debezium.transforms.Filter
transforms.filterUser.language=jsr223.groovy
transforms.filterUser.condition="source.user != 'system_user'"

该配置会直接丢弃由system_user产生的所有变更事件，不流入Kafka。

2. PostgreSQL逻辑复制端过滤 #

如果使用PostgreSQL原生逻辑复制（搭配wal2json、pgoutput等解码插件），可以在创建发布时指定过滤规则，或者通过解码插件的参数过滤特定用户的变更。比如wal2json支持通过filter-users参数排除指定用户的变更，在配置复制槽时添加该参数即可。

3. Kafka流处理层过滤 #

如果事件已经流入Kafka，可以用Kafka Streams或KSQL编写简单的过滤逻辑，将特定用户的事件从流中剔除。比如KSQL语句：

CREATE STREAM filtered_events AS
SELECT * FROM raw_cdc_events
WHERE source->>'user' != 'system_user';

4. 数据库触发器辅助（备选方案） #

如果你的CDC工具不支持自动捕获用户信息（极少情况），可以通过数据库触发器在变更发生时自动记录操作用户到辅助字段或日志表，但这种方式不如CDC原生捕获的信息准确，且会增加数据库开销，仅作为备选。

参考内容（原SQL Server CDC相关问题翻译） #

原问题讨论的是SQL Server的CDC捕获变更用户的方案：SQL Server CDC默认会在生成的变更表中包含__$username字段，直接记录执行变更的用户，无需额外添加业务表字段，可直接基于该字段过滤特定用户的变更事件。

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