PostgreSQL Upsert触发CardinalityViolation错误的排查与解决咨询

从Databricks向PostgreSQL的edh_services.ec_users表加载数据时，执行以下Upsert SQL语句触发了CardinalityViolation错误：

INSERT INTO edh_services.ec_users
    (cdc_ID,organization_Id,user_email_ID,rec_crt_ts,rec_updt_ts) SELECT DISTINCT 
   cdc_ID, organization_Id,user_email_ID,rec_crt_ts,rec_updt_ts \
FROM 
    edh_services.ec_users_stg 
ON CONFLICT (cdc_ID) DO UPDATE SET cdc_ID = excluded.cdc_ID, 
organization_Id = excluded.organization_Id,
user_email_ID = excluded.user_email_ID,
rec_crt_ts = excluded.rec_crt_ts, 
rec_updt_ts = excluded.rec_updt_ts 

错误信息 #

---------------------------------------------------------------------------
CardinalityViolation                      Traceback (most recent call last)
<command-1477374869543974> in <module>
     31 upsert_script = f"INSERT INTO {table_name} (" +upsert_sql1[0:-1]+") SELECT DISTINCT cdc_ID, "+ tab_sql1[0:-1]+f" FROM {stg_table_name} ON CONFLICT (cdc_ID) DO UPDATE SET   "+upsert_sql2[0:-1]
     32 print(upsert_script)
---> 33 cur.execute(upsert_script)
     34 con.commit()
     35 

CardinalityViolation: ON CONFLICT DO UPDATE command cannot affect row a second time
HINT:  Ensure that no rows proposed for insertion within the same command have duplicate constrained values.

目标表结构 #

create table edh_services.EC_Users
(
cdc_ID  Varchar(200),
organization_Id  BigSerial,
user_email_ID  Varchar(500),
rec_crt_ts  TIMESTAMP with Time zone,
rec_updt_ts  TIMESTAMP with Time zone,
constraint EC_Users_pk primary key (cdc_ID),
constraint EC_Users_fk foreign key (organization_Id)  references edh_services.EC_Organization (organization_Id) match simple
on update no action
on delete no action
)

问题 #

请问有什么方法可以避免该错误？需要排查哪些列的重复值？

一、需要排查的重复列 #

错误提示明确指向主键约束的重复值，核心排查点是临时表edh_services.ec_users_stg中的cdc_ID列：

• 即使使用了SELECT DISTINCT，如果同一cdc_ID对应的其他列（如organization_Id、user_email_ID等）存在不同值，DISTINCT会保留多条同cdc_ID但其他列不同的记录，导致Upsert时同一主键被多次匹配，触发错误。
• 需要确认这些重复cdc_ID对应的记录中，哪些是业务上需要保留的最新/有效数据。

二、避免错误的方法 #

1. 用窗口函数按主键去重，保留最新记录

通过ROW_NUMBER()窗口函数按cdc_ID分组，取更新时间最晚的记录，确保每个主键只有一条待插入/更新的记录：

INSERT INTO edh_services.ec_users
    (cdc_ID,organization_Id,user_email_ID,rec_crt_ts,rec_updt_ts)
SELECT 
    cdc_ID, organization_Id,user_email_ID,rec_crt_ts,rec_updt_ts
FROM (
    SELECT 
        *,
        ROW_NUMBER() OVER (PARTITION BY cdc_ID ORDER BY rec_updt_ts DESC) AS rn
    FROM edh_services.ec_users_stg
) t
WHERE t.rn = 1
ON CONFLICT (cdc_ID) DO UPDATE SET 
    organization_Id = excluded.organization_Id,
    user_email_ID = excluded.user_email_ID,
    rec_crt_ts = excluded.rec_crt_ts, 
    rec_updt_ts = excluded.rec_updt_ts

注：Upsert时无需更新cdc_ID（主键冲突时excluded.cdc_ID与目标表值一致），可去掉该字段的更新语句。

2. 提前清理临时表的重复数据

先对临时表做去重处理，再执行Upsert：

-- 创建去重后的临时表
CREATE TEMP TABLE ec_users_stg_dedup AS
SELECT 
    cdc_ID, organization_Id,user_email_ID,rec_crt_ts,rec_updt_ts
FROM (
    SELECT 
        *,
        ROW_NUMBER() OVER (PARTITION BY cdc_ID ORDER BY rec_updt_ts DESC) AS rn
    FROM edh_services.ec_users_stg
) t
WHERE t.rn = 1;

-- 使用去重后的表执行Upsert
INSERT INTO edh_services.ec_users
    (cdc_ID,organization_Id,user_email_ID,rec_crt_ts,rec_updt_ts)
SELECT * FROM ec_users_stg_dedup
ON CONFLICT (cdc_ID) DO UPDATE SET 
    organization_Id = excluded.organization_Id,
    user_email_ID = excluded.user_email_ID,
    rec_crt_ts = excluded.rec_crt_ts, 
    rec_updt_ts = excluded.rec_updt_ts;

3. 调整DISTINCT逻辑或用聚合函数去重

如果业务允许对同cdc_ID的其他列做聚合处理，可改用GROUP BY配合聚合函数（如取最新时间、最大值等）：

INSERT INTO edh_services.ec_users
    (cdc_ID,organization_Id,user_email_ID,rec_crt_ts,rec_updt_ts)
SELECT 
    cdc_ID,
    MAX(organization_Id) AS organization_Id, -- 根据业务选择合适的聚合规则
    MAX(user_email_ID) AS user_email_ID,
    MIN(rec_crt_ts) AS rec_crt_ts,
    MAX(rec_updt_ts) AS rec_updt_ts
FROM edh_services.ec_users_stg
GROUP BY cdc_ID
ON CONFLICT (cdc_ID) DO UPDATE SET 
    organization_Id = excluded.organization_Id,
    user_email_ID = excluded.user_email_ID,
    rec_crt_ts = excluded.rec_crt_ts, 
    rec_updt_ts = excluded.rec_updt_ts;

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