fix(tradein): data-layer tail A2+B+C — trgm index, FDW options, matview refresh (Refs #769) #804
4 changed files with 213 additions and 0 deletions
|
|
@ -468,6 +468,51 @@ async def trigger_listing_source_snapshot_run(
|
|||
return run_id
|
||||
|
||||
|
||||
async def trigger_refresh_search_matview_run(
|
||||
db: Session, schedule_row: dict[str, Any]
|
||||
) -> int | None:
|
||||
"""Создать scrape_runs + запустить REFRESH MATERIALIZED VIEW CONCURRENTLY listings_search_mv.
|
||||
|
||||
listings_search_mv активно читается search_query.py (SELECT FROM listings_search_mv).
|
||||
Без периодического refresh матвью устаревает по мере появления новых listings.
|
||||
Задача синхронная (DB-only, БЕЗ внешних HTTP-вызовов) — гоняем в run_in_executor
|
||||
по образцу trigger_asking_to_sold_ratio_run (#648).
|
||||
|
||||
refresh_search_matview() использует собственное psycopg-соединение с autocommit=True
|
||||
(REFRESH CONCURRENTLY не может работать внутри транзакции). scrape_runs lifecycle
|
||||
(mark_done/mark_failed) управляется через отдельную SessionLocal.
|
||||
|
||||
SAFE to enable — schedule seed enabled=true (088), pure internal DB op.
|
||||
Окно 03:00-04:00 UTC — ночное, до rosreestr_dkp_import (04:00-06:00 UTC).
|
||||
|
||||
Returns run_id (или None если skip — есть running run).
|
||||
"""
|
||||
run_id = _claim_run(db, schedule_row)
|
||||
if run_id is None:
|
||||
return None
|
||||
|
||||
async def _run() -> None:
|
||||
run_db = SessionLocal()
|
||||
try:
|
||||
from app.tasks.refresh_search_matview import refresh_search_matview
|
||||
|
||||
loop = asyncio.get_event_loop()
|
||||
# refresh_search_matview() manages its own connection with autocommit=True
|
||||
# (required for REFRESH MATERIALIZED VIEW CONCURRENTLY outside a transaction).
|
||||
await loop.run_in_executor(None, refresh_search_matview)
|
||||
runs_mod.mark_done(run_db, run_id, {})
|
||||
except Exception:
|
||||
logger.exception("scheduler: refresh_search_matview crashed run_id=%d", run_id)
|
||||
runs_mod.mark_failed(run_db, run_id, "refresh_search_matview failed", {})
|
||||
finally:
|
||||
run_db.close()
|
||||
|
||||
task = asyncio.create_task(_run())
|
||||
task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None)
|
||||
logger.info("scheduler: triggered refresh_search_matview run_id=%d", run_id)
|
||||
return run_id
|
||||
|
||||
|
||||
async def trigger_asking_to_sold_ratio_run(db: Session, schedule_row: dict[str, Any]) -> int | None:
|
||||
"""Создать scrape_runs + launch recompute_asking_to_sold_ratios в executor (sync DB-only task).
|
||||
|
||||
|
|
@ -782,6 +827,8 @@ async def scheduler_loop() -> None:
|
|||
await trigger_listing_source_snapshot_run(db, sch)
|
||||
elif source == "asking_to_sold_ratio_refresh":
|
||||
await trigger_asking_to_sold_ratio_run(db, sch)
|
||||
elif source == "refresh_search_matview":
|
||||
await trigger_refresh_search_matview_run(db, sch)
|
||||
else:
|
||||
logger.warning("scheduler: unknown source=%s, skip", source)
|
||||
finally:
|
||||
|
|
|
|||
|
|
@ -0,0 +1,34 @@
|
|||
-- 086_deals_address_trgm_index.sql
|
||||
-- Issue #769 Part A2 (finding #10) — GIN trigram index on deals.address.
|
||||
--
|
||||
-- Context:
|
||||
-- The deals table (002_core_tables.sql line 84) has an `address text` column used
|
||||
-- in street-match queries (import_rosreestr_dkp in scheduler.py builds ILIKE / text
|
||||
-- predicates). Without a trigram index the planner falls back to seq-scan on the
|
||||
-- full deals table. pg_trgm is already enabled in 050_search_optimization.sql but
|
||||
-- only for the listings table — not for deals.
|
||||
--
|
||||
-- Idempotency:
|
||||
-- CREATE EXTENSION IF NOT EXISTS — safe on re-run (ext already active = no-op).
|
||||
-- CREATE INDEX IF NOT EXISTS — safe on re-run.
|
||||
-- Both inside a single BEGIN/COMMIT — plain CREATE INDEX (not CONCURRENTLY) is
|
||||
-- required to stay transaction-safe (CONCURRENTLY cannot run inside a txn block).
|
||||
-- Acceptable here because deals is populated incrementally via FDW import batches
|
||||
-- and the table starts small.
|
||||
--
|
||||
-- Dependencies:
|
||||
-- 002_core_tables.sql (deals table with address column).
|
||||
--
|
||||
-- Deploy order:
|
||||
-- Apply after 002_core_tables.sql. No application code changes required.
|
||||
-- Post-deploy QA: EXPLAIN (ANALYZE, BUFFERS) SELECT ... FROM deals WHERE address
|
||||
-- ILIKE '%улица%'; should show Bitmap Index Scan on deals_address_trgm_idx.
|
||||
|
||||
BEGIN;
|
||||
|
||||
CREATE EXTENSION IF NOT EXISTS pg_trgm;
|
||||
|
||||
CREATE INDEX IF NOT EXISTS deals_address_trgm_idx
|
||||
ON deals USING gin (address gin_trgm_ops);
|
||||
|
||||
COMMIT;
|
||||
69
tradein-mvp/backend/data/sql/087_fdw_server_options.sql
Normal file
69
tradein-mvp/backend/data/sql/087_fdw_server_options.sql
Normal file
|
|
@ -0,0 +1,69 @@
|
|||
-- 087_fdw_server_options.sql
|
||||
-- Issue #769 Part B (finding #13) — FDW server performance options.
|
||||
--
|
||||
-- Context:
|
||||
-- SERVER gendesign_remote was created by 060_postgres_fdw_extension.sql.
|
||||
-- The current options are: host, dbname, port, extensions, connect_timeout.
|
||||
-- Two planning options are missing:
|
||||
-- fetch_size — rows fetched per round-trip (default 100, raise to 10000
|
||||
-- for bulk cursor-based pagination in import_rosreestr_dkp).
|
||||
-- use_remote_estimate — let planner query ANALYZE stats from the remote server
|
||||
-- to produce accurate cost estimates for the 6.8M-row
|
||||
-- rosreestr_deals foreign table and quarter_price_index MV.
|
||||
--
|
||||
-- Idempotency:
|
||||
-- ALTER SERVER ... OPTIONS (ADD ...) raises ERROR if the option already exists.
|
||||
-- ALTER SERVER ... OPTIONS (SET ...) raises ERROR if the option does not exist yet.
|
||||
-- Therefore a DO $$ block checks pg_foreign_server.srvoptions for each option:
|
||||
-- - If present → SET (update the value in case it was previously set differently).
|
||||
-- - If absent → ADD (insert for the first time).
|
||||
-- This is safe to run repeatedly regardless of prior state.
|
||||
--
|
||||
-- Dependencies:
|
||||
-- 060_postgres_fdw_extension.sql (gendesign_remote server must exist).
|
||||
--
|
||||
-- Deploy order:
|
||||
-- Apply after 060_postgres_fdw_extension.sql. No FDW USER MAPPING changes required.
|
||||
-- Effect is immediate — next FDW query uses the new planner options.
|
||||
|
||||
BEGIN;
|
||||
|
||||
DO $$
|
||||
DECLARE
|
||||
v_opts text[];
|
||||
BEGIN
|
||||
-- Read current srvoptions array for gendesign_remote.
|
||||
SELECT COALESCE(srvoptions, ARRAY[]::text[])
|
||||
INTO v_opts
|
||||
FROM pg_foreign_server
|
||||
WHERE srvname = 'gendesign_remote';
|
||||
|
||||
IF NOT FOUND THEN
|
||||
RAISE EXCEPTION 'FDW server gendesign_remote not found — apply 060_postgres_fdw_extension.sql first';
|
||||
END IF;
|
||||
|
||||
-- fetch_size: rows per FDW cursor round-trip (default 100 → 10000)
|
||||
IF v_opts @> ARRAY['fetch_size=10000'] THEN
|
||||
NULL; -- already correct value, no-op
|
||||
ELSIF EXISTS (
|
||||
SELECT 1 FROM unnest(v_opts) AS opt WHERE opt LIKE 'fetch_size=%'
|
||||
) THEN
|
||||
ALTER SERVER gendesign_remote OPTIONS (SET fetch_size '10000');
|
||||
ELSE
|
||||
ALTER SERVER gendesign_remote OPTIONS (ADD fetch_size '10000');
|
||||
END IF;
|
||||
|
||||
-- use_remote_estimate: push ANALYZE stats collection to remote for better plans
|
||||
IF v_opts @> ARRAY['use_remote_estimate=true'] THEN
|
||||
NULL; -- already correct value, no-op
|
||||
ELSIF EXISTS (
|
||||
SELECT 1 FROM unnest(v_opts) AS opt WHERE opt LIKE 'use_remote_estimate=%'
|
||||
) THEN
|
||||
ALTER SERVER gendesign_remote OPTIONS (SET use_remote_estimate 'true');
|
||||
ELSE
|
||||
ALTER SERVER gendesign_remote OPTIONS (ADD use_remote_estimate 'true');
|
||||
END IF;
|
||||
END
|
||||
$$;
|
||||
|
||||
COMMIT;
|
||||
|
|
@ -0,0 +1,63 @@
|
|||
-- 088_scrape_schedules_seed_search_matview_refresh.sql
|
||||
-- Issue #769 Part C (finding #14) — seed scrape_schedules row for daily
|
||||
-- listings_search_mv materialized view refresh.
|
||||
--
|
||||
-- Context:
|
||||
-- listings_search_mv (created in 050_search_optimization.sql) is actively read by
|
||||
-- search_query.py (SELECT FROM listings_search_mv) for the /api/v1/search endpoint.
|
||||
-- The refresh callable (app/tasks/refresh_search_matview.py) exists but was NEVER
|
||||
-- wired into the in-app scheduler: no scrape_schedules row, no dispatch branch in
|
||||
-- scheduler_loop(). Without a scheduled refresh the matview becomes stale after the
|
||||
-- initial CREATE MATERIALIZED VIEW population.
|
||||
--
|
||||
-- Fix:
|
||||
-- 1. scheduler.py — added trigger_refresh_search_matview_run() + dispatch branch
|
||||
-- `elif source == "refresh_search_matview"` in scheduler_loop().
|
||||
-- 2. This migration — seeds the scrape_schedules row (enabled=true, 03:00-04:00 UTC).
|
||||
--
|
||||
-- Schedule window 03:00-04:00 UTC:
|
||||
-- Nightly, before rosreestr_dkp_import (04:00-06:00 UTC). Refresh takes <10s on a
|
||||
-- non-empty matview (CONCURRENTLY — readers are not blocked).
|
||||
--
|
||||
-- next_run_at bootstrapped to tomorrow 03:00 UTC so the scheduler does not fire
|
||||
-- immediately on deploy (same pattern as 078/079/082).
|
||||
--
|
||||
-- Idempotent: ON CONFLICT (source) DO NOTHING — safe to re-apply.
|
||||
--
|
||||
-- Dependencies:
|
||||
-- 052_scrape_schedules.sql (table + UNIQUE(source)).
|
||||
-- 050_search_optimization.sql (listings_search_mv must exist in DB).
|
||||
-- scheduler.py change (trigger_refresh_search_matview_run must be deployed).
|
||||
--
|
||||
-- Deploy order:
|
||||
-- Apply this migration after deploying the scheduler.py change so the scheduler
|
||||
-- can dispatch the new source name correctly on first fire.
|
||||
|
||||
BEGIN;
|
||||
|
||||
INSERT INTO scrape_schedules (
|
||||
source,
|
||||
enabled,
|
||||
window_start_hour,
|
||||
window_end_hour,
|
||||
next_run_at,
|
||||
default_params
|
||||
)
|
||||
VALUES
|
||||
(
|
||||
'refresh_search_matview',
|
||||
true, -- SAFE: pure internal DB, REFRESH CONCURRENTLY
|
||||
3,
|
||||
4,
|
||||
((CURRENT_DATE + INTERVAL '1 day') + make_interval(hours => 3)) AT TIME ZONE 'UTC',
|
||||
'{}'::jsonb
|
||||
)
|
||||
ON CONFLICT (source) DO NOTHING;
|
||||
|
||||
COMMENT ON TABLE scrape_schedules IS
|
||||
'In-app scheduler config (заменяет cron-script setup). '
|
||||
'Sources: avito_city_sweep, yandex_city_sweep (dormant, #561), '
|
||||
'cian_history_backfill, rosreestr_dkp_import, listing_source_snapshot (#570), '
|
||||
'asking_to_sold_ratio_refresh (#648), refresh_search_matview (#769).';
|
||||
|
||||
COMMIT;
|
||||
Loading…
Add table
Reference in a new issue