Реализация Full-Text Search для веб-приложения

Реализация Full-Text Search для веб-приложения

Full-text search — поиск по смыслу слов, а не точному совпадению. LIKE '%запрос%' не масштабируется и не понимает морфологию: "купить", "купил", "купленный" — разные строки. FTS обрабатывает все три.

PostgreSQL FTS: встроенный вариант

Для большинства проектов встроенный FTS PostgreSQL закрывает задачу без внешних сервисов.

Подготовка схемы:

ALTER TABLE products
    ADD COLUMN search_vector TSVECTOR
                             GENERATED ALWAYS AS (
        to_tsvector('russian',
            coalesce(title, '') || ' ' ||
            coalesce(description, '') || ' ' ||
            coalesce(brand, '')
        )
    ) STORED;

CREATE INDEX idx_products_fts ON products USING GIN (search_vector);

GENERATED ALWAYS AS ... STORED — PostgreSQL 12+. Колонка обновляется автоматически при INSERT/UPDATE, не нужен триггер.

Для мультиязычного поиска и разных весов полей:

-- Без GENERATED (гибкая настройка):
UPDATE products SET search_vector =
    setweight(to_tsvector('russian', coalesce(title, '')), 'A') ||
    setweight(to_tsvector('russian', coalesce(brand, '')), 'B') ||
    setweight(to_tsvector('russian', coalesce(description, '')), 'C');

CREATE OR REPLACE FUNCTION products_search_update() RETURNS TRIGGER AS $$
BEGIN
    NEW.search_vector :=
        setweight(to_tsvector('russian', coalesce(NEW.title, '')), 'A') ||
        setweight(to_tsvector('russian', coalesce(NEW.brand, '')), 'B') ||
        setweight(to_tsvector('russian', coalesce(NEW.description, '')), 'C');
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER products_search_trigger
    BEFORE INSERT OR UPDATE ON products
    FOR EACH ROW EXECUTE FUNCTION products_search_update();

Поиск:

-- Простой запрос
SELECT id, title,
       ts_rank(search_vector, query)   AS rank,
       ts_headline('russian', description, query,
           'MaxWords=30, MinWords=15, StartSel=<b>, StopSel=</b>'
       ) AS excerpt
FROM products,
     plainto_tsquery('russian', 'беспроводные наушники') AS query
WHERE search_vector @@ query
ORDER BY rank DESC
LIMIT 20;

-- websearch_to_tsquery: поддерживает "фразы", -исключения, OR
SELECT id, title
FROM products
WHERE search_vector @@ websearch_to_tsquery('russian', '"беспроводные наушники" -проводные')
ORDER BY ts_rank(search_vector, websearch_to_tsquery('russian', '"беспроводные наушники" -проводные')) DESC;

ts_headline генерирует сниппет с подсвеченными совпадениями.

Elasticsearch: когда нужен внешний движок

PostgreSQL FTS ограничен: нет fuzzy search, нет синонимов из коробки, нет агрегаций по фасетам. Если нужно все три — Elasticsearch или OpenSearch.

Схема индекса:

PUT /products
{
  "settings": {
    "analysis": {
      "analyzer": {
        "russian_analyzer": {
          "type": "custom",
          "tokenizer": "standard",
          "filter": ["lowercase", "russian_stop", "russian_stemmer"]
        }
      },
      "filter": {
        "russian_stop": {
          "type": "stop",
          "stopwords": "_russian_"
        },
        "russian_stemmer": {
          "type": "stemmer",
          "language": "russian"
        }
      }
    }
  },
  "mappings": {
    "properties": {
      "title":       { "type": "text", "analyzer": "russian_analyzer", "boost": 3 },
      "brand":       { "type": "text", "analyzer": "russian_analyzer", "boost": 2 },
      "description": { "type": "text", "analyzer": "russian_analyzer" },
      "category_id": { "type": "keyword" },
      "price":       { "type": "double" },
      "status":      { "type": "keyword" },
      "created_at":  { "type": "date" }
    }
  }
}

Поиск с фасетами:

POST /products/_search
{
  "query": {
    "bool": {
      "must": {
        "multi_match": {
          "query": "беспроводные наушники",
          "fields": ["title^3", "brand^2", "description"],
          "type": "best_fields",
          "fuzziness": "AUTO"
        }
      },
      "filter": [
        { "term":  { "status": "published" } },
        { "range": { "price": { "gte": 1000, "lte": 15000 } } }
      ]
    }
  },
  "aggs": {
    "by_brand": {
      "terms": { "field": "brand.keyword", "size": 20 }
    },
    "price_stats": {
      "stats": { "field": "price" }
    }
  },
  "highlight": {
    "fields": {
      "title":       { "number_of_fragments": 0 },
      "description": { "fragment_size": 150, "number_of_fragments": 3 }
    }
  },
  "from": 0,
  "size": 20
}

Синхронизация с PostgreSQL:

# Вариант 1: синхронно в сервисе
async def create_product(data: ProductCreate, db: AsyncSession) -> Product:
    product = Product(**data.dict())
    db.add(product)
    await db.flush()          # получаем id

    await es.index(
        index='products',
        id=str(product.id),
        document=product_to_es_doc(product),
    )

    await db.commit()
    return product

# Вариант 2: через CDC (Change Data Capture)
# Debezium читает WAL PostgreSQL и публикует события в Kafka
# Consumer подписывается и обновляет Elasticsearch

Вариант с CDC более надёжен: данные попадут в ES даже если сервис упал в момент записи.

Типизированный клиент (Python)

from elasticsearch import AsyncElasticsearch
from pydantic import BaseModel
from typing import Any


class SearchResult(BaseModel):
    id: str
    score: float
    title: str
    price: float
    highlight: dict[str, list[str]] = {}


async def search_products(
    query: str,
    category_id: int | None = None,
    price_min: float | None = None,
    price_max: float | None = None,
    page: int = 1,
    per_page: int = 20,
) -> tuple[list[SearchResult], int]:

    es = AsyncElasticsearch(hosts=['http://localhost:9200'])

    filters: list[dict[str, Any]] = [{"term": {"status": "published"}}]
    if category_id:
        filters.append({"term": {"category_id": category_id}})
    if price_min or price_max:
        filters.append({"range": {"price": {
            **({"gte": price_min} if price_min else {}),
            **({"lte": price_max} if price_max else {}),
        }}})

    body = {
        "query": {
            "bool": {
                "must": {"multi_match": {
                    "query": query,
                    "fields": ["title^3", "brand^2", "description"],
                    "fuzziness": "AUTO",
                }},
                "filter": filters,
            }
        },
        "highlight": {"fields": {"title": {}, "description": {"fragment_size": 150}}},
        "from": (page - 1) * per_page,
        "size": per_page,
    }

    resp = await es.search(index="products", body=body)
    total = resp["hits"]["total"]["value"]
    hits = [
        SearchResult(
            id=h["_id"],
            score=h["_score"],
            highlight=h.get("highlight", {}),
            **h["_source"],
        )
        for h in resp["hits"]["hits"]
    ]
    return hits, total

Выбор решения

Сроки

PostgreSQL FTS (триггер, индекс, запросы, highlight): 1 день. Elasticsearch с русским анализатором, фасетами и CDC-синхронизацией через Debezium: 3–4 дня.