[TypeORM, MySQL] Bulk INSERT / UPDATE 하기

acho·2024년 7월 12일

배경

급히 구현하다보니 여러 행에 대한 처리가 필요할 때 for 문 안에서 쿼리를 실행하는 부분이 많이 있었습니다.
리팩토링을 진행하며 이런 부분을 모두 한 번의 쿼리로 처리할 수 있도록 변경했습니다.
그 중 TypeORM으로 여러 행을 INSERT, UPDATE 하는 방법에 대해 글을 써보려고 합니다.

쿼리 실행 횟수를 줄여야 하는 이유

와이어 샤크로 찍어본 Request Query 패킷

서버는 클라이언트와 통신할 때처럼 쿼리를 실행할 때마다 데이터베이스에 Request를 보내고, 응답을 기다리게 됩니다.
이 때 대기하는 동안 논리 흐름이 중단됩니다. 즉 CPU가 놀게 됩니다.
이를 I/O burst라고 합니다.
(I/O는 Input, Output을 뜻합니다. 네트워크 전송을 입력/출력으로 표현하는 이유는 컴퓨터 내부적으로 네트워크 요청을 처리하는 엔드포인트가 file discriptor이기 때문입니다.)
이처럼 쿼리를 날릴 때마다 I/O burst가 일어나기 때문에 쿼리 횟수는 최대한 줄이는 게 좋습니다.

TypeORM으로 bulk INSERT 하기

이전에는 TypeORM의 save 메소드를 사용하고 있었습니다.
save 메소드는 1) SELECT 쿼리로 중복 행이 존재하는지 확인한 뒤 2) 있으면 UPDATE, 없으면 INSERT 하는 두 단계로 나뉩니다.
이 때문에 인자로 엔티티의 배열을 넘겨도 한번에 INSERT 하지 않고, 한 행씩 SELECT - INSERT 하게 됩니다.

SELECT `Excrements`.`id` AS `Excrements_id`, `Excrements`.`journal_id` AS `Excrements_journal_id`, `Excrements`.`dog_id` AS `Excrements_dog_id`, `Excrements`.`type` AS `Excrements_type`, ST_AsText(`Excrements`.`coordinate`) AS `Excrements_coordinate` FROM `excrements` `Excrements` WHERE ((`Excrements`.`journal_id` = ?) AND (`Excrements`.`dog_id` = ?) AND (`Excrements`.`type` = ?)) LIMIT 1 -- PARAMETERS: [228,1,"FECES"]
INSERT INTO `excrements`(`id`, `journal_id`, `dog_id`, `type`, `coordinate`) VALUES (DEFAULT, ?, ?, ?, ST_GeomFromText(?, 4326)) -- PARAMETERS: [228,1,"FECES","POINT(10.4 180)"]
SELECT `Excrements`.`id` AS `Excrements_id`, `Excrements`.`journal_id` AS `Excrements_journal_id`, `Excrements`.`dog_id` AS `Excrements_dog_id`, `Excrements`.`type` AS `Excrements_type`, ST_AsText(`Excrements`.`coordinate`) AS `Excrements_coordinate` FROM `excrements` `Excrements` WHERE ((`Excrements`.`journal_id` = ?) AND (`Excrements`.`dog_id` = ?) AND (`Excrements`.`type` = ?)) LIMIT 1 -- PARAMETERS: [228,1,"URINE"]
INSERT INTO `excrements`(`id`, `journal_id`, `dog_id`, `type`, `coordinate`) VALUES (DEFAULT, ?, ?, ?, ST_GeomFromText(?, 4326)) -- PARAMETERS: [228,1,"URINE","POINT(30.4 180)"]

TypeORM의 또 다른 메소드인 insert는 이러한 확인 과정을 거치지 않습니다.
이에 따라 인자로 받은 엔티티를 한 번에 INSERT 해줍니다.

insert(entity: QueryDeepPartialEntity<Entity> | QueryDeepPartialEntity<Entity>[]): Promise<InsertResult>;

Entity의 배열을 인자로 받아 InsertResult를 반환합니다.

INSERT INTO `excrements`(`id`, `journal_id`, `dog_id`, `type`, `coordinate`) VALUES (DEFAULT, ?, ?, ?, ST_GeomFromText(?, 4326)), (DEFAULT, ?, ?, ?, ST_GeomFromText(?, 4326)), (DEFAULT, ?, ?, ?, ST_GeomFromText(?, 4326)), (DEFAULT, ?, ?, ?, ST_GeomFromText(?, 4326)), (DEFAULT, ?, ?, ?, ST_GeomFromText(?, 4326)), (DEFAULT, ?, ?, ?, ST_GeomFromText(?, 4326)) -- PARAMETERS: [227,1,"FECES","POINT(10.4 180)",227,1,"URINE","POINT(30.4 180)",227,2,"FECES","POINT(10.4 180)",227,2,"URINE","POINT(30.4 180)",227,3,"FECES","POINT(10.4 180)",227,3,"URINE","POINT(30.4 180)"]

확인 절차를 거치지 않는 만큼, 중복 조건에 걸리는 행이 있는 경우 에러가 발생하니 주의가 필요합니다.

TypeORM으로 bulk UPDATE 하기

bulk UPDATE는 조금 더 복잡합니다.
우선 여러 행을 업데이트 하되 모두 같은 값으로 업데이트 할 때는, Repository에 정의된 update 메소드를 사용할 수 있습니다.

await update({ id: In([1, 2, 3]) }, 10);

하지만 각 행을 서로 다른 값으로 UPDATE 할 때는 다른 방법이 필요합니다.

update(criteria: string | string[] | number | number[] | Date | Date[] | ObjectId | ObjectId[] | FindOptionsWhere<Entity>, partialEntity: QueryDeepPartialEntity<Entity>): Promise<UpdateResult>;

2번 인자인 partialEntity: QueryDeepPartialEntity가 업데이트할 엔티티인데, 어째서인지 배열 형태로 받지 못하게 되어있습니다.

우선 SQL로는 어떻게 구현하는지 찾아보았습니다.

1. SELECT - CASE

UPDATE a 
SET fruit = (CASE id WHEN 1 THEN 'apple'
                     WHEN 2 THEN 'orange'
                     WHEN 3 THEN 'peach'
             END)
WHERE id IN(1,2 ,3);

SELECT -CASE 구문과 WHERE - IN 을 조합해 조건에 따라 다른 값으로 행을 업데이트 하는 방법입니다.

2. IF

UPDATE a
SET fruit = IF (id = 1, 'apple', 
			IF (id = 2, 'orange', 
            IF (id = 3, 'peach', fruit)));

IF를 활용하는 방법입니다. 위와 대체로 비슷합니다.

3. ON DUPLICATE KEY

INSERT into `table` (id,fruit)
    VALUES (1,'apple'), (2,'orange'), (3,'peach')
    ON DUPLICATE KEY UPDATE fruit = VALUES(fruit);

INSERT 하되, 중복되는 행이 있으면 해당하는 값으로 UPDATE하는 방법입니다. 대체로 가장 추천되는 방법이었습니다.
실제로 INSERT를 하지 않는데도 PK가 증가해 추천하지 않는다는 글을 보았는데 보았는데, 실제로 해보니 증가하지 않았습니다. MySQL 버전이 업데이트 되며 고쳐진 버그인 것 같습니다.

3번이 구문도 간단하고, typeorm에 구현된 메소드도 있어 사용하기로 했습니다.

.orUpdate()

Reposiotry에는 없고, QueryBuilder에 정의되어 있는 메소드입니다.

orUpdate(overwrite: string[], conflictTarget?: string | string[], orUpdateOptions?: InsertOrUpdateOptions): this;
    /**
     * Creates INSERT express used to perform insert query.
     */

1번 인자 : 업데이트할 컬럼입니다.
2번 인자 : 중복 체크할 컬럼입니다.

await this.entityManager
            .createQueryBuilder(TodayWalkTime, 'todayWalkTime')
            .insert()
            .into(TodayWalkTime, ['id', 'duration'])
            .values(updateEntities)
            .orUpdate(['duration'], ['id'])
            .execute();

values로 받은 값에서 duration 컬럼의 값을 id가 중복된다면 update 합니다.

아래와 같은 쿼리가 실행됩니다.

INSERT INTO `today_walk_time`(`id`, `duration`) VALUES (?, ?), (?, ?), (?, ?) ON DUPLICATE KEY UPDATE `duration` = VALUES(`duration`) -- PARAMETERS: [1,172801,2,172802,3,205200]