170 lines
12 KiB
PL/PgSQL
170 lines
12 KiB
PL/PgSQL
-- 166_objective_lots_history_writer.sql
|
||
-- #307 OBJ-6 — diff-gated objective_lots_history writer (DB-level trigger).
|
||
--
|
||
-- ПРОБЛЕМА (ground-truth, измерено на prod 2026-06-17):
|
||
-- objective_lots_history содержит 974 071 строк, но это НЕ delta-tracking, а wholesale-дамп:
|
||
-- snapshot_date | rows
|
||
-- 2026-05-17 | 324 217
|
||
-- 2026-05-19 | 324 510
|
||
-- 2026-06-03 | 325 344
|
||
-- = 3 даты × ~324k = по строке на КАЖДЫЙ лот в КАЖДЫЙ sync, вне зависимости от того, менялось
|
||
-- ли что-то. Источник: data/sql/70_parse_objective_raw.py пишет напрямую
|
||
-- INSERT INTO objective_lots_history (...) ... ON CONFLICT (objective_lot_id, snapshot_date)
|
||
-- DO NOTHING
|
||
-- для каждого лота при каждом sync (Module_Objective_Parser). Никакого diff нет — поэтому
|
||
-- таблица раздувается, а реальные изменения (price delta, free→reserved→sold, экспозиция)
|
||
-- утопают в шуме одинаковых снимков. Аудит SiteFinder_V2_Groundtruth_Audit_0602 уже отметил
|
||
-- «снапшотов всего 2-3 → динамика остатков data-starved».
|
||
--
|
||
-- ЦЕЛЬ OBJ-6: каждый sync лота фиксирует в history строку ТОЛЬКО когда реально изменилось
|
||
-- tracked-поле (цена за м², суммарные цены, status, is_sold) относительно ПОСЛЕДНЕГО снимка
|
||
-- этого лота. Это даёт чистую дельта-историю: price delta, переход статуса
|
||
-- free→reserved→sold, длительность экспозиции — без раздувания одинаковыми строками.
|
||
--
|
||
-- ПОЧЕМУ ТРИГГЕР (а не правка python-sync):
|
||
-- • Параллельный воркер сейчас редактирует objective-mapper (70_parse_objective_raw.py) →
|
||
-- избегаем file-overlap, не трогаем python.
|
||
-- • BEFORE/AFTER ROW-триггер на objective_lots ловит ВСЕ пути записи разом: основной
|
||
-- ON CONFLICT (objective_lot_id) DO UPDATE python-парсера (UPDATE-путь), INSERT новых лотов,
|
||
-- ручной SQL, будущие загрузчики. Один инвариант в БД вместо логики в каждом писателе.
|
||
-- • Diff-gate в БД дёшев: один индексный lookup последнего снимка по
|
||
-- objective_lots_history_lot_idx (objective_lot_id, snapshot_date DESC) — LIMIT 1.
|
||
--
|
||
-- ТРИГГЕР: AFTER INSERT OR UPDATE FOR EACH ROW на objective_lots.
|
||
-- AFTER (не BEFORE): пишем в ДРУГУЮ таблицу (history), строка objective_lots уже
|
||
-- зафиксирована — побочная вставка, не мутация NEW. FOR EACH ROW: нужен per-lot diff.
|
||
--
|
||
-- TRACKED-ПОЛЯ (diff-gate) — ровно «экономика лота», ради которой ведётся история:
|
||
-- price_per_m2_rub, price_calculated_total_rub, price_offer_total_rub, status, is_sold.
|
||
-- Сравнение через IS DISTINCT FROM — корректно трактует NULL (NULL vs NULL = НЕ изменилось,
|
||
-- value vs NULL = изменилось). Если последнего снимка НЕТ (новый лот) → пишем всегда.
|
||
--
|
||
-- price_delta_pct в history: вычисляем snapshot-over-snapshot
|
||
-- (NEW.price_per_m2_rub / prev.price_per_m2_rub − 1) × 100, округление до 2 знаков.
|
||
-- Это ОТЛИЧАЕТСЯ от objective_lots.price_delta_pct (внутренняя «Дельта цена, %» Объектива
|
||
-- из исходного отчёта) — здесь нужна именно дельта между снимками для velocity-аналитики.
|
||
-- Если prev отсутствует или prev.price_per_m2_rub = 0/NULL → price_delta_pct = NULL.
|
||
--
|
||
-- КОЛОНКИ history (точно по схеме objective_lots_history, ничего не выдумываем):
|
||
-- objective_lot_id, snapshot_date, status, is_sold, price_calculated_total_rub,
|
||
-- price_per_m2_rub, price_offer_total_rub, price_delta_pct, area_pd, contract_date,
|
||
-- registration_date, contract_type, bank_name, raw_id, complex_id
|
||
-- (id/fetched_at — defaults: nextval / now()).
|
||
--
|
||
-- ON CONFLICT (objective_lot_id, snapshot_date) DO UPDATE — НЕ DO NOTHING:
|
||
-- если за тот же день уже есть снимок (в т.ч. от python-парсера) и пришло НОВОЕ значение
|
||
-- tracked-поля, обновляем снимок этого дня свежими цифрами + пересчитанным delta. Это чинит
|
||
-- баг текущего python-писателя (DO NOTHING терял внутридневное изменение). Дубликат-строка
|
||
-- при коллизии не создаётся (защищён UNIQUE-констрейнтом).
|
||
--
|
||
-- СОСУЩЕСТВОВАНИЕ С PYTHON-ПИСАТЕЛЕМ: пока 70_parse_objective_raw.py продолжает свой прямой
|
||
-- INSERT ... DO NOTHING, триггер безопасен — его собственный INSERT уходит в DO UPDATE при
|
||
-- коллизии snapshot_date, дубль не создаётся. Чистый эффект: со следующего sync новые/
|
||
-- изменённые лоты получают корректную diff-строку, а wholesale-INSERT парсера для
|
||
-- НЕизменившихся лотов перекрывается DO NOTHING (его строка остаётся, но триггер её не
|
||
-- плодит). FOLLOW-UP (backend-задача, отдельный PR): убрать прямой history-INSERT из
|
||
-- 70_parse_objective_raw.py, оставив запись только за триггером — тогда history станет
|
||
-- полностью diff-gated. Эта миграция самодостаточна и безопасна без той правки.
|
||
--
|
||
-- BACKFILL не делаем: существующие 974k строк — исторический wholesale-дамп, не трогаем
|
||
-- (никаких DROP/TRUNCATE/DELETE). Триггер влияет только на будущие записи.
|
||
--
|
||
-- ПРОИЗВОДИТЕЛЬНОСТЬ на bulk-sync (~1.1M upsert): на строку — один индексный lookup
|
||
-- последнего снимка (Index Scan по objective_lots_history_lot_idx, LIMIT 1) + диф 5 полей.
|
||
-- Для НЕизменившихся лотов (типичный sync — меняется меньшинство) — ранний выход без INSERT.
|
||
-- Пренебрежимо относительно стоимости самого upsert в objective_lots.
|
||
--
|
||
-- Idempotent: CREATE OR REPLACE FUNCTION + DROP TRIGGER IF EXISTS перед CREATE TRIGGER
|
||
-- (PG16 не имеет CREATE OR REPLACE TRIGGER — паттерн как в 129_cad_parcels_quarter_cad_trigger).
|
||
-- BEGIN/COMMIT — атомарно. Re-apply чист. Auto-applied deploy.yml через _schema_migrations.
|
||
--
|
||
-- Deploy order: чистая СХЕМА, без зависимости от кода — безопасно катить в любой момент.
|
||
|
||
BEGIN;
|
||
|
||
-- ── 1. Триггер-функция: diff-gated запись снимка лота в objective_lots_history ──────────
|
||
CREATE OR REPLACE FUNCTION objective_lots_write_history() RETURNS TRIGGER AS $$
|
||
DECLARE
|
||
prev objective_lots_history%ROWTYPE;
|
||
has_prev boolean := false;
|
||
delta_pct numeric := NULL;
|
||
BEGIN
|
||
-- Последний снимок этого лота (любой более ранней или той же даты) — для diff и delta.
|
||
-- Index Scan по objective_lots_history_lot_idx (objective_lot_id, snapshot_date DESC).
|
||
SELECT * INTO prev
|
||
FROM objective_lots_history
|
||
WHERE objective_lot_id = NEW.objective_lot_id
|
||
AND snapshot_date <= NEW.snapshot_date
|
||
ORDER BY snapshot_date DESC
|
||
LIMIT 1;
|
||
has_prev := FOUND;
|
||
|
||
-- Diff-gate: пишем только если снимка ещё нет ИЛИ изменилось хоть одно tracked-поле.
|
||
-- IS DISTINCT FROM — NULL-safe (NULL vs NULL = не изменилось).
|
||
IF has_prev
|
||
AND prev.price_per_m2_rub IS NOT DISTINCT FROM NEW.price_per_m2_rub
|
||
AND prev.price_calculated_total_rub IS NOT DISTINCT FROM NEW.price_calculated_total_rub
|
||
AND prev.price_offer_total_rub IS NOT DISTINCT FROM NEW.price_offer_total_rub
|
||
AND prev.status IS NOT DISTINCT FROM NEW.status
|
||
AND prev.is_sold IS NOT DISTINCT FROM NEW.is_sold
|
||
THEN
|
||
-- Ничего из tracked не поменялось → не плодим снимок.
|
||
RETURN NULL; -- AFTER-триггер: возвращаемое значение игнорируется, RETURN NULL = no-op.
|
||
END IF;
|
||
|
||
-- snapshot-over-snapshot дельта цены за м² (для velocity-аналитики).
|
||
IF has_prev
|
||
AND prev.price_per_m2_rub IS NOT NULL
|
||
AND prev.price_per_m2_rub <> 0
|
||
AND NEW.price_per_m2_rub IS NOT NULL
|
||
THEN
|
||
delta_pct := round((NEW.price_per_m2_rub / prev.price_per_m2_rub - 1) * 100, 2);
|
||
END IF;
|
||
|
||
INSERT INTO objective_lots_history (
|
||
objective_lot_id, snapshot_date, status, is_sold,
|
||
price_calculated_total_rub, price_per_m2_rub, price_offer_total_rub,
|
||
price_delta_pct, area_pd, contract_date, registration_date,
|
||
contract_type, bank_name, raw_id, complex_id
|
||
) VALUES (
|
||
NEW.objective_lot_id, NEW.snapshot_date, NEW.status, NEW.is_sold,
|
||
NEW.price_calculated_total_rub, NEW.price_per_m2_rub, NEW.price_offer_total_rub,
|
||
delta_pct, NEW.area_pd, NEW.contract_date, NEW.registration_date,
|
||
NEW.contract_type, NEW.bank_name, NEW.raw_id, NEW.complex_id
|
||
)
|
||
ON CONFLICT (objective_lot_id, snapshot_date) DO UPDATE SET
|
||
status = EXCLUDED.status,
|
||
is_sold = EXCLUDED.is_sold,
|
||
price_calculated_total_rub = EXCLUDED.price_calculated_total_rub,
|
||
price_per_m2_rub = EXCLUDED.price_per_m2_rub,
|
||
price_offer_total_rub = EXCLUDED.price_offer_total_rub,
|
||
price_delta_pct = EXCLUDED.price_delta_pct,
|
||
area_pd = EXCLUDED.area_pd,
|
||
contract_date = EXCLUDED.contract_date,
|
||
registration_date = EXCLUDED.registration_date,
|
||
contract_type = EXCLUDED.contract_type,
|
||
bank_name = EXCLUDED.bank_name,
|
||
raw_id = EXCLUDED.raw_id,
|
||
complex_id = EXCLUDED.complex_id,
|
||
fetched_at = now();
|
||
|
||
RETURN NULL; -- AFTER ROW trigger — результат не используется.
|
||
END;
|
||
$$ LANGUAGE plpgsql;
|
||
|
||
COMMENT ON FUNCTION objective_lots_write_history() IS
|
||
'AFTER INSERT OR UPDATE-триггер objective_lots (#307 OBJ-6): diff-gated запись снимка в '
|
||
'objective_lots_history. Пишет строку ТОЛЬКО при изменении tracked-полей (price_per_m2_rub, '
|
||
'price_calculated_total_rub, price_offer_total_rub, status, is_sold) против последнего '
|
||
'снимка лота (IS DISTINCT FROM, NULL-safe). Считает snapshot-over-snapshot price_delta_pct. '
|
||
'ON CONFLICT(objective_lot_id,snapshot_date) DO UPDATE — внутридневное изменение обновляет '
|
||
'снимок дня, дубль не создаётся. Заменяет wholesale-дамп python-парсера на дельта-историю.';
|
||
|
||
-- ── 2. Триггер AFTER INSERT OR UPDATE на objective_lots ─────────────────────────────────
|
||
-- PG16 не имеет CREATE OR REPLACE TRIGGER → DROP IF EXISTS + CREATE (idempotent, паттерн 129).
|
||
DROP TRIGGER IF EXISTS objective_lots_write_history ON objective_lots;
|
||
CREATE TRIGGER objective_lots_write_history
|
||
AFTER INSERT OR UPDATE ON objective_lots
|
||
FOR EACH ROW EXECUTE FUNCTION objective_lots_write_history();
|
||
|
||
COMMIT;
|