Postgres가 모든 것을 한다: 하나의 데이터베이스에서 queue, search, vector

1. 개요

일반적인 시스템은 여러 구성 요소를 갖고 있습니다: PostgreSQL이 데이터를 저장하고, message broker가 queue를 처리하고, search engine이 검색을 담당하고, 최근에는 AI 기능을 위한 vector database까지 추가됩니다. 각 구성 요소마다 운영, 모니터링, 백업 및 데이터 동기화의 부담이 따릅니다.

“Postgres가 모든 것을 한다”는 추세는 반대로 질문을 던집니다: PostgreSQL이 이미 충분히 잘 대부분의 작업을 할 수 있다면, 새로운 시스템이 정말 필요할까요? 이 글에서는 Spring Boot 3.x와 함께 PostgreSQL을 job queue, full-text search, vector search로 사용할 것입니다. 또한 언제 멈추고 전문 도구를 선택해야 하는지도 알게 될 것입니다.

2. 이 추세가 주목받는 이유

몇 가지 주요 이유가 있습니다:

• 운영 복잡성 감소. 추가하는 각 시스템은 밤중에 깨질 수 있는 또 다른 것입니다. 작은 팀에서는 이 비용이 성능상의 이점보다 보통 더 큽니다.
• 트랜잭션 일관성. 주문과 이메일 발송 job이 같은 데이터베이스에 있을 때, 하나의 트랜잭션으로 둘 다 저장합니다. 주문은 저장되었는데 메시지가 손실되거나 그 반대인 상황이 없습니다.
• Extension 생태계. PostgreSQL은 강력한 extension 메커니즘을 갖고 있습니다. 예를 들어 pgvector는 vector 데이터 타입과 k-nearest neighbor 검색 인덱스를 추가합니다.
• AI 기능이 embedding 저장 수요를 가져옵니다. 많은 팀이 semantic search 기능을 추가하고 싶지만 그것만을 위해 새로운 데이터베이스를 구축하고 싶지 않습니다.

이 추세의 정신은 단순하게 시작하고, 필요함을 증명하는 데이터가 있을 때 분리하는 것입니다. PostgreSQL이 모든 시스템을 대체할 수 있다는 뜻은 아닙니다.

3. 준비

3.1. Maven 의존성

JDBC, PostgreSQL 드라이버, 테스트를 위한 Testcontainers가 필요합니다. Spring Boot가 버전을 관리하므로 선언할 필요가 없습니다:

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-jdbc</artifactId>
    </dependency>
    <dependency>
        <groupId>org.postgresql</groupId>
        <artifactId>postgresql</artifactId>
        <scope>runtime</scope>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-testcontainers</artifactId>
        <scope>test</scope>
    </dependency>
    <dependency>
        <groupId>org.testcontainers</groupId>
        <artifactId>postgresql</artifactId>
        <scope>test</scope>
    </dependency>
    <dependency>
        <groupId>org.testcontainers</groupId>
        <artifactId>junit-jupiter</artifactId>
        <scope>test</scope>
    </dependency>
</dependencies>

3.2. pgvector가 포함된 PostgreSQL 실행

pgvector/pgvector 이미지는 PostgreSQL과 vector extension을 미리 패키징했습니다. 사용 중인 PostgreSQL 주 버전에 해당하는 태그를 선택하세요:

services:
  postgres:
    image: pgvector/pgvector:pg16
    environment:
      POSTGRES_DB: shop
      POSTGRES_USER: shop
      POSTGRES_PASSWORD: secret
    ports:
      - "5432:5432"

연결하고 시작 시 자동으로 schema.sql을 실행하도록 Spring Boot를 구성하세요:

spring.datasource.url=jdbc:postgresql://localhost:5432/shop
spring.datasource.username=shop
spring.datasource.password=secret
spring.sql.init.mode=always

실제 프로젝트에서는 schema.sql 대신 Flyway나 Liquibase를 사용해야 합니다.

4. FOR UPDATE SKIP LOCKED를 사용한 Job queue

4.1. 스키마

최소한의 queue는 하나의 테이블만 필요합니다:

CREATE TABLE IF NOT EXISTS job_queue (
    id         BIGSERIAL PRIMARY KEY,
    job_type   TEXT        NOT NULL,
    payload    JSONB       NOT NULL,
    run_at     TIMESTAMPTZ NOT NULL DEFAULT now(),
    created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE INDEX IF NOT EXISTS idx_job_queue_run_at ON job_queue (run_at);

run_at 컬럼은 job 실행을 예약할 수 있습니다. run_at의 인덱스는 worker가 전체 테이블을 스캔하지 않고 가장 빨리 실행되어야 하는 job을 가져올 수 있도록 합니다.

4.2. Job 가져오기 및 처리

핵심은 FOR UPDATE SKIP LOCKED 절입니다. 많은 worker가 동시에 쿼리를 실행합니다. 각 worker는 한 행을 잠그고, 다른 worker는 잠긴 행을 건너뛰고 기다리지 않습니다:

public record Job(long id, String type, String payload) {}

@Repository
public class JobQueue {

    private final JdbcTemplate jdbc;

    public JobQueue(JdbcTemplate jdbc) {
        this.jdbc = jdbc;
    }

    public void enqueue(String type, String payloadJson) {
        jdbc.update("INSERT INTO job_queue (job_type, payload) VALUES (?, ?::jsonb)",
                type, payloadJson);
    }

    @Transactional
    public boolean processNext(Consumer<Job> handler) {
        List<Job> jobs = jdbc.query("""
                SELECT id, job_type, payload::text
                FROM job_queue
                WHERE run_at <= now()
                ORDER BY run_at
                LIMIT 1
                FOR UPDATE SKIP LOCKED
                """,
                (rs, i) -> new Job(rs.getLong(1), rs.getString(2), rs.getString(3)));

        if (jobs.isEmpty()) {
            return false;
        }
        Job job = jobs.getFirst();
        handler.accept(job);
        jdbc.update("DELETE FROM job_queue WHERE id = ?", job.id());
        return true;
    }
}

설명:

• enqueue는 job을 삽입합니다. ?::jsonb는 JSON 문자열을 JSONB 타입으로 캐스팅합니다.
• processNext는 트랜잭션 내에서 실행됩니다. SELECT ... FOR UPDATE SKIP LOCKED 쿼리는 아직 아무도 받지 않은 정확히 하나의 job을 잠급니다.
• handler가 성공하면, job은 삭제되고 트랜잭션이 커밋됩니다.
• handler가 예외를 던지면, 트랜잭션이 롤백됩니다. Job은 여전히 테이블에 남아 있으며 다시 처리됩니다. Worker가 도중에 죽으면, PostgreSQL은 연결이 닫힐 때 자동으로 잠금을 해제합니다.

Worker는 간단한 @Scheduled 메서드가 될 수 있습니다:

@Component
public class EmailWorker {

    private final JobQueue jobQueue;

    public EmailWorker(JobQueue jobQueue) {
        this.jobQueue = jobQueue;
    }

    @Scheduled(fixedDelay = 1000)
    public void poll() {
        while (jobQueue.processNext(job ->
                System.out.println("Job " + job.id() + "에 이메일 발송: " + job.payload()))) {
            // job이 남지 않을 때까지 계속
        }
    }
}

설정 클래스에 @EnableScheduling을 추가하는 것을 기억하세요. Job이 있을 때 콘솔에 다음이 출력됩니다:

Job 1에 이메일 발송: {"orderId": 42, "email": "an@example.com"}

참고: job 처리 중에 트랜잭션을 유지하는 방식은 매우 간단하고 worker가 충돌할 때 안전합니다. 대신 트랜잭션은 처리 시간 동안 지속됩니다. Job이 오래 실행되면(느린 API 호출, 큰 파일 처리), 상태 컬럼과 locked_until이 있는 모델로 전환하거나 전문 queue 라이브러리로 이동해야 합니다. 또한 하나의 실패한 job이 무한정 재시도되는 것을 피하기 위해 attempts 컬럼을 추가해야 합니다.

4.3. 실제 이점: 비즈니스 트랜잭션과 함께 enqueue

이것이 Postgres를 queue로 사용하는 가장 강력한 이유입니다:

@Service
public class OrderService {

    private final JdbcTemplate jdbc;
    private final JobQueue jobQueue;

    public OrderService(JdbcTemplate jdbc, JobQueue jobQueue) {
        this.jdbc = jdbc;
        this.jobQueue = jobQueue;
    }

    @Transactional
    public void placeOrder(long orderId, String customerEmail) {
        jdbc.update("INSERT INTO orders (id, customer_email) VALUES (?, ?)", orderId, customerEmail);
        jobQueue.enqueue("SEND_CONFIRMATION_EMAIL",
                "{\"orderId\": %d, \"email\": \"%s\"}".formatted(orderId, customerEmail));
    }
}

주문과 job이 함께 커밋되거나 함께 롤백됩니다. 별도의 broker를 사용하면 이러한 보장을 얻기 위해 transactional outbox 패턴을 직접 구축해야 합니다. (예제는 간결함을 위해 formatted를 사용합니다. 실제 코드는 Jackson을 사용하여 payload를 직렬화하여 문자를 올바르게 escape해야 합니다.)

Worker가 계속 polling하는 것을 원하지 않으면 PostgreSQL의 LISTEN/NOTIFY를 결합하여 새 job이 있을 때 worker를 깨울 수 있습니다. Polling은 여전히 폴백으로 유지되어야 합니다.

5. tsvector와 GIN index를 사용한 Full-text search

5.1. 스키마

PostgreSQL은 내장된 full-text search를 가지고 있습니다. 데이터가 변경될 때마다 생성된 컬럼을 사용하여 검색 벡터를 자동 업데이트합니다:

CREATE TABLE IF NOT EXISTS product (
    id            BIGSERIAL PRIMARY KEY,
    name          TEXT NOT NULL,
    description   TEXT NOT NULL,
    search_vector tsvector GENERATED ALWAYS AS (
        setweight(to_tsvector('simple', name), 'A') ||
        setweight(to_tsvector('simple', description), 'B')
    ) STORED
);

CREATE INDEX IF NOT EXISTS idx_product_search ON product USING GIN (search_vector);

설명:

• to_tsvector('simple', ...)는 텍스트를 토큰으로 분할합니다. simple 구성은 소문자로만 변환하고 stemming을 하지 않습니다. 이 선택은 PostgreSQL이 기본적으로 베트남어 사전을 갖고 있지 않기 때문에 베트남어에 적합합니다.
• setweight는 이름에 A 가중치, 설명에 B 가중치를 할당합니다. 이름에서의 일치가 설명에서의 일치보다 높은 순위를 얻습니다.
• GIN 인덱스는 tsvector 컬럼에서 빠른 쿼리를 돕습니다.

5.2. 쿼리

public record ProductHit(long id, String name, double rank) {}

@Repository
public class ProductSearch {

    private final JdbcTemplate jdbc;

    public ProductSearch(JdbcTemplate jdbc) {
        this.jdbc = jdbc;
    }

    public List<ProductHit> search(String userQuery) {
        return jdbc.query("""
                SELECT id, name, ts_rank(search_vector, q) AS rank
                FROM product, websearch_to_tsquery('simple', ?) AS q
                WHERE search_vector @@ q
                ORDER BY rank DESC
                LIMIT 10
                """,
                (rs, i) -> new ProductHit(rs.getLong("id"), rs.getString("name"), rs.getDouble("rank")),
                userQuery);
    }
}

websearch_to_tsquery는 일반적인 검색 상자와 유사한 구문을 허용합니다: "áo khoác" -len은 “áo khoác” 구문을 포함하고 “len”을 포함하지 않는 것을 의미합니다. @@ 연산자는 벡터가 쿼리와 일치하는지 확인합니다. ts_rank는 결과를 정렬하는 데 사용됩니다.

“Áo khoác gió” 및 “Quần jean”(설명에 “khoác ngoài”이 있음)을 포함하는 샘플 데이터로, khoác 쿼리는 다음을 반환합니다:

ProductHit[id=1, name=Áo khoác gió, rank=0.6079271]
ProductHit[id=2, name=Quần jean, rank=0.24317084]

참고: 사용자는 종종 악센트 없이 입력합니다(“ao khoac”). unaccent extension을 사용할 수 있습니다. 기본적으로 unaccent 함수는 IMMUTABLE로 표시되지 않으므로 생성된 컬럼이나 인덱스에서 사용하려면 자신의 IMMUTABLE 함수에 래핑해야 합니다. 철자 오류를 허용하는 검색을 위해 pg_trgm extension을 추가로 사용할 수 있습니다.

6. pgvector를 사용한 Vector search

6.1. 스키마

Embedding은 AI 모델에서 생성한 실수 배열입니다. 의미적으로 가까운 텍스트는 비슷한 벡터를 생성합니다. Extension을 활성화하고 컬럼을 추가합니다:

CREATE EXTENSION IF NOT EXISTS vector;

ALTER TABLE product ADD COLUMN IF NOT EXISTS embedding vector(3);

CREATE INDEX IF NOT EXISTS idx_product_embedding
    ON product USING hnsw (embedding vector_cosine_ops);

가독성을 위해 3차원을 사용하는 예입니다. 실제로 차원 수는 사용 중인 embedding 모델과 정확히 일치해야 하며, 보통 수백에서 수천 차원입니다. HNSW 인덱스는 매우 빠른 approximate nearest neighbor 검색을 허용합니다. vector_cosine_ops는 인덱스가 cosine 거리를 제공함을 나타냅니다.

6.2. Vector 저장 및 검색

추가 라이브러리가 필요하지 않습니다. float[]을 [0.1,0.2,0.3] 형식의 문자열로 변환한 다음 ::vector로 캐스팅합니다:

@Repository
public class ProductVectorSearch {

    private final JdbcTemplate jdbc;

    public ProductVectorSearch(JdbcTemplate jdbc) {
        this.jdbc = jdbc;
    }

    public void saveEmbedding(long productId, float[] embedding) {
        jdbc.update("UPDATE product SET embedding = ?::vector WHERE id = ?",
                toLiteral(embedding), productId);
    }

    public List<String> findSimilar(float[] queryEmbedding, int limit) {
        return jdbc.queryForList("""
                SELECT name FROM product
                WHERE embedding IS NOT NULL
                ORDER BY embedding <=> ?::vector
                LIMIT ?
                """, String.class, toLiteral(queryEmbedding), limit);
    }

    private static String toLiteral(float[] v) {
        StringJoiner joiner = new StringJoiner(",", "[", "]");
        for (float f : v) {
            joiner.add(Float.toString(f));
        }
        return joiner.toString();
    }
}

<=> 연산자는 cosine 거리를 계산합니다. ORDER BY ... LIMIT은 가장 가까운 제품들을 가져옵니다. 쿼리 벡터는 제품에 대한 embedding을 생성한 것과 동일한 모델에서 가져옵니다, 예를 들어 사용자가 “비가 올 때 입을 옷”을 입력할 때.

제품과 embedding이 같은 테이블에 있기 때문에 필터링을 쉽게 결합할 수 있습니다: WHERE category = 'outerwear' AND price < 500000 ORDER BY embedding <=> ?. 또한 ts_rank를 벡터 거리와 혼합하여 hybrid search를 수행할 수 있습니다. 별도의 vector database를 사용하면 이러한 속성을 동기화해야 합니다.

참고: 높은 선택도의 WHERE을 HNSW 인덱스와 결합할 때, 반환된 결과 수가 LIMIT보다 적을 수 있습니다. 이유는 인덱스가 제한된 후보 세트를 탐색한 후 필터링하기 때문입니다. 실제 데이터로 신중히 테스트하고 pgvector의 튜닝 파라미터를 참조하세요.

7. Testcontainers로 테스트 작성

테스트는 실제 PostgreSQL에서 실행되며 pgvector가 포함된 이미지를 사용합니다. @ServiceConnection은 datasource를 자동으로 구성합니다:

@SpringBootTest
@Testcontainers
class PostgresFeaturesTest {

    @Container
    @ServiceConnection
    static PostgreSQLContainer<?> postgres = new PostgreSQLContainer<>(
            DockerImageName.parse("pgvector/pgvector:pg16").asCompatibleSubstituteFor("postgres"));

    @Autowired JobQueue jobQueue;
    @Autowired ProductSearch productSearch;
    @Autowired ProductVectorSearch vectorSearch;
    @Autowired JdbcTemplate jdbc;

    @Test
    void processesEachJobExactlyOnce() {
        jobQueue.enqueue("SEND_EMAIL", "{\"orderId\": 1}");
        jobQueue.enqueue("SEND_EMAIL", "{\"orderId\": 2}");
        List<Long> processed = new ArrayList<>();

        while (jobQueue.processNext(job -> processed.add(job.id()))) { }

        assertThat(processed).hasSize(2).doesNotHaveDuplicates();
        assertThat(jdbc.queryForObject("SELECT count(*) FROM job_queue", Long.class)).isZero();
    }

    @Test
    void failedJobStaysInQueue() {
        jobQueue.enqueue("SEND_EMAIL", "{\"orderId\": 3}");

        assertThatThrownBy(() -> jobQueue.processNext(job -> {
            throw new IllegalStateException("SMTP down");
        })).isInstanceOf(IllegalStateException.class);

        assertThat(jdbc.queryForObject("SELECT count(*) FROM job_queue", Long.class)).isEqualTo(1);
    }

    @Test
    void searchesByTextAndVector() {
        jdbc.update("INSERT INTO product (name, description) VALUES ('Áo khoác gió', 'Chống nước nhẹ')");
        jdbc.update("INSERT INTO product (name, description) VALUES ('Quần jean', 'Vải dày')");
        long jacketId = jdbc.queryForObject("SELECT id FROM product WHERE name = 'Áo khoác gió'", Long.class);
        long jeansId = jdbc.queryForObject("SELECT id FROM product WHERE name = 'Quần jean'", Long.class);
        vectorSearch.saveEmbedding(jacketId, new float[]{0.9f, 0.1f, 0.0f});
        vectorSearch.saveEmbedding(jeansId, new float[]{0.0f, 0.2f, 0.9f});

        assertThat(productSearch.search("khoác"))
                .extracting(ProductHit::name)
                .containsExactly("Áo khoác gió");
        assertThat(vectorSearch.findSimilar(new float[]{1.0f, 0.0f, 0.0f}, 1))
                .containsExactly("Áo khoác gió");
    }
}

두 번째 테스트는 queue의 가장 중요한 동작을 확인합니다: handler가 실패할 때 job이 손실되지 않습니다. 테스트는 각 메서드에 데이터를 직접 설정합니다. 실제 프로젝트에서는 테스트가 서로 독립적이 되도록 @BeforeEach에서 테이블을 정리해야 합니다.

8. 언제 Postgres를 모든 것에 사용하면 안 되는가

모든 것을 한 곳에 넣는 것도 비용이 있습니다. 다음과 같은 경우 전문 시스템을 고려해야 합니다:

• queue throughput이 매우 높거나 streaming이 필요합니다. 이벤트 스트림을 재생해야 하고, 독립적인 많은 consumer group이 있거나, 메시지를 오래 보관해야 한다면 Kafka나 유사한 도구를 사용하세요. Postgres의 queue는 지속적인 DELETE/UPDATE로 인해 많은 dead tuple을 생성합니다. autovacuum을 세심하게 모니터링해야 합니다.
• 복잡한 검색. 다차원 faceting, 입력 중 제안, 베트남어를 위한 좋은 언어 처리, 자세한 relevance 튜닝: 전문 search engine이 훨씬 잘 합니다.
• 매우 큰 양의 vector. HNSW 인덱스는 많은 RAM이 필요합니다. 벡터 데이터가 대부분의 리소스를 차지할 때, 이는 메인 트랜잭션 workload와 경합합니다.
• 불균형한 부하가 서로에게 영향을 미칩니다. 무거운 벡터 쿼리나 job의 대량 도입이 결제 API를 느리게 할 수 있습니다. Read replica는 일부 읽기 부하를 분리하는 데 도움이 되지만 완전히 해결하지는 못합니다.
• 팀이 이미 전문 시스템을 잘 운영하고 있습니다. 인프라가 이미 존재하고 안정적이면, 단순히 “정리”하기 위해 모든 것을 Postgres로 옮기는 것이 가치가 없을 수 있습니다.

참고: 합리적인 전략은 각 기능을 인터페이스(JobQueue, ProductSearch…) 뒤에 래핑하는 것입니다. 모니터링 데이터가 Postgres가 병목임을 보여주면, 비즈니스 코드를 수정하지 않고 기본 구현을 변경할 수 있습니다.

9. 결론

PostgreSQL은 FOR UPDATE SKIP LOCKED, tsvector와 GIN index를 사용한 full-text search, 그리고 pgvector extension을 통한 vector search로 job queue를 처리할 수 있는 충분한 능력이 있습니다. 가장 큰 이점은 운영해야 할 시스템이 적고, 비즈니스 데이터, job, embedding을 같은 트랜잭션에 저장할 수 있다는 것입니다.

이를 작은 중간 규모 시스템의 기본 시작점으로 생각하세요, 최종 목표가 아닙니다. Postgres로 시작하고, 신중하게 측정하고, 데이터가 필요함을 보여줄 때만 전문 도구로 분리하세요.