flink cdc同步pgsql到doris数据不一致？

[UTTUCK]

create table dwd_user_address_source (
id BIGINT,
.....
PRIMARY KEY(id) NOT ENFORCED
)
with
(
'connector' = 'postgres-cdc',
'changelog-mode' = 'upsert',
'hostname' = '${PG_HOSTNAME}',
'port' = '${PG_PORT}',
'username' = '${PG_USERNAME}',
'password' = '${PG_PASSWORD}',
'database-name' = 'jdb',
'schema-name' = 'public',
'table-name' = xxxxxx,
'slot.name' = xxxxxx,
'decoding.plugin.name' = 'pgoutput',
'scan.incremental.snapshot.enabled' = 'true'
);

CREATE TABLE dwd_user_address_sink (
id BIGINT,
.....
)
WITH
(
'connector' = 'doris',
'sink.enable.batch-mode' = 'true',
'sink.buffer-flush.max-rows' = '${SINK_BUFFER_FLUSH_MAX_ROWS}',
'sink.buffer-flush.interval' = '${SINK_BUFFER_FLUSH_INTERVAL}',
'fenodes' = '${DORIS_FENODES}',
'username' = '${DORIS_USERNAME}',
'password' = '${DORIS_PASSWORD}',
'table.identifier' = xxxxxx,
'sink.label-prefix' = xxxxxx,
'sink.properties.format' = 'json',
'sink.properties.read_json_by_line' = 'true'
);

INSERT INTO dwd_user_address_sink
SELECT
id,
.....
FROM dwd_user_address_source
where status = 'normal'
;
简简单单的同步，只做了where status = 'normal'筛选，但是doris上比pg上数据多了几条，在pg上数据status 是destroy被废弃，但是被doris同步过来而且status 变为normal，现在怀疑有几种可能：
1.是wal转upsert流后乱序 导致+i被推迟了
2.doris多并发写入导致+i 后写入
3.批写入有+i的批次被重复提交了