我正在使用Google Cloud Datastream与 Cloud SQL for PostgreSQL作为数据源，想了解Datastream预计在PostgreSQL的 DELETE事件中包含哪些内容。

我的源表类似于此：

CREATE TABLE public.example_events (
  event_id varchar(255) PRIMARY KEY,
  order_id bigint NOT NULL,
  event_type int NOT NULL,
  start_time timestamp NOT NULL,
  end_time timestamp NOT NULL
);

重要细节是event_id是主键，但order_id是一个非主键列，我在下游需要用于路由/排序。

当我检查Datastream Avro输出的删除事件时，我发现只有主键列被填充，所有非主键列都是NULL。

示例（已脱敏）：

{
  "source_metadata": {
    "primary_keys": { "array": ["col_key"] },
    "is_deleted": { "boolean": true },
    "change_type": { "string": "DELETE" }
  },
  "payload": {
    "col_key": { "string": "EVT_AF_002" },
    "col_order_id": null,
    "col_event_type": null,
    "col_start_time": null,
    "col_end_time": null,
    "updated_at": null
  }
}

因此，在实际应用中，删除事件通过主键来标识行，但不会给我非键列order_id。

我想确认的是：

1. 当表的主键不包含order_id时，这是否是Datastream对 PostgreSQL删除事件的预期行为？
2. 是否存在任何受支持的Datastream/PostgreSQL配置，可以让删除事件包含像order_id这样的非主键列？

有没有人就Cloud SQL for PostgreSQL + Datastream的预期行为进行过确认？

解决方案

正如 [@Laurenz Albe] 已经建议的那样，你可以把 replica identity 从 default 切换到 full：

alter table public.example_events replica identity full;

这在 [GCP Datastream手册] 中有说明：

正确配置replica identity

REPLICA IDENTITY 设置告诉PostgreSQL在 UPDATE 和 DELETE 事件中向WAL写入哪些数据，从而使Datastream能识别哪些行被更改。

如果你使用BigQuery作为目标，请避免将 REPLICA IDENTITY 设置为 FULL。Datastream将日志的列用作BigQuery MERGE 操作的逻辑键。如果 REPLICA IDENTITY 设置为 FULL，且一张表有超过16列，这将超过BigQuery在 MERGE 操作中对主键的16列限制并导致流中断。

按偏好排序的建议如下：

• 最佳：使用主键。默认设置的 REPLICA IDENTITY DEFAULT 会自动且高效地使用现有主键。
• 良好：如果不存在主键，创建一个 UNIQUE NOT NULL 索引并设置 REPLICA IDENTITY USING INDEX INDEX_NAME。
• 最不推荐：仅在没有唯一标识符的表上使用 REPLICA IDENTITY FULL 设置。请注意对性能的影响、16列的限制，以及在复制到BigQuery时对主键数据类型的支持限制。

我创建了一个Cloud SQL Postgres沙箱实例，复制了你的示例

create table public.example_events (
  event_id varchar(255) primary key,
  order_id bigint not null,
  event_type int not null,
  start_time timestamp not null,
  end_time timestamp not null
);
insert into public.example_events values('event_id1', 1, 1, now(), now());
alter table public.example_events replica identity default;
create publication mypubname1 for table public.example_events;
alter user postgres with replication;
select pg_create_logical_replication_slot('myrepslotname1', 'pgoutput');

为该实例分配了一个外部网络接口，在实例的授权网络中对白名单Datastream IP，打开了它们的防火墙，设置了一个Datastream，将该表的Avro数据写入到一个存储桶中，一旦启动，它就回填了我上面插入的初始行。

又添加了几条，然后删除了一条：

with cte as(select ctid from public.example_events limit 1)
delete from public.example_events where ctid in(table cte);

CDC:

"source_metadata": {
  "table": "example_events",
  "schema": "public",
  "primary_keys": [
    "event_id",
    "order_id",
    "event_type",
    "start_time",
    "end_time"
  ],
  "is_deleted": true,
  "change_type": "DELETE",
  "tx_id": 2451,
  "lsn": "0/4DA5C48"
},
"payload": {
  "event_id": "event_id2",
  "order_id": 0,
  "event_type": 0,
  "start_time": 0,
  "end_time": 0
}

载荷中仅记录了主键。将 replica identity 改为 full，又删除了另一行：

alter table public.example_events replica identity full;

with cte as(select ctid from public.example_events limit 1)
delete from public.example_events where ctid in(table cte)returning*;

CDC:

"source_metadata": {
  "table": "example_events",
  "schema": "public",
  "primary_keys": [
    "event_id",
    "order_id",
    "event_type",
    "start_time",
    "end_time"
  ],
  "is_deleted": true,
  "change_type": "DELETE",
  "tx_id": 2459,
  "lsn": "0/4DA5F98"
},
"payload": {
  "event_id": "event_id3",
  "order_id": 3,
  "event_type": 3,
  "start_time": 1773832259613355,
  "end_time": 1773832259613355
}

载荷现在显示了所有列，正如你所想要的。