批量克隆记录时如何高效替换id字段及关联引用值

问题场景 #

假设数据库表中存在如下两条记录：

[
    {
        id: 1,
        name: 'Michael',
        associations: [
            {
                from: 1,
                to: 2
            }
        ]
    },
    {
        id: 2,
        name: 'John',
        associations: [
            {
                from: 2,
                to: 1
            }
        ]
    },
]

需要克隆这两个对象，最终得到如下结果：

[
    {
        id: 1,
        name: 'Michael',
        associations: [
            {
                from: 1,
                to: 2
            }
        ]
    },
    {
        id: 2,
        name: 'John',
        associations: [
            {
                from: 2,
                to: 1
            }
        ]
    },
    {
        id: 3,
        name: 'Michael',
        associations: [
            {
                from: 3,
                to: 4
            }
        ]
    },
    {
        id: 4,
        name: 'John',
        associations: [
            {
                from: 4,
                to: 3
            }
        ]
    },
]

当前实现逻辑为：遍历所有记录逐条执行INSERT操作，为每个元素维护oldId与newId的映射关系，插入完成后查询所有包含对应newId的记录，再替换关联字段的id值。该方案在数据量较小时可正常运行，但当需要克隆数千条记录、且每条记录包含数千条关联数据时，会产生严重的性能损耗。
当前实现代码如下，入口函数为cloneElements：

const updateElementAssociationsById = async ({ id, associations }) => {
    return dbQuery(id, associations); // UPDATE ... SET associations WHERE id
}

const handleUpdateAssociations = async ({ data }) => {
    const promises = [];
    const records = await dbQuery(data.map(({ newId }) => newId)) ; // SELECT * FROM ... WHERE id in ANY 

    records.forEach(({ id, ...record }) => {
        const associations = [];

        record.associations.forEach((association) => {
            const sourceElement = data.find(({ oldId }) => oldId === association.from);
            const targetElement = data.find(({ oldId }) => oldId === association.to)

            association.from = sourceElement.newId;
            association.to = targetElement.newId;

            associations.push(association);
        });

        promises.push(updateElementAssociationsById({ id, associations }))
    });

    await Promise.all(promises);
}

const createElement = async ({ element, data }) => {
    const newElement = await dbQuery(element); // INSERT INTO ... RETURNING *;

    data.push({
        oldId: element.id,
        newId: newElement.id,
    });   
}

const cloneElements = async (records) => {
    const promises = [];
    const data = [];

    records.forEach((element) => {
        promises.push(createElement({ element, records, data }));
    });

    await Promise.all(promises);

    await handleUpdateAssociations({ data });
}

核心性能瓶颈 #

现有实现性能差的原因非常明确，主要有4个问题：

• 单条循环INSERT产生大量数据库IO：数千条记录对应数千次数据库请求，网络往返、事务提交的开销占总耗时的70%以上
• 关联ID查找效率极低：用数组find()方法做ID匹配，单次查找时间复杂度为O(n)，总关联数大的时候光内存遍历就要消耗大量时间
• 冗余读写操作：插入新记录后又全量SELECT回查，再逐条UPDATE关联字段，等于多做了一轮全量读+N轮单条写，完全没有必要
• 并发写入无控制：直接用Promise.all发起所有写请求，很容易把数据库连接池打满，反而触发限流导致整体变慢

优化后实现方案 #

核心思路是把能在内存里做完的事全做完，尽量减少数据库交互次数，具体步骤：

1. 提前构建O(1)复杂度的ID映射结构，用Map替代数组存储新旧ID对应关系，避免循环查找
2. 批量插入基础记录，一次性拿到所有新生成的ID，和原ID对齐存入映射表，省去插入后回查的步骤
3. 直接在内存中遍历所有待克隆记录，根据ID映射替换关联字段里的所有旧ID，构造出完整的带正确关联的新记录
4. 批量执行写入/更新，不要单条循环发请求，控制好批量大小（建议每批500-1000条，避免单条SQL过长）

优化后参考代码：

const BATCH_SIZE = 500;

// 批量插入基础记录，一次性返回所有新旧ID映射
const batchCreateElements = async (elements) => {
    const idMap = new Map();
    // 分批处理避免单SQL过长
    for (let i = 0; i < elements.length; i += BATCH_SIZE) {
        const batch = elements.slice(i, i + BATCH_SIZE);
        // 批量INSERT，返回结果按插入顺序包含oldId和newId
        const batchResult = await dbQuery(`
            INSERT INTO elements (name, temp_old_id)
            VALUES ${batch.map(() => '(?, ?)').join(',')}
            RETURNING id as newId, temp_old_id as oldId
        `, batch.flatMap(item => [item.name, item.id]));
        
        batchResult.forEach(row => idMap.set(row.oldId, row.newId));
    }
    return idMap;
}

// 内存中替换所有关联ID，构造完整记录
const buildClonedRecords = (originRecords, idMap) => {
    return originRecords.map(record => {
        const newId = idMap.get(record.id);
        const newAssociations = record.associations.map(assoc => ({
            from: idMap.get(assoc.from),
            to: idMap.get(assoc.to)
        }));
        return {
            id: newId,
            name: record.name,
            associations: newAssociations
        };
    });
}

// 批量更新关联字段，不需要单条执行
const batchUpdateAssociations = async (clonedRecords) => {
    for (let i = 0; i < clonedRecords.length; i += BATCH_SIZE) {
        const batch = clonedRecords.slice(i, i + BATCH_SIZE);
        // 用批量UPDATE的方式一次更新一批
        await dbQuery(`
            UPDATE elements SET associations = tmp.associations
            FROM (VALUES ${batch.map(() => '(?::int, ?::jsonb)').join(',')}) as tmp(id, associations)
            WHERE elements.id = tmp.id
        `, batch.flatMap(item => [item.id, JSON.stringify(item.associations)]));
    }
}

const cloneElements = async (records) => {
    // 第一步：批量插基础记录，拿到ID映射
    const idMap = await batchCreateElements(records);
    // 第二步：内存里替换完所有关联ID
    const clonedRecords = buildClonedRecords(records, idMap);
    // 第三步：批量更新关联字段
    await batchUpdateAssociations(clonedRecords);
    return clonedRecords;
}

如果数据库使用自增序列ID，还可以提前申请一批连续的序列值，直接在内存中完成所有ID分配、关联替换工作，最后单次批量插入完整记录，全程只需要1-2次数据库交互，万级记录克隆也能在几百毫秒内完成。

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