PostgreSQL批量插入如何避免回滚并返回失败行？Node pg适用

刚好碰到过类似的需求，来给你梳理几个可行的方案，分场景来看：

1. 处理唯一约束/主键冲突（最常用场景） #

如果你的插入失败只是因为唯一约束（比如email重复）或者主键冲突，那PostgreSQL自带的ON CONFLICT语法就能搞定，既不会回滚成功的行，还能直接拿到失败的行列表。

举个例子，假设你的test表email字段是唯一约束：

WITH input_rows AS (
  VALUES 
    ('abcd', 'abcd@example.com'),
    ('efgh', 'efgh@example.com'),
    ('abcd', 'abcd@example.com') -- 这条会因为email重复失败
),
inserted AS (
  INSERT INTO test(name, email)
  SELECT name, email FROM input_rows
  ON CONFLICT (email) DO NOTHING -- 冲突时跳过该行
  RETURNING *
)
-- 合并成功和失败的结果
SELECT 
  'success' AS status, name, email FROM inserted
UNION ALL
SELECT 
  'failed' AS status, name, email FROM input_rows
WHERE (name, email) NOT IN (SELECT name, email FROM inserted);

执行这条SQL后，会直接返回所有行的状态，成功的标success，失败的标failed，一目了然。

2. 处理任意类型的插入错误（通用方案） #

如果失败是因为非空约束、类型不匹配这类问题，上面的方法就不管用了——PostgreSQL会直接让整个INSERT语句报错，导致全量回滚。这时候推荐用PL/pgSQL函数来逐行处理，捕获异常，既保留成功的行，还能返回失败行的错误信息。

先在数据库里创建一个函数：

CREATE OR REPLACE FUNCTION bulk_insert_test(inputs JSONB)
RETURNS TABLE(status TEXT, name TEXT, email TEXT, error TEXT) AS $$
DECLARE
  input RECORD;
BEGIN
  -- 遍历传入的JSON数组
  FOR input IN SELECT * FROM jsonb_to_recordset(inputs) AS x(name TEXT, email TEXT) LOOP
    BEGIN
      -- 尝试插入当前行
      INSERT INTO test(name, email) VALUES (input.name, input.email);
      -- 插入成功，返回success状态
      RETURN NEXT ('success', input.name, input.email, NULL);
    EXCEPTION
      -- 捕获所有异常，返回失败状态和错误信息
      WHEN OTHERS THEN
        RETURN NEXT ('failed', input.name, input.email, SQLERRM);
    END;
  END LOOP;
END;
$$ LANGUAGE plpgsql;

然后在Node.js的pg模块里调用这个函数就行：

const { Pool } = require('pg');
const pool = new Pool({ /* 这里填你的数据库连接配置 */ });

async function batchInsert() {
  const data = [
    { name: 'abcd', email: 'abcd@example.com' },
    { name: 'efgh', email: null }, // 违反email非空约束
    { name: 'ijkl', email: 'ijkl@example.com' }
  ];

  const result = await pool.query(
    'SELECT * FROM bulk_insert_test($1::JSONB)',
    [JSON.stringify(data)]
  );

  console.log('插入结果：', result.rows);
  // 过滤出失败的行
  const failedRows = result.rows.filter(row => row.status === 'failed');
  console.log('失败的行：', failedRows);
}

batchInsert();

这个函数会自动给每一行创建独立的保存点，遇到错误只会跳过当前行，不会影响已经插入成功的行，而且能返回具体的错误原因。

3. Node端直接逐行处理（不用改数据库） #

如果不想在数据库里创建函数，也可以直接在Node端循环执行单条INSERT，用Promise.allSettled来处理每个请求的结果：

const { Pool } = require('pg');
const pool = new Pool({ /* 你的连接配置 */ });

async function batchInsert() {
  const rows = [
    { name: 'abcd', email: 'abcd@example.com' },
    { name: 'efgh', email: null },
    { name: 'ijkl', email: 'ijkl@example.com' }
  ];

  // 把每一行的插入变成Promise
  const insertPromises = rows.map(async (row) => {
    try {
      await pool.query('INSERT INTO test(name, email) VALUES ($1, $2)', [row.name, row.email]);
      return { ...row, status: 'success', error: null };
    } catch (err) {
      return { ...row, status: 'failed', error: err.message };
    }
  });

  // 等待所有Promise完成（不管成功失败）
  const results = await Promise.allSettled(insertPromises);
  // 整理成统一格式的结果
  const finalResults = results.map(res => {
    if (res.status === 'fulfilled') return res.value;
    return { ...res.reason.row, status: 'failed', error: res.reason.message };
  });

  console.log('插入结果：', finalResults);
  const failedRows = finalResults.filter(r => r.status === 'failed');
  console.log('失败的行：', failedRows);
}

batchInsert();

这个方法的好处是不用动数据库，但缺点是性能会差一些——每一行都要发一个独立的SQL请求，适合数据量不大的场景。

4. 临时表预处理（大数据量场景） #

如果要插入的数据量特别大，上面的逐行处理性能不够，可以先把所有数据插入临时表，再从临时表筛选符合约束的行插入目标表，同时收集失败的行：

-- 创建和目标表结构一致的临时表
CREATE TEMP TABLE temp_test (LIKE test INCLUDING CONSTRAINTS);

-- 批量插入所有数据到临时表（如果数据量超大，推荐用pg的COPY命令，比INSERT快很多）
INSERT INTO temp_test(name, email) VALUES 
  ('abcd', 'abcd@example.com'),
  ('efgh', NULL),
  ('ijkl', 'ijkl@example.com');

-- 插入符合条件的行到目标表，同时返回结果
WITH inserted AS (
  INSERT INTO test(name, email)
  SELECT name, email FROM temp_test
  WHERE email IS NOT NULL -- 这里根据你的实际约束添加检查条件
  RETURNING *
)
SELECT 
  'success' AS status, name, email FROM inserted
UNION ALL
SELECT 
  'failed' AS status, name, email FROM temp_test
WHERE (name, email) NOT IN (SELECT name, email FROM inserted);

-- 用完删除临时表
DROP TABLE temp_test;

不过这个方法需要你提前明确所有的约束条件，手动做检查，否则插入临时表的时候还是会报错。如果不确定约束，还是用PL/pgSQL的方案更稳妥。

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