TypeORM如何确认查询执行完成？NestJS Postgres动态表插入顺序控制问题

回调调用支持说明 #

TypeORM 的 queryRunner.query 方法原生仅支持 Promise 调用形式，不直接接受回调函数作为入参。如果需要使用回调风格，可以手动做一层Promise转回调的封装，示例如下：

// 封装回调形式的查询方法
function runQueryByCallback(
  queryRunner: QueryRunner,
  sql: string,
  params?: any[],
  callback?: (err: Error | null, result?: any) => void
) {
  queryRunner.query(sql, params)
    .then(res => callback?.(null, res))
    .catch(err => callback?.(err))
}

// 调用示例
runQueryByCallback(queryRunner, 'INSERT INTO "test" VALUES ($1, $2)', [1, 'test'], (err, res) => {
  if (err) {
    console.error('执行失败', err)
    return
  }
  console.log('执行成功', res)
})

大数据量插入报错问题优化 #

你当前代码的await insertInto写法本身已经可以保证上一条插入完全执行（事务提交完成）后才会执行下一条插入，执行顺序符合你的要求，大数据量报错的核心问题不在执行顺序，而是批量插入的逻辑不合理：

1. 直接拼接整段INSERT ... VALUES语句，1000万条数据拼接后的SQL长度会远超PostgreSQL默认的单条语句长度限制、参数数量上限，直接触发数据库层错误
2. 每次插入都单独创建连接、开启独立事务，高负载下连接开销和事务开销过大，也容易触发连接池耗尽报错

优化方案 #

• 拆分插入批次：将1000万条数据拆分为每批1000~5000条，单批次执行完成后再处理下一批，避免单条SQL过长
• 合并关联表事务：将两个有依赖关系的表插入放到同一个事务中，减少事务和连接的开销，同时保证数据一致性，任意一步失败都会全部回滚
• 可选优化：如果不需要强实时查询，插入前可临时关闭目标表的索引、调整PostgreSQL的wal_buffers、maintenance_work_mem参数提升写入性能，插入完成后再重建索引

优化后的代码示例：

export class Test {    
  constructor(
    private readonly connection: Connection,
  ) {}

  public async insertData(table1: string, table2: string, arr1: any[], arr2: any[], batchSize = 2000) {
    const queryRunner = this.connection.createQueryRunner();
    await queryRunner.connect();
    await queryRunner.startTransaction();

    try {
      // 分批插入表1
      await this.batchInsert(queryRunner, table1, arr1, batchSize);
      // 表1插入完成后再插入表2
      await this.batchInsert(queryRunner, table2, arr2, batchSize);

      await queryRunner.commitTransaction();
    } catch (e) {
      await queryRunner.rollbackTransaction();
      throw new InternalServerErrorException(e, 'Error while executing custom query. Rollback transaction.')
    } finally {
      await queryRunner.release();
    }
  }

  private async batchInsert(queryRunner: QueryRunner, table: string, data: any[], batchSize: number) {
    // 拆分批次
    for (let i = 0; i < data.length; i += batchSize) {
      const batch = data.slice(i, i + batchSize);
      // 预处理当前批次的插入数据，使用参数化查询避免SQL注入
      const placeholders = batch.map((_, idx) => `(${new Array(batch[0].length).fill(0).map((_, pos) => `$${idx * batch[0].length + pos + 1}`).join(',')})`).join(',');
      const params = batch.flat();
      await queryRunner.query(`INSERT INTO "${table}" VALUES ${placeholders}`, params);
    }
  }
}

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