feat(tradein): search matview + indexes (Phase 3.1) (#469)
Multi-Source Integration Phase 3.1. Foundation for /api/v1/search p95 < 100ms on 100K+ listings. SQL 050_search_optimization.sql: - CREATE EXTENSION IF NOT EXISTS pg_trgm - ALTER listings ADD COLUMN tsv tsvector GENERATED ALWAYS AS (...) STORED - 10 indexes (geom GIST, composite, gin_trgm, GIN tsv, partial kadastr, ...) - CREATE MATERIALIZED VIEW listings_search_mv (per-listing aggregated across sources) - 5 matview indexes (UNIQUE on listing_id for REFRESH CONCURRENTLY) Celery task tradein.refresh_search_matview: - REFRESH MATERIALIZED VIEW CONCURRENTLY listings_search_mv (psycopg v3) - Logged duration via logger.info - Graceful try/except ImportError fallback (no celery_app.py in tradein-mvp yet) - Beat schedule snippet in docstring for future wiring Fixups (deep-code-reviewer 2026-05-23): - h.ratings_count → h.reviews_count (alias house_ratings_count preserved) - Strip +psycopg dialect prefix from DSN for libpq compat Refs Master Plan sec 5.1 + 6.5 + 10.1.
This commit is contained in:
parent
51650276b2
commit
e46b7d778e
3 changed files with 263 additions and 0 deletions
0
tradein-mvp/backend/app/tasks/__init__.py
Normal file
0
tradein-mvp/backend/app/tasks/__init__.py
Normal file
70
tradein-mvp/backend/app/tasks/refresh_search_matview.py
Normal file
70
tradein-mvp/backend/app/tasks/refresh_search_matview.py
Normal file
|
|
@ -0,0 +1,70 @@
|
|||
"""Refresh listings_search_mv materialized view (daily 3 AM).
|
||||
|
||||
Currently tradein-mvp has no Celery infrastructure (verified — no celery_app.py exists
|
||||
in tradein-mvp/backend/app/). This module provides both:
|
||||
- A Celery task wrapper (active only if celery is installed and celery_app present).
|
||||
- A plain async refresh_search_matview() callable for cron / one-shot invocation.
|
||||
|
||||
NEEDS COORDINATION: main session must decide whether to bootstrap Celery in tradein-mvp
|
||||
or run via external scheduler (systemd timer / OS cron). Until then, no Beat schedule.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import time
|
||||
|
||||
import psycopg
|
||||
|
||||
from app.core.config import settings
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def refresh_search_matview() -> None:
|
||||
"""REFRESH MATERIALIZED VIEW CONCURRENTLY listings_search_mv (psycopg v3, sync).
|
||||
|
||||
Logs duration. Idempotent. Safe to run during read traffic (CONCURRENTLY).
|
||||
"""
|
||||
start = time.monotonic()
|
||||
# DATABASE_URL is SQLAlchemy dialect form (postgresql+psycopg://...) in tradein-mvp
|
||||
# (see tradein-mvp/docker-compose.prod.yml). libpq / psycopg.connect() accepts only
|
||||
# postgresql:// or postgres:// — strip the +psycopg dialect prefix.
|
||||
dsn = settings.database_url.replace("postgresql+psycopg://", "postgresql://", 1)
|
||||
with psycopg.connect(dsn, autocommit=True) as conn:
|
||||
with conn.cursor() as cur:
|
||||
cur.execute("REFRESH MATERIALIZED VIEW CONCURRENTLY listings_search_mv")
|
||||
elapsed = time.monotonic() - start
|
||||
logger.info("listings_search_mv refresh completed in %.2fs", elapsed)
|
||||
|
||||
|
||||
# Celery-task wrapper — only registers if celery_app exists.
|
||||
try:
|
||||
from app.celery_app import celery_app # type: ignore[import-not-found]
|
||||
except ImportError:
|
||||
logger.info(
|
||||
"Celery not configured in tradein-mvp — refresh_search_matview() callable directly. "
|
||||
"Bootstrap app/celery_app.py to enable scheduled execution.",
|
||||
)
|
||||
else:
|
||||
|
||||
@celery_app.task(name="tradein.refresh_search_matview") # type: ignore[misc]
|
||||
def refresh_search_matview_task() -> None:
|
||||
"""Celery wrapper for refresh_search_matview()."""
|
||||
refresh_search_matview()
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# Beat schedule snippet — paste into app/celery_app.py if/when Celery is bootstrapped:
|
||||
#
|
||||
# from celery.schedules import crontab
|
||||
#
|
||||
# celery_app.conf.beat_schedule = {
|
||||
# 'tradein-refresh-matview': {
|
||||
# 'task': 'tradein.refresh_search_matview',
|
||||
# 'schedule': crontab(hour=3, minute=0),
|
||||
# },
|
||||
# # ... other tradein beat entries
|
||||
# }
|
||||
#
|
||||
# Per master plan sec 10.1 ('refresh-search-matview-daily': crontab(minute=0, hour=3)).
|
||||
# =============================================================================
|
||||
193
tradein-mvp/backend/data/sql/050_search_optimization.sql
Normal file
193
tradein-mvp/backend/data/sql/050_search_optimization.sql
Normal file
|
|
@ -0,0 +1,193 @@
|
|||
-- 050_search_optimization.sql
|
||||
-- Purpose: Search-time performance optimization for cross-source listings.
|
||||
-- Adds pg_trgm + generated tsvector + 10 indexes (master plan sec 5.1)
|
||||
-- + listings_search_mv materialized view (master plan sec 6.5)
|
||||
-- + 5 matview indexes.
|
||||
-- Dependencies: 002_core_tables.sql (listings), 009_houses.sql, 028_matching_tables.sql,
|
||||
-- 030_listings_alter_yandex.sql.
|
||||
-- Deploy order: Apply after 046_views.sql.
|
||||
-- Re-run safe: all ADD COLUMN / CREATE INDEX / CREATE MATERIALIZED VIEW use IF NOT EXISTS.
|
||||
--
|
||||
-- Sources:
|
||||
-- Multi-Source Integration Master Plan sec 5.1 (10 индексов on listings)
|
||||
-- Multi-Source Integration Master Plan sec 6.5 (listings_search_mv)
|
||||
|
||||
BEGIN;
|
||||
|
||||
-- =============================================================================
|
||||
-- Extensions
|
||||
-- =============================================================================
|
||||
CREATE EXTENSION IF NOT EXISTS pg_trgm;
|
||||
|
||||
-- =============================================================================
|
||||
-- Generated tsvector column for full-text search (russian config)
|
||||
-- Master plan sec 5.1 index #6 + sec 6.5 matview tsv definition.
|
||||
-- Weight A → description, weight B → address.
|
||||
-- =============================================================================
|
||||
DO $$
|
||||
BEGIN
|
||||
IF NOT EXISTS (
|
||||
SELECT 1 FROM information_schema.columns
|
||||
WHERE table_name = 'listings' AND column_name = 'tsv'
|
||||
) THEN
|
||||
ALTER TABLE listings ADD COLUMN tsv tsvector
|
||||
GENERATED ALWAYS AS (
|
||||
setweight(to_tsvector('russian', coalesce(description, '')), 'A')
|
||||
|| setweight(to_tsvector('russian', coalesce(address, '')), 'B')
|
||||
) STORED;
|
||||
END IF;
|
||||
END
|
||||
$$;
|
||||
|
||||
-- =============================================================================
|
||||
-- 10 indexes per master plan sec 5.1
|
||||
-- =============================================================================
|
||||
|
||||
-- 1) Hot path: active listings by geo radius (GIST partial)
|
||||
CREATE INDEX IF NOT EXISTS listings_geom_active_idx
|
||||
ON listings USING GIST (geom)
|
||||
WHERE is_active = true;
|
||||
|
||||
-- 2) Active + filter composite: rooms + price_rub + area_m2 + scraped_at DESC
|
||||
CREATE INDEX IF NOT EXISTS listings_active_filter_idx
|
||||
ON listings (rooms, price_rub, area_m2, scraped_at DESC)
|
||||
WHERE is_active = true;
|
||||
|
||||
-- 3) House + price (price range within a house)
|
||||
CREATE INDEX IF NOT EXISTS listings_house_price_idx
|
||||
ON listings (house_id_fk, price_rub)
|
||||
WHERE is_active = true;
|
||||
|
||||
-- 4) Recency sort (scraped_at DESC)
|
||||
CREATE INDEX IF NOT EXISTS listings_scraped_desc_idx
|
||||
ON listings (scraped_at DESC)
|
||||
WHERE is_active = true;
|
||||
|
||||
-- 5) Address fuzzy search (trigram)
|
||||
CREATE INDEX IF NOT EXISTS listings_address_trgm_idx
|
||||
ON listings USING GIN (address gin_trgm_ops)
|
||||
WHERE address IS NOT NULL;
|
||||
|
||||
-- 6) Full-text search (tsv column above)
|
||||
CREATE INDEX IF NOT EXISTS listings_tsv_idx
|
||||
ON listings USING GIN (tsv);
|
||||
|
||||
-- 7) Cadastral exact lookup (partial)
|
||||
CREATE INDEX IF NOT EXISTS listings_kadastr_idx
|
||||
ON listings (kadastr_num)
|
||||
WHERE kadastr_num IS NOT NULL;
|
||||
|
||||
-- 8) listing_sources (listing_id) — for source aggregation
|
||||
CREATE INDEX IF NOT EXISTS listing_sources_listing_idx2
|
||||
ON listing_sources (listing_id);
|
||||
|
||||
-- 9) listing_sources (listing_id, price_rub) — for price divergence detection
|
||||
CREATE INDEX IF NOT EXISTS listing_sources_price_divergence_idx2
|
||||
ON listing_sources (listing_id, price_rub)
|
||||
WHERE price_rub IS NOT NULL;
|
||||
|
||||
-- 10) houses GIST + fingerprint + kadastr (consolidated houses indexes)
|
||||
CREATE INDEX IF NOT EXISTS houses_geom_idx2
|
||||
ON houses USING GIST (geom);
|
||||
CREATE INDEX IF NOT EXISTS houses_fingerprint_idx2
|
||||
ON houses (address_fingerprint)
|
||||
WHERE address_fingerprint IS NOT NULL;
|
||||
CREATE INDEX IF NOT EXISTS houses_kadastr_idx2
|
||||
ON houses (cadastral_number)
|
||||
WHERE cadastral_number IS NOT NULL;
|
||||
|
||||
COMMIT;
|
||||
|
||||
-- =============================================================================
|
||||
-- Materialized view: listings_search_mv (master plan sec 6.5)
|
||||
-- Separate BEGIN/COMMIT block — matview creation can be slow on populated tables.
|
||||
-- =============================================================================
|
||||
BEGIN;
|
||||
|
||||
DROP MATERIALIZED VIEW IF EXISTS listings_search_mv;
|
||||
|
||||
CREATE MATERIALIZED VIEW listings_search_mv AS
|
||||
SELECT
|
||||
l.id AS listing_id,
|
||||
l.source,
|
||||
l.source_url,
|
||||
l.address,
|
||||
l.geom,
|
||||
l.lat,
|
||||
l.lon AS lng,
|
||||
l.rooms,
|
||||
l.area_m2 AS total_area,
|
||||
l.floor,
|
||||
l.total_floors,
|
||||
l.price_rub,
|
||||
l.price_per_m2,
|
||||
l.kadastr_num,
|
||||
l.is_active,
|
||||
l.scraped_at,
|
||||
-- House denorm
|
||||
h.id AS house_id,
|
||||
h.year_built,
|
||||
h.house_class,
|
||||
h.developer_name,
|
||||
h.rating AS house_rating,
|
||||
h.reviews_count AS house_ratings_count,
|
||||
-- Cross-source aggregates
|
||||
(SELECT count(*) FROM listing_sources ls WHERE ls.listing_id = l.id) AS source_count,
|
||||
(SELECT array_agg(DISTINCT ext_source) FROM listing_sources ls WHERE ls.listing_id = l.id) AS sources,
|
||||
(SELECT bool_or(ext_source = 'avito') FROM listing_sources ls WHERE ls.listing_id = l.id) AS has_avito,
|
||||
(SELECT bool_or(ext_source = 'cian') FROM listing_sources ls WHERE ls.listing_id = l.id) AS has_cian,
|
||||
(SELECT bool_or(ext_source = 'yandex_realty') FROM listing_sources ls WHERE ls.listing_id = l.id) AS has_yandex,
|
||||
-- Price percentile within house
|
||||
(SELECT percentile_cont(0.5) WITHIN GROUP (ORDER BY ll.price_per_m2)
|
||||
FROM listings ll
|
||||
WHERE ll.house_id_fk = l.house_id_fk AND ll.is_active = true) AS house_median_ppm2,
|
||||
-- DEFAULT: district / distance_to_metro_m / last_price_change / photos_count
|
||||
-- not present in current schema — placeholders. (vault sec 6.5 doesn't define them either;
|
||||
-- task description listed them but underlying columns don't exist yet — populate post-Phase 4.)
|
||||
NULL::text AS district,
|
||||
NULL::int AS distance_to_metro_m,
|
||||
NULL::timestamptz AS last_price_change,
|
||||
NULL::int AS photos_count,
|
||||
-- Trigram-ready columns
|
||||
l.address AS address_trgm,
|
||||
-- Aggregated tsv (description + address + developer_name)
|
||||
to_tsvector('russian',
|
||||
coalesce(l.description, '') || ' ' ||
|
||||
coalesce(l.address, '') || ' ' ||
|
||||
coalesce(h.developer_name, '')
|
||||
) AS tsv
|
||||
FROM listings l
|
||||
LEFT JOIN houses h ON h.id = l.house_id_fk
|
||||
WHERE l.is_active = true
|
||||
AND COALESCE(l.canonical, true) = true;
|
||||
|
||||
-- =============================================================================
|
||||
-- 5 matview indexes per master plan sec 6.5
|
||||
-- =============================================================================
|
||||
CREATE UNIQUE INDEX listings_search_mv_id_idx
|
||||
ON listings_search_mv (listing_id);
|
||||
|
||||
CREATE INDEX listings_search_mv_geom_idx
|
||||
ON listings_search_mv USING GIST (geom);
|
||||
|
||||
CREATE INDEX listings_search_mv_filters_idx
|
||||
ON listings_search_mv (rooms, price_rub, total_area, scraped_at DESC);
|
||||
|
||||
CREATE INDEX listings_search_mv_address_trgm_idx
|
||||
ON listings_search_mv USING GIN (address_trgm gin_trgm_ops);
|
||||
|
||||
CREATE INDEX listings_search_mv_tsv_idx
|
||||
ON listings_search_mv USING GIN (tsv);
|
||||
|
||||
CREATE INDEX listings_search_mv_sources_idx
|
||||
ON listings_search_mv (has_avito, has_cian, has_yandex);
|
||||
|
||||
COMMIT;
|
||||
|
||||
-- =============================================================================
|
||||
-- Initial populate: REFRESH CONCURRENTLY would fail (matview just created — never populated).
|
||||
-- First populate is implicit via CREATE MATERIALIZED VIEW AS SELECT ... above.
|
||||
-- Subsequent refreshes (Celery task) use CONCURRENTLY.
|
||||
-- Skip explicit REFRESH here — empty source data scenarios will result in empty matview but
|
||||
-- creation is still valid.
|
||||
-- =============================================================================
|
||||
Loading…
Add table
Reference in a new issue