PostgreSQLですべて: queue、search、vectorをひとつのデータベースで

1. 概要

典型的なシステムには多くのコンポーネントがあります：データを保存するPostgreSQL、キューを担当するメッセージブローカー、検索エンジン、そして最近ではAI機能用のベクトルデータベースが追加されています。各コンポーネントには、運用、監視、バックアップ、データ同期が伴います。

“PostgreSQLですべて”トレンドは逆の質問を投げかけます：PostgreSQLがそのほとんどをすでに十分にこなせるなら、新しいシステムを追加する必要があるのか？

この記事では、PostgreSQLをSpring Boot 3.xでjob queue、full-text search、vector searchとして使用します。また、いつやめて専用ツールを選ぶべきかも学びます。

2. このトレンドが注目される理由

主な理由があります：

• 運用の複雑性を減らす。 追加するシステムごとに、夜中に壊れるリスクが増えます。小規模なチームでは、このコストはパフォーマンスのメリットより大きいことがしばしばあります。
• トランザクション単位での一貫性。 注文とメール送信ジョブが同じデータベースにあると、両方をひとつのトランザクションで記録できます。注文は保存されたがメッセージが失われる、またはその逆、という状況が発生しません。
• 拡張エコシステム。 PostgreSQLは強力なextensionメカニズムを備えています。例えばpgvectorはベクトルデータ型と最近傍検索インデックスを追加します。
• AI機能に伴う埋め込みの保存需要。 多くのチームが、そのためだけに新しいデータベースを構築することなく、セマンティック検索機能を追加したいと考えています。

このトレンドの本質は、シンプルに始めて、必要性が実証されてから分離することです。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拡張機能が事前にパッケージ化されています。使用するPostgreSQLのメジャーバージョンに対応するタグを選択します：

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

Spring Bootを設定して、接続し、起動時にschema.sqlを自動実行するようにします：

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を使用したジョブキュー

4.1. スキーマ

最小限のキューは1つのテーブルで十分です：

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列でジョブの実行をスケジュールできます。run_at上のインデックスにより、ワーカーはテーブル全体をスキャンすることなく、最も早いジョブを取得できます。

4.2. ジョブの取得と処理

重要な点はFOR UPDATE SKIP LOCKED句です。複数のワーカーが同時にクエリを実行します。各ワーカーは1行をロックし、他のワーカーはロックされた行をスキップします（待機する代わりに）：

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はジョブを挿入します。?::jsonbは JSON文字列をJSONB型にキャストします。
• processNextはトランザクション内で実行されます。SELECT ... FOR UPDATE SKIP LOCKEDは、まだ誰にも取得されていないジョブを正確に1つロックします。
• handlerが完了すると、ジョブが削除され、トランザクションがコミットされます。
• handlerが例外をスローするとトランザクションはロールバックされます。ジョブはテーブルに残り、再処理されます。ワーカーが途中で死ぬと、PostgreSQLは接続を閉じるときに自動的にロックを解放します。

ワーカーはシンプルな@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.id() + " にメールを送信中: " + job.payload()))) {
            // ジョブがなくなるまで続行
        }
    }
}

@EnableSchedulingを設定クラスに追加することを忘れないでください。ジョブがあるとき、コンソールに以下が出力されます：

ジョブ 1 にメールを送信中: {"orderId": 42, "email": "an@example.com"}

注意: ジョブ処理中にトランザクションを保持する方法は非常にシンプルで、ワーカーがクラッシュしても安全です。一方、トランザクションは処理時間の長さによって延長されます。ジョブが長時間実行される場合（遅い API 呼び出し、大きなファイル処理）、statusとlocked_until列を持つモデルに移行するか、専用のキューライブラリに切り替えるべきです。また、エラーが無限にリトライされるのを防ぐために、attempts列を追加すべきです。

4.3. 真の利点：ビジネストランザクションと一緒にエンキュー

これがPostgresをキューとして使用する最強の理由です：

@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));
    }
}

注文とジョブは一緒にコミットまたはロールバックされます。別のブローカーを使用する場合、この保証を達成するためにtransactional outboxパターンを手動で構築する必要があります。（例では簡潔性のためにformattedを使用しています。実際のコードではペイロードをJacksonでシリアライズして、文字を正しくエスケープすべきです。）

ワーカーが継続的にポーリングしたくない場合、PostgreSQLのLISTEN/NOTIFYを組み合わせて、新しいジョブがあるときにワーカーを起動できます。ポーリングはフォールバック機構として保持すべきです。

5. tsvectorとGINインデックスを使用したフルテキスト検索

5.1. スキーマ

PostgreSQLにはfull-text searchが組み込まれています。generated列を使用して、データが変更されるたびに検索ベクトルを自動的に更新します：

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設定は小文字への変換のみを行い、ステミングは行いません。このオプションはベトナム語に適しています。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拡張機能を使用できます。unaccent関数はデフォルトでIMMUTABLEとマークされていないため、generated列またはインデックスで使用するには、独自のIMMUTABLE関数でラップする必要があります。スペルミスに対応した検索には、pg_trgm拡張機能を追加で使用できます。

6. pgvectorを使用したベクトル検索

6.1. スキーマ

埋め込みはAIモデルによって生成される実数の配列です。意味が近いテキストは、ベクトルも近くなります。拡張機能を有効にして列を追加します：

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次元を使用しています。実際には、次元数は選択した埋め込みモデルに正確に一致する必要があり、通常は数百から数千次元です。HNSWインデックスは非常に高速な近似最近傍検索を可能にします。vector_cosine_opsはインデックスがコサイン距離を提供することを示します。

6.2. ベクトルの保存と検索

追加ライブラリは不要です。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();
    }
}

演算子<=>はコサイン距離を計算します。ORDER BY ... LIMITは最も類似した製品を取得します。クエリベクトルは、製品の埋め込みを生成したのと同じモデルから取得されます。例えば、ユーザーが「đồ mặc khi trời mưa」（雨の日の服装）と入力した場合です。

製品と埋め込みが同じテーブルにあるため、フィルタリングが簡単に結合できます：WHERE category = 'outerwear' AND price < 500000 ORDER BY embedding <=> ?。また、ts_rankスコアとベクトル距離をブレンドしてハイブリッド検索を実装することもできます。別のベクトルデータベースを使用する場合、これらの属性をそこに同期する必要があります。

注意: 高い選択性を持つ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ダウン");
        })).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ó");
    }
}

2番目のテストはキューの最も重要な動作を検証します：ハンドラーがエラーを発生させても、ジョブが失われないこと。テストは各メソッド内でデータを直接設定しています。実際のプロジェクトでは、テストが相互に独立するよう、@BeforeEachでテーブルをクリーンアップすべきです。

8. PostgreSQLを何もかも使用するべきでない場合

すべてを1つの場所に集約することにもコストがあります。以下の場合は専用システムの検討が必要です：

• キュースループットが非常に高いか、ストリーミングが必要な場合。 イベントストリームのリプレイ、複数の独立したコンシューマーグループ、長期的なメッセージ保持が必要な場合は、KafkaまたはそれらのツールをKafkaやそれに類するツール使用してください。Postgres上のキューは継続的なDELETE/UPDATEによる多くのデッドタプルを生成します。vacuumを注意深く監視する必要があります。
• 複雑な検索。 多次元ファセティング、入力時のサジェスト、ベトナム語の処理、詳細なランキングの微調整：専用の検索エンジンの方が優れています。
• ベクトルの量が非常に多い場合。 HNSWインデックスは多くのRAMを必要とします。ベクトルデータが大部分のリソースを占める場合、メインのトランザクショナルワークロードと競合します。
• 負荷の不均等が相互に影響する場合。 重いベクトルクエリまたはジョブのバースト実行は、決済APIを遅くさせることができます。読み取りレプリカは読み取り負荷の一部を分離するのに役立ちますが、完全には解決できません。
• チームが専用システムを既に適切に運用している場合。 インフラが既に存在し、安定している場合、単に「スッキリ」させるためにすべてをPostgresに移す価値があるかは不確実です。

注意: 合理的な戦略は、各機能をインタフェース（JobQueue、ProductSearch…）の背後でラップすることです。監視データがPostgresがボトルネックであることを示した場合、ビジネスロジックのコードを変更することなく、基礎となる実装を交換できます。

9. 結論

PostgreSQLはFOR UPDATE SKIP LOCKEDによるジョブキュー、tsvectorとGINインデックスによるフルテキスト検索、pgvector拡張によるベクトル検索を処理できます。最大のメリットは、運用すべきシステムが少なく、ビジネスデータ、ジョブ、埋め込みを同じトランザクション内に記録できることです。

これを小規模および中規模システムのデフォルトの出発点と見なしてください。最終的な目標ではなく。Postgresから始め、注意深く測定して、データが必要な場合にのみ専用ツールに移行してください。