如何在NestJS中使用Postgres的COPY命令实现大数据批量写入

你当前的写法不符合Postgres的COPY FROM STDIN协议要求：psql控制台能直接运行带行数据的COPY语句，是因为psql客户端自行处理了stdin输入逻辑，将COPY语句和后续行数据分开传输给数据库。而普通的SQL查询接口只能接收标准SQL语句，不支持直接在COPY语句后拼接行数据和结束符\.，因此会报语法错误。

要在Node.js/TypeORM场景下使用COPY批量插入，推荐使用pg-copy-streams库配合Postgres原生的流复制协议实现，性能可以满足千万级数据的写入需求。

步骤1：安装依赖 #

# 核心依赖
npm install pg-copy-streams
# TypeScript 类型补充（非TS项目可忽略）
npm install @types/pg-copy-streams -D

步骤2：代码实现 #

import { from } from 'pg-copy-streams';
import { Readable } from 'stream';
import { pipeline } from 'stream/promises';

// 业务逻辑代码
const queryRunner = this.connection.createQueryRunner();
await queryRunner.connect();
await queryRunner.startTransaction();

try {
  // 拿到TypeORM封装的原生Postgres客户端
  const pgClient = (queryRunner as any).databaseConnection;

  // 构造COPY语句，可根据需求调整格式、分隔符、空值标识等参数
  const copyQuery = `COPY ${table}(key) FROM STDIN WITH (FORMAT text, DELIMITER E'\t', NULL '\\N')`;
  const copyStream = pgClient.query(from(copyQuery));

  // 构造待插入数据的可读流，千万级数据建议边读边写避免内存溢出
  const dataRows = ['123', '321', '147'];
  const inputStream = Readable.from(
    dataRows.map(row => `${row}\t\n`).join('')
    // 多字段场景按COPY字段顺序拼接：`${field1}\t${field2}\t${field3}\n`
  );

  // 等待流传输完成
  await pipeline(inputStream, copyStream);

  await queryRunner.commitTransaction();
} catch (err) {
  await queryRunner.rollbackTransaction();
  throw err;
} finally {
  await queryRunner.release();
}

替代方案（百万级以内数据适用） #

如果你的数据量级不超过百万，不想引入额外依赖，可以使用Postgres的unnest函数实现批量插入，性能远高于逐行INSERT，写法更简单：

const keys = ['123', '321', '147'];
await queryRunner.query(
  `INSERT INTO ${table}(key) SELECT * FROM unnest($1::text[])`,
  [keys]
);

• 多字段插入时需严格按照COPY语句中指定的字段顺序拼接值，注意转义字段中的分隔符、换行符、反斜杠等特殊字符
• 千万级数据插入不要一次性加载所有数据到内存，需对接API的输入流边解析边写入COPY流，避免OOM
• CSV格式数据可修改COPY语句的FORMAT为csv，按需配置分隔符、引号、是否忽略表头即可

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