PostgreSQL导入CSV遇主键冲突时如何调整ID并更新关联表记录

方案说明 #

你需要的逻辑无法直接通过ON CONFLICT子句实现：ON CONFLICT仅支持两种冲突处理逻辑，要么忽略当前待插入行（DO NOTHING），要么修改表中已经存在的冲突行（DO UPDATE），无法自动为冲突的待插入行分配新ID后再插入，因此需要额外实现新旧ID的映射逻辑。

针对你的大表场景（5TB正式表、500万待导入数据），下面是兼顾性能和数据一致性的实现方案：

前置准备 #

如果允许修改表结构，建议先将company表的id字段设置为自增主键，避免手动生成ID带来的并发冲突问题：

-- PostgreSQL 10+ 支持的IDENTITY自增语法，兼容原有ID数据
ALTER TABLE company 
ALTER COLUMN id ADD GENERATED ALWAYS AS IDENTITY,
ADD PRIMARY KEY (id);
-- 同步自增序列的起始值为当前表的最大ID
SELECT setval(pg_get_serial_sequence('company', 'id'), COALESCE(MAX(id), 1)) FROM company;

完整导入脚本 #

BEGIN;
-- 1. 创建公司ID映射临时表，存储CSV原始ID和正式库新ID的对应关系
CREATE TEMP TABLE tmp_company_map (
    old_csv_id INT8 NOT NULL PRIMARY KEY,
    new_company_id INT8
) ON COMMIT DROP;

-- 2. 导入company.csv的原始ID到映射表
\COPY tmp_company_map(old_csv_id) FROM 'company.csv' WITH CSV HEADER DELIMITER as ',';

-- 3. 导入无冲突的公司ID，直接使用CSV原始ID
WITH non_conflict_ids AS (
    SELECT old_csv_id 
    FROM tmp_company_map
    WHERE NOT EXISTS (SELECT 1 FROM company c WHERE c.id = old_csv_id)
),
inserted_non_conflict AS (
    INSERT INTO company(id)
    SELECT old_csv_id FROM non_conflict_ids
    ON CONFLICT DO NOTHING
    RETURNING id
)
UPDATE tmp_company_map
SET new_company_id = old_csv_id
WHERE old_csv_id IN (SELECT id FROM inserted_non_conflict);

-- 4. 处理冲突的公司ID：自动分配新ID插入，同步更新映射表
WITH conflict_ids AS (
    SELECT old_csv_id FROM tmp_company_map WHERE new_company_id IS NULL
),
inserted_conflict AS (
    -- 利用自增属性自动生成新ID
    INSERT INTO company DEFAULT VALUES
    SELECT FROM conflict_ids
    RETURNING id
)
UPDATE tmp_company_map t
SET new_company_id = sq.new_id
FROM (
    SELECT 
        c.old_csv_id,
        i.id AS new_id,
        ROW_NUMBER() OVER(ORDER BY c.old_csv_id) AS rn
    FROM conflict_ids c
    JOIN (SELECT id, ROW_NUMBER() OVER() AS rn FROM inserted_conflict) i 
    ON c.rn = i.rn
) sq
WHERE t.old_csv_id = sq.old_csv_id;

-- 5. 导入人员CSV数据到临时表
CREATE TEMP TABLE tmp_people (
    name VARCHAR(100) NOT NULL,
    old_company_id INT8 NOT NULL
) ON COMMIT DROP;
\COPY tmp_people(name, old_company_id) FROM 'people.csv' WITH CSV HEADER DELIMITER as ',';

-- 6. 替换人员的关联公司ID后插入正式表
INSERT INTO people(name, company_id)
SELECT tp.name, tcm.new_company_id
FROM tmp_people tp
JOIN tmp_company_map tcm ON tp.old_company_id = tcm.old_csv_id
ON CONFLICT DO NOTHING;

COMMIT;

性能优化说明 #

• 所有查询仅需要比对临时表中的ID和正式company表的主键索引，不需要扫描全量5TB大表，执行效率极高
• 临时表的主键索引可以大幅提升关联替换的速度，500万条数据的关联操作耗时通常在秒级
• 整个操作在单事务中完成，不会出现数据不一致的问题，如果执行失败会自动回滚所有变更

无自增列适配方案 #

如果不允许修改company表结构添加自增属性，可以替换第4步的冲突处理逻辑，手动生成新ID：

-- 替换第4步的冲突处理逻辑
WITH conflict_ids AS (
    SELECT old_csv_id FROM tmp_company_map WHERE new_company_id IS NULL
),
max_exist_id AS (
    SELECT COALESCE(MAX(id), 0) AS max_id FROM company
),
new_id_mapping AS (
    SELECT 
        old_csv_id,
        max_id + ROW_NUMBER() OVER(ORDER BY old_csv_id) AS new_id
    FROM conflict_ids, max_exist_id
),
inserted_conflict AS (
    INSERT INTO company(id)
    SELECT new_id FROM new_id_mapping
    RETURNING id
)
UPDATE tmp_company_map t
SET new_company_id = nm.new_id
FROM new_id_mapping nm
WHERE t.old_csv_id = nm.old_csv_id;

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