Upsert

Upsert는 "Update"와 "Insert"의 합성어로, 하나의 연산으로 데이터가 존재하면 업데이트하고 존재하지 않으면 새로 삽입하는 데이터베이스 기능입니다. 전통적으로 "SELECT로 확인 → 있으면 UPDATE, 없으면 INSERT" 패턴은 두 쿼리 사이에 경쟁 조건(Race Condition)이 발생할 수 있지만, Upsert는 원자적(Atomic)으로 처리하여 이 문제를 해결합니다.

각 데이터베이스마다 Upsert를 구현하는 문법이 다릅니다. PostgreSQL은 `INSERT ... ON CONFLICT DO UPDATE`, MySQL은 `INSERT ... ON DUPLICATE KEY UPDATE` 또는 `REPLACE INTO`, SQL Server는 `MERGE` 문을 사용합니다. MongoDB는 `updateOne/updateMany`에 `upsert: true` 옵션을 제공합니다. ORM에서도 대부분 upsert 메서드를 지원합니다.

Upsert의 핵심은 "충돌 감지 키"입니다. 어떤 컬럼을 기준으로 존재 여부를 판단할지 명시해야 합니다. PostgreSQL의 경우 PRIMARY KEY나 UNIQUE 제약 조건이 있는 컬럼을 `ON CONFLICT (column)` 절에 지정합니다. 복합 키(Composite Key)도 지원되며, 부분 인덱스(Partial Index)와 조합하여 더 세밀한 충돌 감지도 가능합니다.

Upsert가 특히 유용한 사용 사례로는 외부 API 데이터 동기화, 이벤트 소싱에서 상태 집계, 캐시 업데이트, 사용자 마지막 활동 시간 갱신 등이 있습니다. 대량 데이터 일괄 처리 시에도 개별 존재 여부를 확인하지 않고 한 번에 처리할 수 있어 성능 향상에 크게 기여합니다.

PostgreSQL: INSERT ON CONFLICT (권장)

-- 테이블 생성 (email이 UNIQUE)
CREATE TABLE users (
    id SERIAL PRIMARY KEY,
    email VARCHAR(255) UNIQUE NOT NULL,
    name VARCHAR(255),
    login_count INT DEFAULT 0,
    last_login TIMESTAMP,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- 기본 Upsert: email 충돌 시 name과 last_login 업데이트
INSERT INTO users (email, name, last_login)
VALUES ('user@example.com', 'John Doe', NOW())
ON CONFLICT (email)
DO UPDATE SET
    name = EXCLUDED.name,
    last_login = EXCLUDED.last_login,
    updated_at = CURRENT_TIMESTAMP;

-- login_count 증가하는 Upsert
INSERT INTO users (email, name, login_count, last_login)
VALUES ('user@example.com', 'John Doe', 1, NOW())
ON CONFLICT (email)
DO UPDATE SET
    login_count = users.login_count + 1,
    last_login = EXCLUDED.last_login,
    updated_at = CURRENT_TIMESTAMP;

-- 조건부 업데이트 (WHERE 절)
INSERT INTO users (email, name)
VALUES ('user@example.com', 'New Name')
ON CONFLICT (email)
DO UPDATE SET
    name = EXCLUDED.name
WHERE users.name != EXCLUDED.name;  -- 이름이 다를 때만 업데이트

-- 충돌 시 아무것도 안 함 (삽입만 시도)
INSERT INTO users (email, name)
VALUES ('user@example.com', 'John Doe')
ON CONFLICT (email) DO NOTHING;

-- Upsert 결과 확인 (RETURNING)
INSERT INTO users (email, name)
VALUES ('user@example.com', 'John Doe')
ON CONFLICT (email)
DO UPDATE SET name = EXCLUDED.name
RETURNING id, email,
    (xmax = 0) AS inserted;  -- true면 INSERT, false면 UPDATE

MySQL: ON DUPLICATE KEY UPDATE

-- MySQL 테이블
CREATE TABLE products (
    id INT PRIMARY KEY AUTO_INCREMENT,
    sku VARCHAR(50) UNIQUE NOT NULL,
    name VARCHAR(255),
    price DECIMAL(10,2),
    stock INT DEFAULT 0,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
);

-- 기본 Upsert
INSERT INTO products (sku, name, price, stock)
VALUES ('SKU-001', 'Product A', 29.99, 100)
ON DUPLICATE KEY UPDATE
    name = VALUES(name),
    price = VALUES(price),
    stock = VALUES(stock);

-- MySQL 8.0+: 별칭 사용 (VALUES() deprecated)
INSERT INTO products (sku, name, price, stock)
VALUES ('SKU-001', 'Product A', 29.99, 100) AS new_values
ON DUPLICATE KEY UPDATE
    name = new_values.name,
    price = new_values.price,
    stock = new_values.stock;

-- 재고 증가 (기존 값 활용)
INSERT INTO products (sku, name, price, stock)
VALUES ('SKU-001', 'Product A', 29.99, 50)
ON DUPLICATE KEY UPDATE
    stock = stock + VALUES(stock);

-- REPLACE INTO (주의: 기존 row 삭제 후 INSERT)
-- AUTO_INCREMENT가 변경되고 연관 CASCADE 삭제 발생 가능
REPLACE INTO products (sku, name, price, stock)
VALUES ('SKU-001', 'Product A', 29.99, 100);

MongoDB: updateOne with upsert

// MongoDB upsert 예제
const { MongoClient } = require('mongodb');

async function upsertExamples() {
    const client = new MongoClient('mongodb://localhost:27017');
    await client.connect();
    const db = client.db('myapp');
    const users = db.collection('users');

    // 기본 Upsert: email로 찾아서 업데이트, 없으면 삽입
    const result = await users.updateOne(
        { email: 'user@example.com' },  // 필터 (찾을 조건)
        {
            $set: {
                name: 'John Doe',
                lastLogin: new Date()
            },
            $setOnInsert: {
                // INSERT 시에만 설정되는 필드
                createdAt: new Date(),
                loginCount: 0
            },
            $inc: {
                loginCount: 1  // 항상 증가
            }
        },
        { upsert: true }  // upsert 활성화
    );

    console.log(`Matched: ${result.matchedCount}, Modified: ${result.modifiedCount}, Upserted: ${result.upsertedCount}`);

    // 벌크 Upsert (대량 처리)
    const bulkOps = [
        { email: 'user1@example.com', name: 'User 1', status: 'active' },
        { email: 'user2@example.com', name: 'User 2', status: 'active' },
        { email: 'user3@example.com', name: 'User 3', status: 'inactive' }
    ].map(user => ({
        updateOne: {
            filter: { email: user.email },
            update: {
                $set: { name: user.name, status: user.status },
                $setOnInsert: { createdAt: new Date() }
            },
            upsert: true
        }
    }));

    const bulkResult = await users.bulkWrite(bulkOps);
    console.log(`Bulk upsert - Inserted: ${bulkResult.upsertedCount}, Modified: ${bulkResult.modifiedCount}`);

    // findOneAndUpdate with upsert (업데이트된 문서 반환)
    const updatedUser = await users.findOneAndUpdate(
        { email: 'user@example.com' },
        {
            $set: { lastLogin: new Date() },
            $inc: { loginCount: 1 }
        },
        {
            upsert: true,
            returnDocument: 'after'  // 업데이트 후 문서 반환
        }
    );
    console.log(updatedUser);

    await client.close();
}

upsertExamples();

Python SQLAlchemy ORM Upsert

from sqlalchemy import create_engine, Column, Integer, String, DateTime, func
from sqlalchemy.orm import sessionmaker, declarative_base
from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.dialects.mysql import insert as mysql_insert
from datetime import datetime

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True, autoincrement=True)
    email = Column(String(255), unique=True, nullable=False)
    name = Column(String(255))
    login_count = Column(Integer, default=0)
    last_login = Column(DateTime)
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)


# PostgreSQL Upsert
def upsert_user_postgres(session, email: str, name: str) -> User:
    """PostgreSQL ON CONFLICT DO UPDATE"""
    stmt = pg_insert(User).values(
        email=email,
        name=name,
        login_count=1,
        last_login=datetime.utcnow()
    )

    # 충돌 시 업데이트할 컬럼 지정
    stmt = stmt.on_conflict_do_update(
        index_elements=['email'],  # 충돌 감지 컬럼
        set_={
            'name': stmt.excluded.name,
            'login_count': User.login_count + 1,
            'last_login': stmt.excluded.last_login,
            'updated_at': datetime.utcnow()
        }
    ).returning(User)

    result = session.execute(stmt)
    session.commit()
    return result.scalar_one()


# MySQL Upsert
def upsert_user_mysql(session, email: str, name: str) -> int:
    """MySQL ON DUPLICATE KEY UPDATE"""
    stmt = mysql_insert(User).values(
        email=email,
        name=name,
        login_count=1,
        last_login=datetime.utcnow()
    )

    stmt = stmt.on_duplicate_key_update(
        name=stmt.inserted.name,
        login_count=User.login_count + 1,
        last_login=stmt.inserted.last_login,
        updated_at=datetime.utcnow()
    )

    result = session.execute(stmt)
    session.commit()
    return result.lastrowid


# 벌크 Upsert (PostgreSQL)
def bulk_upsert_users(session, users_data: list[dict]) -> int:
    """여러 사용자를 한 번에 upsert"""
    stmt = pg_insert(User).values(users_data)

    stmt = stmt.on_conflict_do_update(
        index_elements=['email'],
        set_={
            'name': stmt.excluded.name,
            'login_count': User.login_count + 1,
            'last_login': stmt.excluded.last_login,
            'updated_at': datetime.utcnow()
        }
    )

    result = session.execute(stmt)
    session.commit()
    return result.rowcount


# 사용 예시
if __name__ == "__main__":
    engine = create_engine("postgresql://user:pass@localhost/mydb")
    Session = sessionmaker(bind=engine)
    session = Session()

    # 단일 upsert
    user = upsert_user_postgres(session, "john@example.com", "John Doe")
    print(f"Upserted user: {user.email}, login_count: {user.login_count}")

    # 벌크 upsert
    users_data = [
        {"email": "user1@example.com", "name": "User 1", "login_count": 1, "last_login": datetime.utcnow()},
        {"email": "user2@example.com", "name": "User 2", "login_count": 1, "last_login": datetime.utcnow()},
        {"email": "user3@example.com", "name": "User 3", "login_count": 1, "last_login": datetime.utcnow()},
    ]
    count = bulk_upsert_users(session, users_data)
    print(f"Bulk upserted {count} users")

    session.close()

Redis: 키 기반 자연스러운 Upsert

import redis
import json
from datetime import datetime

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

# Redis는 기본적으로 SET이 upsert 동작
# 키가 있으면 덮어쓰고, 없으면 생성

def upsert_user_session(user_id: str, session_data: dict):
    """사용자 세션 upsert"""
    key = f"session:{user_id}"
    session_data['updated_at'] = datetime.utcnow().isoformat()

    # 기존 데이터 유지하면서 업데이트 (HSET)
    r.hset(key, mapping=session_data)
    r.expire(key, 3600)  # 1시간 TTL

def upsert_with_conditional(user_id: str, field: str, value: str):
    """조건부 upsert: 필드가 없을 때만 설정 (HSETNX)"""
    key = f"user:{user_id}"
    return r.hsetnx(key, field, value)  # 1이면 삽입, 0이면 이미 존재

def upsert_counter(key: str, increment: int = 1):
    """카운터 upsert: 없으면 생성하고 증가"""
    return r.incrby(key, increment)  # 키가 없으면 0에서 시작

# 사용 예시
upsert_user_session("user123", {
    "ip": "192.168.1.1",
    "user_agent": "Mozilla/5.0",
    "last_page": "/dashboard"
})

# 첫 로그인 시간은 최초 1회만 설정
upsert_with_conditional("user123", "first_login", datetime.utcnow().isoformat())

# 페이지뷰 카운터
page_views = upsert_counter("pageviews:homepage:2024-01-15")
print(f"Today's homepage views: {page_views}")