PostgreSQL Full-Text Search and Elasticsearch

Search is one of the important features users interact with a system. A user types “Taylor Swift” and expects results in under a second. At small scale this is trivial. At millions of events, it becomes one of the harder problems to get right.

The naive implementation reaches for a LIKE query:

SELECT * FROM events
 WHERE name        LIKE '%Taylor%'
    OR description LIKE '%Taylor%';

This works correctly and is easy to write. The problem is performance. The wildcard on both sides of the search term — %Taylor% — means no standard index can help. The database has no choice but to read every single row in the table, check each one against the pattern, and discard the ones that do not match. This is called a full table scan, and its cost grows linearly with the number of rows. At 10,000 events it is imperceptible. At 10 million events it can take several seconds, and under concurrent load it can saturate the database entirely.

Search queries also have requirements beyond just finding a match. Users expect results ordered by relevance — a search for “Taylor Swift” should return Taylor Swift headline events above a community fan meetup that happens to mention her name. Users also make typos — “Tayler Swft” should still find the right results. Neither of these is possible with LIKE alone.

This post covers two progressively more powerful solutions: PostgreSQL’s built-in full-text search, and Elasticsearch backed by Change Data Capture (Demzium).

Solution 1: PostgreSQL Full-Text Search Link to heading

PostgreSQL ships with a full-text search engine built in. It is not as feature-rich as Elasticsearch, but it requires zero additional infrastructure and handles the majority of search use cases well.

How it works Link to heading

PostgreSQL converts text into a tsvector — a sorted list of lexemes (normalised word stems) with positional information. A query is expressed as a tsquery. The @@ operator checks whether a document matches a query, and a GIN index makes that check fast.

-- a tsvector for one event row
SELECT to_tsvector('english', 'Taylor Swift Eras Tour');
-- → 'era':3 'swift':2 'taylor':1 'tour':4

Notice that “Taylor”, “Swift”, “Eras”, and “Tour” are all stemmed and lowercased. A search for “tours” would still match “tour”. A search for “TAYLOR” would still match “taylor”. This is something LIKE cannot do.

Field weights and relevance ranking Link to heading

PostgreSQL supports 4 weight tiers — A, B, C, D — which let you express that a match in the event name matters more than a match in the description. The trigger in the schema below assigns:

A — name (most important)
B — performer
C — description
D — venue

ts_rank uses these weights to score each result, so a keyword appearing in the name ranks higher than the same keyword buried in the description.

Schema Link to heading

ALTER TABLE events ADD COLUMN search_vector tsvector;

CREATE OR REPLACE FUNCTION events_search_vector_update() RETURNS trigger AS $$
BEGIN
    NEW.search_vector :=
        setweight(to_tsvector('english', coalesce(NEW.name, '')),        'A') ||
        setweight(to_tsvector('english', coalesce(NEW.performer, '')),    'B') ||
        setweight(to_tsvector('english', coalesce(NEW.description, '')), 'C') ||
        setweight(to_tsvector('english', coalesce(NEW.venue, '')),        'D');
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER events_search_vector_trigger
    BEFORE INSERT OR UPDATE ON events
    FOR EACH ROW EXECUTE FUNCTION events_search_vector_update();

CREATE INDEX idx_events_fts ON events USING GIN(search_vector);

The trigger runs automatically on every insert and update, so search_vector is always current. The GIN index is what makes queries fast — instead of scanning every row, Postgres looks up the lexeme in the index and retrieves only the matching row IDs.

Query Link to heading

SELECT id, name,
       ts_rank(search_vector, to_tsquery('english', 'Taylor')) AS rank
  FROM events
 WHERE search_vector @@ to_tsquery('english', 'Taylor')
 ORDER BY rank DESC;

Tradeoffs Link to heading

The GIN index adds storage overhead and slightly slows down writes, since every insert or update triggers a reindex of that row. For a table with frequent bulk inserts, you can defer index maintenance with GIN_PENDING_LIST_LIMIT. Full-text search also has no concept of edit distance — a typo like “Tayler” will not match “Taylor”.

Solution 2: Elasticsearch with Change Data Capture Link to heading

Elasticsearch is the right tool when you need typo tolerance, multi-language stemming, or more control over relevance tuning than PostgreSQL offers. Its core data structure is an inverted index: for each unique term, it stores a list of documents containing that term. A query for “Taylor” is a direct lookup in that list — not a scan.

Fuzzy search Link to heading

Elasticsearch supports fuzziness: AUTO, which allows a configurable edit distance per token:

Tokens up to 2 characters: no fuzzy matching
Tokens 3–5 characters: 1 allowed edit
Tokens 6+ characters: 2 allowed edits

“Tayler” (1 edit from “Taylor”) and “Swft” (1 edit from “Swift”) both fall within tolerance, so a search for “Tayler Swft” still returns Taylor Swift events. This is effectively impossible to replicate with SQL.

Keeping Elasticsearch in sync: CDC Link to heading

The challenge with Elasticsearch as a secondary store is keeping it consistent with Postgres. There are two ways to do this, each with a different complexity and freshness tradeoff.

Option A: Polling on `updated_at` Link to heading

The simplest approach is to add an updated_at timestamp column to the events table and periodically query for rows that changed since the last sync.

The last sync timestamp must be persisted to Postgres, for example in a dedicated sync_state table:

CREATE TABLE sync_state (
    key        VARCHAR(50) PRIMARY KEY,
    synced_at  TIMESTAMP WITH TIME ZONE NOT NULL
);

INSERT INTO sync_state (key, synced_at) VALUES ('es_sync', NOW());

The sync function reads the checkpoint at the start of every run and saves it at the end:

def sync_to_es(self) -> int:
    last_sync = self._get_last_sync()

    with self._db() as conn:
        with conn.cursor(...) as cur:
            if last_sync:
                cur.execute(
                    "SELECT * FROM events WHERE updated_at > %s", (last_sync,)
                )
            else:
                cur.execute("SELECT * FROM events")  # full initial load
            rows = [dict(r) for r in cur.fetchall()]

    for row in rows:
        self._es.index(index=ES_INDEX, id=row["id"], document={...})

    self._save_last_sync(datetime.now(tz=timezone.utc))
    return len(rows)

This function runs on a schedule — every few seconds in a background thread or a cron job. The replication lag is equal to the poll interval.

The main limitation is that deletes are invisible. When a row is deleted from Postgres it is gone — the polling query will never see it. To handle deletes, use soft deletes: instead of DELETE, set a deleted_at column, let the poller sync it to Elasticsearch, and filter it out at query time.

Option B: Debezium + Kafka Link to heading

For production systems, the standard approach is Debezium reading the PostgreSQL Write-Ahead Log (WAL). Every insert, update, and delete is captured as an event in the order it occurred, with sub-second latency.

Step 1: Enable logical replication in Postgres.

Debezium connects as a replication client, so Postgres must be configured to expose the WAL:

-- postgresql.conf
wal_level = logical

-- Grant replication permission to the Debezium user
CREATE ROLE debezium WITH REPLICATION LOGIN PASSWORD 'password';
GRANT SELECT ON ALL TABLES IN SCHEMA public TO debezium;

Step 2: Deploy Debezium.

Debezium runs as a Kafka Connect connector. Add it to your docker-compose.yml alongside Kafka and Zookeeper:

zookeeper:
  image: confluentinc/cp-zookeeper:7.6.0
  environment:
    ZOOKEEPER_CLIENT_PORT: 2181

kafka:
  image: confluentinc/cp-kafka:7.6.0
  environment:
    KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
    KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092

kafka-connect:
  image: debezium/connect:2.6
  environment:
    BOOTSTRAP_SERVERS: kafka:9092
    GROUP_ID: 1
    CONFIG_STORAGE_TOPIC: connect_configs
    OFFSET_STORAGE_TOPIC: connect_offsets
  ports:
    - "8083:8083"

Step 3: Register the Postgres connector.

Once Kafka Connect is running, register the Debezium Postgres connector via its REST API:

curl -X POST http://localhost:8083/connectors \
  -H "Content-Type: application/json" \
  -d '{
    "name": "events-connector",
    "config": {
      "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
      "database.hostname": "postgres",
      "database.port": "5432",
      "database.user": "debezium",
      "database.password": "password",
      "database.dbname": "search",
      "table.include.list": "public.events",
      "topic.prefix": "ticketing"
    }
  }'

Debezium now publishes every change to the Kafka topic ticketing.public.events. Each message contains the full before and after state of the row, plus the operation type (c = create, u = update, d = delete).

Step 4: Consume from Kafka and write to Elasticsearch.

A consumer reads from the topic and upserts or deletes documents in Elasticsearch:

from kafka import KafkaConsumer
import json

consumer = KafkaConsumer(
    "ticketing.public.events",
    bootstrap_servers="kafka:9092",
    value_deserializer=lambda m: json.loads(m.decode()),
)

for message in consumer:
    payload = message.value["payload"]
    op      = payload["op"]          # 'c', 'u', or 'd'
    after   = payload.get("after")   # row state after the change
    row_id  = payload["before"]["id"] if op == "d" else after["id"]

    if op in ("c", "u"):
        es.index(index="events", id=row_id, document={
            "name":        after["name"],
            "description": after["description"],
            "performer":   after["performer"],
            "venue":       after["venue"],
        })
    elif op == "d":
        es.delete(index="events", id=row_id)

Kafka acts as a durable buffer between Debezium and your consumer. If the Elasticsearch cluster goes down, events accumulate in Kafka and are replayed once it recovers. The consumer tracks its position in the Kafka topic (the offset), so it always knows where to resume after a restart.

The key point in both approaches is that Postgres remains the source of truth. Elasticsearch is a read replica optimised for search. If it goes down, no data is lost. It can be fully rebuilt by replaying the Kafka topic or running a full polling sync from Postgres.

Tradeoffs Link to heading

Running Elasticsearch adds operational complexity: cluster management, index mapping versioning, and ensuring the sync pipeline never falls behind. There is also an inherent replication lag: a newly inserted event is not searchable in Elasticsearch until the CDC pipeline delivers it, which can be anywhere from milliseconds (Debezium) to seconds (polling).

Project Setup Link to heading

search/
├── docker-compose.yml
├── requirements.txt
├── setup.sql
├── search_service.py

# docker-compose.yml
services:
  postgres:
    image: postgres:15-alpine
    environment:
      POSTGRES_DB: search
      POSTGRES_USER: search
      POSTGRES_PASSWORD: search
    # wal_level=logical is required for Debezium to read the WAL
    command: postgres -c wal_level=logical
    ports:
      - "5433:5432"
    volumes:
      - ./setup.sql:/docker-entrypoint-initdb.d/setup.sql
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U search"]
      interval: 5s
      timeout: 5s
      retries: 5

  elasticsearch:
    image: elasticsearch:8.13.0
    environment:
      - discovery.type=single-node
      - xpack.security.enabled=false
      - ES_JAVA_OPTS=-Xms512m -Xmx512m
    ports:
      - "9200:9200"
    healthcheck:
      test: ["CMD-SHELL", "curl -sf http://localhost:9200/_cluster/health | grep -q 'green\\|yellow'"]
      interval: 10s
      timeout: 10s
      retries: 10

  zookeeper:
    image: confluentinc/cp-zookeeper:7.6.0
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
    ports:
      - "2181:2181"

  kafka:
    image: confluentinc/cp-kafka:7.6.0
    environment:
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
    ports:
      - "9092:9092"
    depends_on:
      - zookeeper

  # Debezium Kafka Connect — hosts the Postgres source connector
  debezium:
    image: debezium/connect:2.6
    environment:
      BOOTSTRAP_SERVERS: kafka:9092
      GROUP_ID: 1
      CONFIG_STORAGE_TOPIC: connect_configs
      OFFSET_STORAGE_TOPIC: connect_offsets
      STATUS_STORAGE_TOPIC: connect_status
    ports:
      - "8083:8083"
    depends_on:
      - kafka
      - postgres

# requirements.txt
psycopg2-binary==2.9.10
elasticsearch==8.13.0

-- setup.sql
CREATE TABLE events (
    id            SERIAL PRIMARY KEY,
    name          VARCHAR(255) NOT NULL,
    description   TEXT,
    performer     VARCHAR(255),
    venue         VARCHAR(255),
    date          TIMESTAMP,
    updated_at    TIMESTAMP NOT NULL DEFAULT NOW(),
    search_vector tsvector
);

-- Keeps search_vector current on every insert and update
CREATE OR REPLACE FUNCTION events_search_vector_update() RETURNS trigger AS $$
BEGIN
    NEW.search_vector :=
        setweight(to_tsvector('english', coalesce(NEW.name, '')),        'A') ||
        setweight(to_tsvector('english', coalesce(NEW.performer, '')),    'B') ||
        setweight(to_tsvector('english', coalesce(NEW.description, '')), 'C') ||
        setweight(to_tsvector('english', coalesce(NEW.venue, '')),        'D');
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER events_search_vector_trigger
    BEFORE INSERT OR UPDATE ON events
    FOR EACH ROW EXECUTE FUNCTION events_search_vector_update();

CREATE INDEX idx_events_fts        ON events USING GIN(search_vector);
CREATE INDEX idx_events_updated_at ON events(updated_at);

-- Persists the CDC polling checkpoint across service restarts
CREATE TABLE sync_state (
    key        VARCHAR(50) PRIMARY KEY,
    synced_at  TIMESTAMP WITH TIME ZONE NOT NULL
);

INSERT INTO sync_state (key, synced_at) VALUES ('es_sync', NOW());

INSERT INTO events (name, description, performer, venue, date) VALUES
    ('Taylor Swift Eras Tour', 'The legendary Eras Tour featuring hits from every album era', 'Taylor Swift', 'SoFi Stadium', '2024-08-10 19:00:00'),
    ('Taylor Swift at Madison Square', 'An intimate show at MSG spanning all eras', 'Taylor Swift', 'Madison Square Garden', '2024-09-15 20:00:00'),
    ('Taylor Nation Fan Meetup', 'Connect with fellow Swifties at this fan-organised event', 'Community Event', 'Nashville Convention Center', '2024-07-04 10:00:00'),
    ('Ed Sheeran Mathematics Tour', 'Ed Sheeran performs greatest hits and new material', 'Ed Sheeran', 'Wembley Stadium', '2024-07-20 18:30:00'),
    ('Beyonce Renaissance World Tour', 'Beyonce brings the Renaissance album to life on stage', 'Beyonce', 'MetLife Stadium', '2024-10-05 19:00:00'),
    ('Coldplay Music of the Spheres', 'An eco-friendly spectacular world tour', 'Coldplay', 'Rose Bowl', '2024-11-12 19:30:00'),
    ('Drake It''s All A Blur Tour', 'Drake and 21 Savage on tour together', 'Drake', 'Scotiabank Arena', '2024-08-25 20:00:00'),
    ('Classical Night: Beethoven', 'An evening of classical masterpieces performed live', 'City Philharmonic Orchestra', 'Carnegie Hall', '2024-09-01 19:00:00'),
    ('Jazz Under the Stars', 'Outdoor jazz concert featuring local and national artists', 'Various Artists', 'Central Park', '2024-08-03 20:00:00'),
    ('Rock Festival 2024', 'A three-day rock festival featuring top bands from around the world', 'Various Artists', 'Download Festival Grounds', '2024-06-14 12:00:00');

# search_service.py
from datetime import datetime, timezone
from typing import Optional

import psycopg2
import psycopg2.extras
from elasticsearch import Elasticsearch

ES_INDEX = "events"


class SearchService:
    def __init__(
        self,
        pg_dsn: str = "postgresql://search:search@localhost:5433/search",
        es_url: str = "http://localhost:9200",
    ):
        self._pg_dsn = pg_dsn
        self._es = Elasticsearch(es_url)
        self._ensure_es_index()

    def _db(self):
        return psycopg2.connect(self._pg_dsn)

    def _ensure_es_index(self) -> None:
        if not self._es.indices.exists(index=ES_INDEX):
            self._es.indices.create(
                index=ES_INDEX,
                mappings={
                    "properties": {
                        "name":        {"type": "text", "analyzer": "english"},
                        "description": {"type": "text", "analyzer": "english"},
                        "performer":   {"type": "text", "analyzer": "english"},
                        "venue":       {"type": "text", "analyzer": "english"},
                        "date":        {"type": "date"},
                    }
                },
            )

    def like_search(self, keyword: str) -> list[dict]:
        with self._db() as conn:
            with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
                cur.execute(
                    """
                    SELECT id, name, performer, venue
                      FROM events
                     WHERE name ILIKE %s OR description ILIKE %s OR performer ILIKE %s
                    """,
                    (f"%{keyword}%",) * 3,
                )
                return [dict(r) for r in cur.fetchall()]

    def fts_search(self, query: str) -> list[dict]:
        with self._db() as conn:
            with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
                cur.execute(
                    """
                    SELECT id, name, performer, venue,
                           ts_rank(search_vector, to_tsquery('english', %s)) AS rank
                      FROM events
                     WHERE search_vector @@ to_tsquery('english', %s)
                     ORDER BY rank DESC
                    """,
                    (query, query),
                )
                return [dict(r) for r in cur.fetchall()]

    def es_search(self, query: str, fuzzy: bool = False) -> list[dict]:
        resp = self._es.search(
            index=ES_INDEX,
            query={
                "multi_match": {
                    "query":  query,
                    "fields": ["name^3", "performer^2", "description", "venue"],
                    **({"fuzziness": "AUTO"} if fuzzy else {}),
                }
            },
        )
        return [
            {**hit["_source"], "score": round(hit["_score"], 4)}
            for hit in resp["hits"]["hits"]
        ]

    def _get_last_sync(self) -> Optional[datetime]:
        with self._db() as conn:
            with conn.cursor() as cur:
                cur.execute(
                    "SELECT synced_at FROM sync_state WHERE key = 'es_sync'"
                )
                row = cur.fetchone()
                return row[0] if row else None

    def _save_last_sync(self, ts: datetime) -> None:
        with self._db() as conn:
            with conn.cursor() as cur:
                cur.execute(
                    """
                    INSERT INTO sync_state (key, synced_at) VALUES ('es_sync', %s)
                    ON CONFLICT (key) DO UPDATE SET synced_at = EXCLUDED.synced_at
                    """,
                    (ts,),
                )
                conn.commit()

    def sync_to_es(self) -> int:
        last_sync = self._get_last_sync()

        with self._db() as conn:
            with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
                if last_sync:
                    cur.execute(
                        "SELECT * FROM events WHERE updated_at > %s", (last_sync,)
                    )
                else:
                    cur.execute("SELECT * FROM events")
                rows = [dict(r) for r in cur.fetchall()]

        for row in rows:
            self._es.index(
                index=ES_INDEX,
                id=row["id"],
                document={
                    "name":        row["name"],
                    "description": row["description"],
                    "performer":   row["performer"],
                    "venue":       row["venue"],
                    "date":        row["date"].isoformat() if row["date"] else None,
                },
            )

        self._save_last_sync(datetime.now(tz=timezone.utc))
        return len(rows)

Solution 1: PostgreSQL Full-Text Search Link to heading

How it works Link to heading

Field weights and relevance ranking Link to heading

Schema Link to heading

Query Link to heading

Tradeoffs Link to heading

Solution 2: Elasticsearch with Change Data Capture Link to heading

Fuzzy search Link to heading

Keeping Elasticsearch in sync: CDC Link to heading

Option A: Polling on updated_at Link to heading

Option B: Debezium + Kafka Link to heading

Tradeoffs Link to heading

Project Setup Link to heading

Option A: Polling on `updated_at` Link to heading