diff --git a/tradein-mvp/backend/app/services/house_dedup_merge.py b/tradein-mvp/backend/app/services/house_dedup_merge.py index 330b66c0..0f244b97 100644 --- a/tradein-mvp/backend/app/services/house_dedup_merge.py +++ b/tradein-mvp/backend/app/services/house_dedup_merge.py @@ -92,6 +92,30 @@ BACKFILL (reduces recurrence): (same as 108) so the matching pipeline's Tier-1/Tier-2 finds the keeper next scrape and does not immediately re-split it. +MERGE JOURNAL — the merge is REVERSIBLE (#2690, migration 230): + Every loser gets a row in `house_merge_log` written in the SAME transaction as the merge: + the full jsonb snapshot of the deleted row, the keeper's snapshot BEFORE the identity + carry-over, the ids of every child row whose FK moved, the full snapshots of every child row + a UNIQUE collision destroyed, plus the grounds — which pass, which cluster-key VALUE fired, + whether the geo guard was on, and the keeper↔loser distance in metres. + + This exists because the merge used to leave no restorable trace: losers were hard-deleted + with their children and the only record of «what went into what» was a log line, in a + container whose logs rotate faster than a day. A day after a run nobody could even NAME the + pairs, and the only rollback was restoring the whole database. + + Undo: `SELECT * FROM house_merge_undo(batch_id)` inside a transaction — restores the loser + rows, points the children back, re-inserts the destroyed children, and un-does the identity + carry-over on the keeper, reporting per record what it could and could not restore. + + NOTE the journal is deliberately NEUTRAL to the merge rule: it changes no cluster key, no + keeper rule and no guard. It only makes whatever the pass decides reversible — which is the + precondition for revisiting those decisions at all (#2690, #1772). + + distance_m is recorded on BOTH passes, including the fias pass whose geo guard is off. That + asymmetry — merge allowed without a proximity check — was invisible in data before; now + «how many merges happened beyond N metres, on which key» is one query. + IDEMPOTENCY: Every UPDATE/DELETE keys off a temp mapping of (loser→keeper). On a clean table the mapping is empty → every statement touches 0 rows → no-op. Re-running is safe. @@ -105,8 +129,10 @@ psycopg v3: all SQL uses CAST(:x AS type), never the colon-colon bound-param cas from __future__ import annotations +import json import logging import time +import uuid from dataclasses import dataclass, field from typing import Any @@ -260,7 +286,16 @@ def _mapping_sql(cluster_key_case: str, *, apply_geo_guard: bool = True) -> str: -- DIFFERENT house_fias_id — provably different buildings the cluster key collapsed (canon -- slash-collapse «Сулимова, 32»/«Сулимова, 3/2»). No-op for the fias pass (one fias per -- cluster) and for canon clusters where at most one side carries a fias. - SELECT id AS loser_id, keeper_id, norm_address + -- + -- cluster_key / distance_m are carried out of the mapping for the MERGE JOURNAL (#2690): + -- cluster_key records WHICH key value fired, distance_m how far apart the two rows were. + -- distance_m is computed even when the geo guard is OFF for this pass — that is precisely + -- the case where nothing else records the distance, and #2690 had no way to ask + -- «how many merges happened at distances the guard would have blocked» from data. + SELECT id AS loser_id, keeper_id, norm_address, cluster_key, + CASE WHEN keeper_geom IS NOT NULL AND loser_geom IS NOT NULL + THEN ST_DistanceSphere(loser_geom, keeper_geom) + END AS distance_m FROM ranked WHERE rn > 1 AND id <> keeper_id{geo_guard} @@ -289,6 +324,7 @@ _STEPS: list[tuple[str, str]] = [ SET house_id_fk = m.keeper_id FROM _1772_dup_mapping m WHERE l.house_id_fk = m.loser_id + RETURNING m.loser_id, l.id AS child_id """, ), ( @@ -298,6 +334,7 @@ _STEPS: list[tuple[str, str]] = [ SET house_id = m.keeper_id FROM _1772_dup_mapping m WHERE hph.house_id = m.loser_id + RETURNING m.loser_id, hph.id AS child_id """, ), ( @@ -307,6 +344,7 @@ _STEPS: list[tuple[str, str]] = [ SET house_id = m.keeper_id FROM _1772_dup_mapping m WHERE hr.house_id = m.loser_id + RETURNING m.loser_id, hr.id AS child_id """, ), ( @@ -316,6 +354,7 @@ _STEPS: list[tuple[str, str]] = [ SET house_id = m.keeper_id FROM _1772_dup_mapping m WHERE hrc.house_id = m.loser_id + RETURNING m.loser_id, hrc.id AS child_id """, ), ( @@ -325,6 +364,7 @@ _STEPS: list[tuple[str, str]] = [ SET house_id = m.keeper_id FROM _1772_dup_mapping m WHERE ev.house_id = m.loser_id + RETURNING m.loser_id, ev.id AS child_id """, ), # ── UNIQUE(ext_source, ext_id): delete colliding losers, re-point rest ───── @@ -340,6 +380,7 @@ _STEPS: list[tuple[str, str]] = [ AND hs2.ext_source = hs.ext_source AND hs2.ext_id = hs.ext_id ) + RETURNING hs.house_id AS loser_id, to_jsonb(hs.*) AS row_snapshot """, ), ( @@ -349,6 +390,7 @@ _STEPS: list[tuple[str, str]] = [ SET house_id = m.keeper_id FROM _1772_dup_mapping m WHERE hs.house_id = m.loser_id + RETURNING m.loser_id, hs.id AS child_id """, ), # ── UNIQUE(normalized_address): delete colliding losers, re-point rest ───── @@ -363,6 +405,7 @@ _STEPS: list[tuple[str, str]] = [ WHERE haa2.house_id = m.keeper_id AND haa2.normalized_address = haa.normalized_address ) + RETURNING haa.house_id AS loser_id, to_jsonb(haa.*) AS row_snapshot """, ), ( @@ -372,6 +415,7 @@ _STEPS: list[tuple[str, str]] = [ SET house_id = m.keeper_id FROM _1772_dup_mapping m WHERE haa.house_id = m.loser_id + RETURNING m.loser_id, haa.id AS child_id """, ), # ── UNIQUE(house_id, source, room_count, prices_type, period, month_date) ── @@ -393,6 +437,7 @@ _STEPS: list[tuple[str, str]] = [ LEFT JOIN _1772_dup_mapping m ON m.loser_id = t2.house_id ) d WHERE t.id = d.id AND d.rn > 1 + RETURNING t.house_id AS loser_id, to_jsonb(t.*) AS row_snapshot """, ), ( @@ -402,6 +447,7 @@ _STEPS: list[tuple[str, str]] = [ SET house_id = m.keeper_id FROM _1772_dup_mapping m WHERE hpd.house_id = m.loser_id + RETURNING m.loser_id, hpd.id AS child_id """, ), # ── UNIQUE(house_id): one evaluation per keeper ─────────────────────────── @@ -419,6 +465,7 @@ _STEPS: list[tuple[str, str]] = [ LEFT JOIN _1772_dup_mapping m ON m.loser_id = t2.house_id ) d WHERE t.id = d.id AND d.rn > 1 + RETURNING t.house_id AS loser_id, to_jsonb(t.*) AS row_snapshot """, ), ( @@ -428,6 +475,7 @@ _STEPS: list[tuple[str, str]] = [ SET house_id = m.keeper_id FROM _1772_dup_mapping m WHERE hie.house_id = m.loser_id + RETURNING m.loser_id, hie.id AS child_id """, ), # ── UNIQUE(house_id, ext_item_id) ───────────────────────────────────────── @@ -445,6 +493,7 @@ _STEPS: list[tuple[str, str]] = [ LEFT JOIN _1772_dup_mapping m ON m.loser_id = t2.house_id ) d WHERE t.id = d.id AND d.rn > 1 + RETURNING t.house_id AS loser_id, to_jsonb(t.*) AS row_snapshot """, ), ( @@ -454,6 +503,7 @@ _STEPS: list[tuple[str, str]] = [ SET house_id = m.keeper_id FROM _1772_dup_mapping m WHERE hs.house_id = m.loser_id + RETURNING m.loser_id, hs.id AS child_id """, ), # ── UNIQUE(house_id, audit_batch) ───────────────────────────────────────── @@ -471,6 +521,7 @@ _STEPS: list[tuple[str, str]] = [ LEFT JOIN _1772_dup_mapping m ON m.loser_id = t2.house_id ) d WHERE t.id = d.id AND d.rn > 1 + RETURNING t.house_id AS loser_id, to_jsonb(t.*) AS row_snapshot """, ), ( @@ -480,10 +531,98 @@ _STEPS: list[tuple[str, str]] = [ SET house_id = m.keeper_id FROM _1772_dup_mapping m WHERE ama.house_id = m.loser_id + RETURNING m.loser_id, ama.id AS child_id """, ), ] +# ── MERGE JOURNAL (#2690) ───────────────────────────────────────────────────── +# +# Every child of houses(id) except `listings` references it through a column named house_id; +# listings uses house_id_fk. The undo function reads the column name back out of the journal +# key ("таблица.колонка"), so this mapping is what makes the reverse UPDATE possible. +_FK_COLUMN = {"listings": "house_id_fk"} + +# The (table, column) pairs the _STEPS pipeline actually handles, derived FROM the steps so the +# set cannot drift away from them. Compared against pg_catalog before every merge — see +# _assert_all_fk_children_handled. +_HANDLED_CHILDREN: frozenset[tuple[str, str]] = frozenset( + (tbl, _FK_COLUMN.get(tbl, "house_id")) for tbl in {label.split("(")[0] for label, _ in _STEPS} +) + +# Live FK children of houses(id), read from the catalog rather than trusted from a comment. +_FK_CHILDREN_SQL = text( + """ + SELECT CAST(CAST(c.conrelid AS regclass) AS text) AS child_table, + a.attname AS fk_column + FROM pg_constraint c + JOIN unnest(c.conkey) AS k(attnum) ON true + JOIN pg_attribute a ON a.attrelid = c.conrelid AND a.attnum = k.attnum + WHERE c.confrelid = CAST('houses' AS regclass) + AND c.contype = 'f' + """ +) + +# One journal row per loser, written from the mapping BEFORE anything is mutated — so loser_row +# is the row as it stood, and keeper_before precedes the identity carry-over. +_JOURNAL_INSERT_SQL = text( + """ + INSERT INTO house_merge_log ( + batch_id, run_id, initiator, merge_pass, cluster_key, geo_guard, distance_m, + norm_address, loser_id, keeper_id, loser_row, keeper_before + ) + SELECT + CAST(:batch_id AS uuid), + CAST(:run_id AS bigint), + CAST(:initiator AS text), + CAST(:merge_pass AS text), + m.cluster_key, + CAST(:geo_guard AS boolean), + m.distance_m, + m.norm_address, + m.loser_id, + m.keeper_id, + to_jsonb(l.*), + to_jsonb(k.*) + FROM _1772_dup_mapping m + JOIN houses l ON l.id = m.loser_id + JOIN houses k ON k.id = m.keeper_id + """ +) + +# Child bookkeeping lands after the steps ran — only then is it known which rows moved and which +# were destroyed by a UNIQUE collision. +_JOURNAL_CHILDREN_SQL = text( + """ + UPDATE house_merge_log + SET children_repointed = CAST(:children_repointed AS jsonb), + children_deleted = CAST(:children_deleted AS jsonb) + WHERE batch_id = CAST(:batch_id AS uuid) + AND loser_id = CAST(:loser_id AS bigint) + """ +) + + +def _assert_all_fk_children_handled(db: Session) -> None: + """Fail the merge if houses(id) gained an FK child the _STEPS pipeline does not handle. + + This is what makes the journal's promise true rather than merely documented. An unhandled + child is not a cosmetic gap: 9 of the 11 FKs are ON DELETE CASCADE, so `DELETE FROM houses` + would destroy its rows silently — no re-point step touches them, no RETURNING records them, + and the journal would claim a complete snapshot it does not have. Migration 133 already + broke on prod for exactly this (a missed child); there the failure was loud. Here it would + be silent, which is worse. Aborting the transaction costs one skipped merge cycle. + """ + live = {(r.child_table, r.fk_column) for r in db.execute(_FK_CHILDREN_SQL).all()} + unhandled = live - _HANDLED_CHILDREN + if unhandled: + raise RuntimeError( + "merge_duplicate_houses: houses(id) has FK children the merge does not handle: " + f"{sorted(unhandled)}. Their rows would be CASCADE-deleted without a journal entry. " + "Add a re-point step to _STEPS (and its RETURNING) before merging again." + ) + + # Delete the loser houses — all FK children are re-pointed or CASCADE by now. _DELETE_LOSERS_SQL = text( """ @@ -607,14 +746,20 @@ def _run_merge_pass( *, build_sql: Any, pass_label: str, + geo_guard: bool, + batch_id: str, + run_id: int | None, + initiator: str, result: DedupMergeResult, ) -> None: """Run ONE merge pass (fias- or canon-key) inside the caller's open transaction. - Builds a fresh loser→keeper mapping for this pass's cluster key, re-points every FK child - (UNIQUE-collision-safe), carries identity/enrichment onto the keeper, deletes the losers and - backfills sources/aliases. Accumulates counters onto `result`. NEVER commits/rolls back — the - caller owns the single transaction wrapping both passes. + Builds a fresh loser→keeper mapping for this pass's cluster key, writes the MERGE JOURNAL + (#2690), re-points every FK child (UNIQUE-collision-safe), carries identity/enrichment onto + the keeper, deletes the losers and backfills sources/aliases. Accumulates counters onto + `result`. NEVER commits/rolls back — the caller owns the single transaction wrapping both + passes, which is also what makes the journal atomic with the merge: there is no ordering in + which the rows vanish but the journal entry does not land (and dry_run rolls back both). """ # Fresh mapping for this pass. ON COMMIT DROP only fires at txn end, so drop the temp table # explicitly — the second pass must rebuild the same-named table within the one transaction. @@ -623,8 +768,8 @@ def _run_merge_pass( mapping = db.execute( text( - "SELECT loser_id, keeper_id, norm_address FROM _1772_dup_mapping " - "ORDER BY keeper_id, loser_id" + "SELECT loser_id, keeper_id, norm_address, cluster_key, distance_m " + "FROM _1772_dup_mapping ORDER BY keeper_id, loser_id" ) ).all() if not mapping: @@ -634,32 +779,73 @@ def _run_merge_pass( result.losers_deleted += len(mapping) result.clusters_merged += len({row.keeper_id for row in mapping}) - # Audit log: every loser→keeper move with its address, for traceability. + # JOURNAL, phase 1 — snapshot loser + keeper BEFORE any statement mutates them. + db.execute( + _JOURNAL_INSERT_SQL, + { + "batch_id": batch_id, + "run_id": run_id, + "initiator": initiator, + "merge_pass": pass_label, + "geo_guard": geo_guard, + }, + ) + + # Container logs rotate faster than a day (#2690), so this line is a convenience, not the + # record — house_merge_log is. Distance is logged too: it is the one number that says + # whether a merge would have survived the geo guard. for row in mapping: logger.info( - "merge_duplicate_houses: pass=%s merge loser_id=%d → keeper_id=%d address=%r", + "merge_duplicate_houses: pass=%s merge loser_id=%d → keeper_id=%d address=%r " + "distance_m=%s batch=%s", pass_label, row.loser_id, row.keeper_id, row.norm_address, + "n/a" if row.distance_m is None else f"{row.distance_m:.0f}", + batch_id, ) + # Per-loser child bookkeeping, collected from each step's RETURNING: survivors by id (the + # rows are intact, only their FK moved), destroyed rows by full snapshot (nothing else is + # left of them). + repointed: dict[int, dict[str, list[int]]] = {} + deleted: dict[int, dict[str, list[Any]]] = {} + for label, sql in _STEPS: - res = db.execute(text(sql)) - rowcount = res.rowcount or 0 - if label == "listings": - result.listings_repointed += rowcount - elif label.endswith("(collision-delete)") or label.endswith("(dedup)"): + rows = db.execute(text(sql)).all() + rowcount = len(rows) + table = label.split("(")[0] + if label.endswith("(collision-delete)") or label.endswith("(dedup)"): result.children_deleted += rowcount - elif label.endswith("(re-point)") or label in ( - "house_placement_history", - "house_reviews", - "house_reliability_checks", - "external_valuations", - ): - result.children_repointed += rowcount + for r in rows: + deleted.setdefault(r.loser_id, {}).setdefault(table, []).append(r.row_snapshot) + else: + key = f"{table}.{_FK_COLUMN.get(table, 'house_id')}" + for r in rows: + repointed.setdefault(r.loser_id, {}).setdefault(key, []).append(r.child_id) + if label == "listings": + result.listings_repointed += rowcount + else: + result.children_repointed += rowcount logger.debug("merge_duplicate_houses: pass=%s step=%s rows=%d", pass_label, label, rowcount) + # JOURNAL, phase 2 — attach the child bookkeeping to the rows written in phase 1. + touched = sorted(set(repointed) | set(deleted)) + if touched: + db.execute( + _JOURNAL_CHILDREN_SQL, + [ + { + "batch_id": batch_id, + "loser_id": loser_id, + "children_repointed": json.dumps(repointed.get(loser_id, {})), + "children_deleted": json.dumps(deleted.get(loser_id, {}), default=str), + } + for loser_id in touched + ], + ) + # Carry identity/enrichment onto the keeper BEFORE the losers vanish, then delete + backfill. db.execute(_CARRY_OVER_IDENTITY_SQL) db.execute(_DELETE_LOSERS_SQL) @@ -667,7 +853,13 @@ def _run_merge_pass( db.execute(_BACKFILL_ALIASES_SQL) -def merge_duplicate_houses(db: Session, *, dry_run: bool = False) -> dict[str, int]: +def merge_duplicate_houses( + db: Session, + *, + dry_run: bool = False, + run_id: int | None = None, + initiator: str = "manual", +) -> dict[str, int]: """Cluster houses by fias UUID, then by canonical address, merging dups onto one keeper. Re-implements migration 108's proven collision-safe pipeline as a RECURRING TWO-PASS job: @@ -680,16 +872,41 @@ def merge_duplicate_houses(db: Session, *, dry_run: bool = False) -> dict[str, i dry_run=True computes counts then ROLLS BACK (no writes). Idempotent: a clean table yields an empty mapping in each pass → every statement is a 0-row no-op. + Every deleted row is journaled to house_merge_log in the SAME transaction (#2690), so a + merge is reversible via house_merge_undo(batch_id); the batch_id is returned in the log line + and stored on every journal row of this call. + Returns the counter dict (DedupMergeResult.to_counters()). """ start = time.monotonic() result = DedupMergeResult(dry_run=dry_run) + batch_id = str(uuid.uuid4()) try: + # Refuse to merge at all if some FK child would be CASCADE-destroyed unjournaled. + _assert_all_fk_children_handled(db) # Pass 1: cluster by the ФИАС building UUID (runs first — most precise building identity). - _run_merge_pass(db, build_sql=_BUILD_MAPPING_SQL_FIAS, pass_label="fias", result=result) + _run_merge_pass( + db, + build_sql=_BUILD_MAPPING_SQL_FIAS, + pass_label="fias", + geo_guard=False, + batch_id=batch_id, + run_id=run_id, + initiator=initiator, + result=result, + ) # Pass 2: cluster by canonical address, with the cross-fias anti-over-merge guard. - _run_merge_pass(db, build_sql=_BUILD_MAPPING_SQL, pass_label="canon", result=result) + _run_merge_pass( + db, + build_sql=_BUILD_MAPPING_SQL, + pass_label="canon", + geo_guard=True, + batch_id=batch_id, + run_id=run_id, + initiator=initiator, + result=result, + ) if result.losers_deleted == 0: # Clean table — both passes empty. Roll back (we only opened temp tables). @@ -716,12 +933,15 @@ def merge_duplicate_houses(db: Session, *, dry_run: bool = False) -> dict[str, i db.commit() logger.info( "merge_duplicate_houses: COMMITTED clusters=%d losers=%d " - "listings_repointed=%d children_deleted=%d children_repointed=%d", + "listings_repointed=%d children_deleted=%d children_repointed=%d " + "batch_id=%s (undo: SELECT * FROM house_merge_undo('%s'))", result.clusters_merged, result.losers_deleted, result.listings_repointed, result.children_deleted, result.children_repointed, + batch_id, + batch_id, ) except Exception: logger.exception("merge_duplicate_houses: FAILED — rolling back") @@ -761,7 +981,7 @@ def run_house_dedup_merge(db: Session, *, run_id: int, params: dict) -> dict[str } try: runs_mod.update_heartbeat(db, run_id, counters) - counters = merge_duplicate_houses(db, dry_run=dry_run) + counters = merge_duplicate_houses(db, dry_run=dry_run, run_id=run_id, initiator="schedule") runs_mod.mark_done(db, run_id, counters) logger.info( "run_house_dedup_merge: run_id=%d DONE clusters=%d losers=%d dry_run=%s", diff --git a/tradein-mvp/backend/data/sql/230_house_merge_log.sql b/tradein-mvp/backend/data/sql/230_house_merge_log.sql new file mode 100644 index 00000000..bf55ec6b --- /dev/null +++ b/tradein-mvp/backend/data/sql/230_house_merge_log.sql @@ -0,0 +1,262 @@ +-- 230_house_merge_log.sql +-- Журнал слияний домов + обратная операция (#2690). +-- +-- WHY: +-- `house_dedup_merge` — НЕ спящая идея, а живой деструктивный проход: расписание +-- `house_dedup_merge` на проде enabled=true, dry_run=false, такт 7 дней. Шесть прогонов +-- с 2026-06-27 уже удалили 119 строк `houses` (счётчики losers_deleted в scrape_runs: +-- 2/39/31/9/6/32). Единственным следом «кто в кого» была строка `logger.info` в контейнере, +-- а логи ротируются быстрее суток. То есть **уже сегодня** нельзя назвать, какой дом в какой +-- свернули 1 августа, — не говоря о том, чтобы вернуть. +-- +-- Пока этого журнала нет, любой разговор о расширении ключа схлопывания (#2690, #1772) +-- ведётся без права на ошибку: единственный откат — restore всей БД на момент до прогона, +-- т.е. выброс недели сбора. Журнал снимает это условие: слияние становится обратимым, +-- и вопрос о ключе можно пересматривать, а не решать «навсегда». +-- +-- Правку НЕ следует читать как одобрение текущего ключа/победителя/гео-стража. Она к ним +-- НЕЙТРАЛЬНА: ни ключ, ни правило выбора победителя, ни гео-страж здесь не меняются. +-- Меняется только одно — теперь есть что откатить. +-- +-- WHAT (одна строка = один проигравший дом): +-- merge_pass / cluster_key / geo_guard / distance_m — ОСНОВАНИЕ слияния. Это не косметика: +-- ровно этих полей не хватило в #2690, чтобы ответить на вопрос «сколько слияний прошло +-- на расстояниях, которые гео-страж заблокировал бы» по ДАННЫМ, а не по ревью. distance_m +-- пишется всегда, даже когда страж для прохода выключен (fias-проход) — тогда он и есть +-- единственная запись о том, насколько далеко разъехались объединённые дома. +-- loser_row — ПОЛНЫЙ jsonb-снимок удаляемой строки (`to_jsonb(h.*)`, все 86 колонок). +-- Ссылка на удалённую строку бесполезна, поэтому хранится содержимое. Снимок целиком, +-- а не список полей: проверено, что `jsonb_populate_record(NULL::houses, loser_row)` +-- восстанавливает строку побайтово, включая PostGIS-geom (to_jsonb отдаёт её GeoJSON'ом, +-- populate_record разбирает обратно входной функцией типа). Побочная выгода: новая +-- колонка в `houses` попадает в снимок и в откат САМА, без правки этой миграции. +-- keeper_before — снимок ПОБЕДИТЕЛЯ до переноса метаданных. Нужен, потому что слияние не +-- только удаляет проигравшего: `_CARRY_OVER_IDENTITY_SQL` дозаполняет победителю NULL-поля +-- идентичности (fias/кадастр/ГАР/DaData) значениями проигравшего. Без этого снимка откат +-- вернул бы дом, но оставил бы его ФИАС на победителе — и следующий же fias-проход слил +-- бы их обратно. +-- children_repointed — {"таблица.колонка": [id, ...]}. Дочерние строки ПЕРЕЖИЛИ слияние, +-- у них сменилась только ссылка, поэтому хранятся id, а не содержимое (иначе одни +-- listings с их raw-payload'ом дали бы ~7 КБ на строку вместо ~8 байт на id). +-- children_deleted — {"таблица": [{строка целиком}, ...]}. Дочерние строки, которые проход +-- УДАЛИЛ из-за коллизии по UNIQUE. Их содержимое уничтожено, id недостаточно — только +-- полный снимок. Таких таблиц шесть (см. _STEPS), строки мелкие. +-- batch_id — один вызов merge_duplicate_houses() (оба прохода). Единица отката. +-- run_id / initiator — кто инициировал: scrape_runs.id для расписания, NULL для ручного. +-- +-- НАМЕРЕННО БЕЗ ВНЕШНИХ КЛЮЧЕЙ на houses(id) и scrape_runs(id): +-- журнал обязан ПЕРЕЖИВАТЬ строки, которые описывает. loser_id указывает на заведомо +-- удалённый дом. keeper_id — на дом, который сам может быть слит следующим прогоном; FK +-- с CASCADE стёр бы историю ровно тогда, когда она нужнее всего, а FK без CASCADE +-- заблокировал бы слияние. То же с run_id: чистка scrape_runs не должна трогать журнал. +-- +-- ОБЪЁМ (замерено на проде 2026-08-06): +-- 9 571 дом, средняя строка houses в jsonb 2 581 Б. Строка журнала ≈ loser_row 2.5 КБ + +-- keeper_before 2.5 КБ + списки id (в среднем 27.9 дочерних строк на дом × ~8 Б) ≈ 5.3 КБ. +-- Наблюдаемый темп — 20 слияний в неделю (119 за 6 прогонов) ≈ 106 КБ/нед ≈ 5.5 МБ/год. +-- Ближайший прогон (замер тем же выражением, что и код): 93 проигравших ≈ 0.5 МБ. +-- Абсолютный потолок, если схлопнуть вообще все дома: 9 571 × 5.3 КБ ≈ 50 МБ против 23 МБ +-- самой таблицы houses. +-- +-- RETENTION: НЕ НУЖЕН, сознательно. Потолок роста — двузначные мегабайты, то есть дешевле +-- любой процедуры чистки; а журнал слияний — это ровно то, что удалять не хочется: его +-- ценность в том, что он отвечает на вопрос «что было год назад», когда логов давно нет. +-- Если объём когда-нибудь станет проблемой, удалять надо не строки, а тяжёлые снимки +-- (loser_row/keeper_before → NULL) у записей старше N лет, сохранив соответствие +-- loser→keeper: оно весит байты и именно оно нужно дольше всего. +-- +-- Dependencies: 002_core_tables.sql (houses), 135_scrape_schedules_seed_house_dedup_merge.sql +-- Пишется в ТОЙ ЖЕ транзакции, что и слияние (см. house_dedup_merge._run_merge_pass) — +-- разрыв «слияние прошло, запись не легла» невозможен по построению; dry_run откатывает и то, +-- и другое вместе. + +BEGIN; + +CREATE TABLE IF NOT EXISTS house_merge_log ( + id bigserial PRIMARY KEY, + merged_at timestamptz NOT NULL DEFAULT now(), + batch_id uuid NOT NULL, + run_id bigint, + initiator text NOT NULL, + merge_pass text NOT NULL, + cluster_key text NOT NULL, + geo_guard boolean NOT NULL, + distance_m double precision, + norm_address text, + loser_id bigint NOT NULL, + keeper_id bigint NOT NULL, + loser_row jsonb NOT NULL, + keeper_before jsonb NOT NULL, + children_repointed jsonb NOT NULL DEFAULT '{}'::jsonb, + children_deleted jsonb NOT NULL DEFAULT '{}'::jsonb +); + +CREATE INDEX IF NOT EXISTS idx_house_merge_log_loser ON house_merge_log (loser_id); +CREATE INDEX IF NOT EXISTS idx_house_merge_log_keeper ON house_merge_log (keeper_id); +CREATE INDEX IF NOT EXISTS idx_house_merge_log_batch ON house_merge_log (batch_id); + +COMMENT ON TABLE house_merge_log IS + 'Журнал слияний домов (#2690): одна строка = один проигравший дом, удалённый проходом ' + 'house_dedup_merge. Пишется в ТОЙ ЖЕ транзакции, что и слияние. Содержит полный снимок ' + 'удалённой строки и перечень перенесённых/удалённых дочерних строк — достаточно, чтобы ' + 'назвать поимённо, что во что свернули, и вернуть обратно (house_merge_undo). Намеренно ' + 'БЕЗ FK на houses/scrape_runs: журнал переживает строки, которые описывает. Retention нет.'; + +COMMENT ON COLUMN house_merge_log.cluster_key IS + 'ЗНАЧЕНИЕ ключа, по которому дома попали в один кластер («addr:вайнера66» / «fias:»), ' + 'а не имя ключа — по нему видно, какое именно совпадение сработало.'; +COMMENT ON COLUMN house_merge_log.geo_guard IS + 'Был ли для этого прохода включён гео-страж 250 м. false = слияние разрешено БЕЗ проверки ' + 'близости; вместе с distance_m это и есть аудит основания (#2690).'; +COMMENT ON COLUMN house_merge_log.distance_m IS + 'ST_DistanceSphere между победителем и проигравшим на момент слияния; NULL = у одной из ' + 'сторон не было geom. Пишется ВСЕГДА, в том числе когда гео-страж выключен.'; +COMMENT ON COLUMN house_merge_log.loser_row IS + 'to_jsonb() удалённой строки houses целиком. Восстановление: ' + 'INSERT INTO houses SELECT r.* FROM jsonb_populate_record(NULL::houses, loser_row) r.'; +COMMENT ON COLUMN house_merge_log.keeper_before IS + 'Снимок победителя ДО переноса метаданных с проигравшего (COALESCE-дозаполнение полей ' + 'идентичности). Без него откат вернул бы дом, но оставил его ФИАС/кадастр на победителе.'; +COMMENT ON COLUMN house_merge_log.children_repointed IS + '{"таблица.колонка": [id, ...]} — дочерние строки, у которых слияние сменило ссылку ' + 'loser→keeper. Строки целы, поэтому хранятся id: откат возвращает ссылку обратно.'; +COMMENT ON COLUMN house_merge_log.children_deleted IS + '{"таблица": [{строка целиком}, ...]} — дочерние строки, УДАЛЁННЫЕ проходом из-за коллизии ' + 'по UNIQUE с победителем. Содержимое уничтожено, поэтому хранится снимок, а не id.'; + +-- ── Обратная операция ──────────────────────────────────────────────────────── +-- +-- Откат одного батча (или его части) по журналу. Транзакционен: вызывающий сам решает +-- COMMIT/ROLLBACK, увидев отчёт. Возвращает СТРОКУ НА КАЖДУЮ запись журнала со статусом — +-- в том числе «не смог», потому что молчаливо-успешный откат хуже отсутствующего. +-- +-- Порядок внутри одной записи важен: сначала воскресить дом (на него ссылаются дети), потом +-- вернуть ссылки детей, потом вернуть удалённых детей, потом снять перенос метаданных с +-- победителя. Записи батча обходятся в обратном порядке (id DESC) — если дом A слили в B, +-- а B потом в C, разматывать надо с конца. +-- +-- ИЗВЕСТНЫЕ ГРАНИЦЫ (сознательные, отражены в статусе): +-- * дочерняя строка, удалённая по коллизии, может не вернуться: место в UNIQUE-ключе занято +-- строкой победителя. ON CONFLICT DO NOTHING + счётчик в статусе, а не тихая потеря; +-- * backfill-строки house_sources/house_address_aliases, которые проход дописал победителю, +-- НЕ удаляются: они собраны из собственных полей победителя и остались бы верны и без +-- слияния; +-- * если id проигравшего уже занят — запись пропускается со статусом, откат не гадает. +CREATE OR REPLACE FUNCTION house_merge_undo( + p_batch uuid, + p_only_losers bigint[] DEFAULT NULL +) +RETURNS TABLE ( + out_log_id bigint, + out_loser_id bigint, + out_keeper_id bigint, + out_status text +) +LANGUAGE plpgsql +AS $$ +DECLARE + rec record; + v_table text; + v_column text; + v_ids bigint[]; + v_rows jsonb; + v_field text; + v_repointed int; + v_restored int; + v_lost int; + v_n int; + -- Список полей ДОЛЖЕН совпадать с SET в house_dedup_merge._CARRY_OVER_IDENTITY_SQL; + -- за расхождением следит тест test_undo_carryover_fields_match_merge_carryover. + c_carry_fields constant text[] := ARRAY[ + 'house_fias_id', 'cadastral_number', 'gar_house_guid', 'gar_flat_count', + 'gar_matched_at', 'gar_match_method', 'dadata_qc_geo', 'dadata_qc_house', + 'dadata_enriched_at' + ]; +BEGIN + FOR rec IN + SELECT * + FROM house_merge_log l + WHERE l.batch_id = p_batch + AND (p_only_losers IS NULL OR l.loser_id = ANY (p_only_losers)) + ORDER BY l.id DESC + LOOP + out_log_id := rec.id; + out_loser_id := rec.loser_id; + out_keeper_id := rec.keeper_id; + + IF EXISTS (SELECT 1 FROM houses h WHERE h.id = rec.loser_id) THEN + out_status := 'skipped: houses.id ' || rec.loser_id || ' занят — уже откачено?'; + RETURN NEXT; + CONTINUE; + END IF; + + -- 1. Воскресить проигравшего целиком из снимка (все колонки, включая geom). + INSERT INTO houses + SELECT r.* FROM jsonb_populate_record(NULL::houses, rec.loser_row) r; + + -- 2. Вернуть ссылки уцелевших детей. Условие «сейчас указывает на победителя» + -- защищает от затирания строк, которые после слияния перепривязали чем-то ещё. + v_repointed := 0; + FOR v_table, v_column, v_ids IN + SELECT split_part(e.key, '.', 1), + split_part(e.key, '.', 2), + ARRAY(SELECT jsonb_array_elements_text(e.value)::bigint) + FROM jsonb_each(rec.children_repointed) AS e + LOOP + EXECUTE format( + 'UPDATE %I SET %I = $1 WHERE id = ANY ($2) AND %I = $3', + v_table, v_column, v_column + ) USING rec.loser_id, v_ids, rec.keeper_id; + GET DIAGNOSTICS v_n = ROW_COUNT; + v_repointed := v_repointed + v_n; + END LOOP; + + -- 3. Вернуть детей, удалённых по коллизии UNIQUE. Место могло остаться занятым + -- строкой победителя — тогда DO NOTHING, и это попадёт в отчёт как «не вернулось». + v_restored := 0; + v_lost := 0; + FOR v_table, v_rows IN + SELECT e.key, e.value FROM jsonb_each(rec.children_deleted) AS e + LOOP + EXECUTE format( + 'INSERT INTO %I SELECT r.* FROM jsonb_array_elements($1) AS el, ' + 'LATERAL jsonb_populate_record(NULL::%I, el) r ON CONFLICT DO NOTHING', + v_table, v_table + ) USING v_rows; + GET DIAGNOSTICS v_n = ROW_COUNT; + v_restored := v_restored + v_n; + v_lost := v_lost + (jsonb_array_length(v_rows) - v_n); + END LOOP; + + -- 4. Снять перенос метаданных с победителя. Только там, где до слияния было NULL И + -- текущее значение всё ещё РОВНО то, что принёс этот проигравший: если поле успел + -- заполнить загрузчик (или донором был другой проигравший кластера) — не трогаем. + -- Сравнение в jsonb-пространстве, чтобы один цикл покрыл text/int/timestamptz. + FOREACH v_field IN ARRAY c_carry_fields LOOP + IF rec.keeper_before ->> v_field IS NULL THEN + EXECUTE format( + 'UPDATE houses SET %I = NULL WHERE id = $1 AND to_jsonb(%I) = $2', + v_field, v_field + ) USING rec.keeper_id, rec.loser_row -> v_field; + END IF; + END LOOP; + + out_status := format( + 'restored: дом %s вернулся, ссылок возвращено %s, дочерних строк восстановлено %s' + || CASE WHEN v_lost > 0 THEN ', НЕ ВЕРНУЛОСЬ ' || v_lost || ' (место занято)' + ELSE '' END, + rec.loser_id, v_repointed, v_restored + ); + RETURN NEXT; + END LOOP; +END; +$$; + +COMMENT ON FUNCTION house_merge_undo(uuid, bigint[]) IS + 'Откат слияния домов по журналу house_merge_log (#2690). Аргументы: batch_id (единица ' + 'отката = один вызов merge_duplicate_houses) и опциональный список loser_id для частичного ' + 'отката. Возвращает строку-статус на КАЖДУЮ запись журнала, включая неудачные. ' + 'Транзакции не открывает и не закрывает — вызывающий смотрит отчёт и решает COMMIT/ROLLBACK: ' + ' BEGIN; SELECT * FROM house_merge_undo(''''); -- прочитать статусы -- COMMIT;'; + +COMMIT; diff --git a/tradein-mvp/backend/tests/test_house_dedup_merge.py b/tradein-mvp/backend/tests/test_house_dedup_merge.py index ef6494e0..63db7822 100644 --- a/tradein-mvp/backend/tests/test_house_dedup_merge.py +++ b/tradein-mvp/backend/tests/test_house_dedup_merge.py @@ -284,9 +284,18 @@ def test_fias_pass_drops_geo_guard_canon_pass_keeps_it() -> None: assert "keeper_geom IS NOT NULL" in canon assert "loser_geom IS NOT NULL" in canon # fias pass drops the distance guard AND the NULL-geom exclusions entirely. - assert "ST_DistanceSphere" not in fias - assert "loser_geom IS NOT NULL" not in fias - assert "keeper_geom IS NOT NULL" not in fias + # + # Asserted on the guard PREDICATE, not on the bare function name: since #2690 the mapping also + # MEASURES the keeper↔loser distance into `distance_m` for the merge journal, on both passes. + # Measuring is the opposite of guarding — the fias pass is precisely where nothing else records + # how far apart the merged rows were — so the name alone can no longer stand in for the guard. + assert "ST_DistanceSphere(loser_geom, keeper_geom) <= 250" not in fias + guard = ( + "AND keeper_geom IS NOT NULL AND loser_geom IS NOT NULL " + "AND ST_DistanceSphere(loser_geom, keeper_geom) <= 250" + ) + assert guard in canon + assert guard not in fias # the cross-fias anti-over-merge guard is untouched in the canon pass. assert "lower(loser_fias) <> lower(keeper_fias)" in canon @@ -294,12 +303,15 @@ def test_fias_pass_drops_geo_guard_canon_pass_keeps_it() -> None: def test_mapping_sql_geo_guard_param_toggles_only_distance_filter() -> None: """_mapping_sql(apply_geo_guard=...) toggles ONLY the 250 m distance filter; the cross-fias guard is emitted regardless, and the default is True (canon-safe).""" + guard = "ST_DistanceSphere(loser_geom, keeper_geom) <= 250" with_guard = _flat(hdm._mapping_sql(hdm._CANON_KEY_EXPR, apply_geo_guard=True)) without_guard = _flat(hdm._mapping_sql(hdm._CANON_KEY_EXPR, apply_geo_guard=False)) - assert "ST_DistanceSphere" in with_guard - assert "ST_DistanceSphere" not in without_guard + assert guard in with_guard + assert guard not in without_guard # default = True (the canon pass must never lose its guard by omission). - assert "ST_DistanceSphere" in _flat(hdm._mapping_sql(hdm._CANON_KEY_EXPR)) + assert guard in _flat(hdm._mapping_sql(hdm._CANON_KEY_EXPR)) + # ...while the journal's distance MEASUREMENT is emitted either way (#2690). + assert "AS distance_m" in with_guard and "AS distance_m" in without_guard # cross-fias guard present in BOTH renderings (independent of the geo guard). assert "lower(loser_fias) <> lower(keeper_fias)" in with_guard assert "lower(loser_fias) <> lower(keeper_fias)" in without_guard @@ -327,7 +339,9 @@ def test_both_passes_share_one_pipeline_no_copy_paste() -> None: assert token in canon and token in fias # the 250 m distance guard is CANON-ONLY (#2187) — fias identity outranks proximity. assert "ST_DistanceSphere(loser_geom, keeper_geom) <= 250" in canon - assert "ST_DistanceSphere" not in fias + assert "ST_DistanceSphere(loser_geom, keeper_geom) <= 250" not in fias + # ...but the journal's distance MEASUREMENT is on both — measuring is not guarding. + assert "AS distance_m" in canon and "AS distance_m" in fias def test_cross_fias_guard_blocks_slash_collapse_over_merge() -> None: @@ -433,28 +447,62 @@ class _FakeResult: class _Row: - def __init__(self, loser_id: int, keeper_id: int, norm_address: str): + def __init__( + self, + loser_id: int, + keeper_id: int, + norm_address: str, + cluster_key: str = "addr:тест", + distance_m: float | None = 12.0, + ): self.loser_id = loser_id self.keeper_id = keeper_id self.norm_address = norm_address + # journal grounds (#2690): which key value fired, and how far apart the rows were. + self.cluster_key = cluster_key + self.distance_m = distance_m + + +class _ChildRow: + """What a step's RETURNING yields: an id for a survivor, a snapshot for a destroyed row.""" + + def __init__(self, loser_id: int, child_id: int = 1): + self.loser_id = loser_id + self.child_id = child_id + self.row_snapshot = {"id": child_id, "house_id": loser_id} + + +class _FKChild: + def __init__(self, child_table: str, fk_column: str): + self.child_table = child_table + self.fk_column = fk_column class _FakeDB: """Session stand-in: build-mapping + a scripted SELECT result, then per-step rowcounts.""" - def __init__(self, mapping_rows: list[_Row], step_rowcount: int = 1): + def __init__( + self, + mapping_rows: list[_Row], + step_rowcount: int = 1, + fk_children: dict[str, str] | None = None, + ): self._mapping_rows = mapping_rows self._step_rowcount = step_rowcount self._mapping_served = False + # The catalog the FK-child guard reads; defaults to the real live set. + self._fk_children = _FK_CHILDREN if fk_children is None else fk_children self.commits = 0 self.rollbacks = 0 self.executed: list[str] = [] - def execute(self, clause: Any, params: dict | None = None) -> _FakeResult: + def execute(self, clause: Any, params: Any = None) -> _FakeResult: sql = str(getattr(clause, "text", clause)) self.executed.append(sql) if "CREATE TEMP TABLE" in sql: return _FakeResult() + if "FROM pg_constraint" in sql: + return _FakeResult(rows=[_FKChild(t, c) for t, c in self._fk_children.items()]) if "SELECT loser_id, keeper_id, norm_address" in sql: # The service now runs TWO passes (fias, then canon). Model «fias pass found the # duplicates, canon pass is clean»: serve the scripted mapping once, empty afterwards. @@ -462,7 +510,13 @@ class _FakeDB: return _FakeResult(rows=[]) self._mapping_served = True return _FakeResult(rows=list(self._mapping_rows)) - # any UPDATE/DELETE/INSERT step (incl. DROP TABLE, carry-over, delete, backfill) + # Steps now RETURN the rows they touched (journal, #2690) — one per scripted rowcount, + # attributed to the first loser so the per-loser bookkeeping has something to bucket. + if "RETURNING" in sql: + loser = self._mapping_rows[0].loser_id if self._mapping_rows else 0 + rows = [_ChildRow(loser, child_id=i + 1) for i in range(self._step_rowcount)] + return _FakeResult(rowcount=self._step_rowcount, rows=rows) + # any other UPDATE/DELETE/INSERT (DROP TABLE, journal, carry-over, delete, backfill) return _FakeResult(rowcount=self._step_rowcount) def commit(self) -> None: @@ -524,7 +578,11 @@ def test_run_wrapper_marks_done_with_counters(monkeypatch: pytest.MonkeyPatch) - monkeypatch.setattr( hdm, "merge_duplicate_houses", - lambda _db, dry_run=False: {"clusters_merged": 3, "losers_deleted": 5, "dry_run": 0}, + lambda _db, dry_run=False, run_id=None, initiator="manual": { + "clusters_merged": 3, + "losers_deleted": 5, + "dry_run": 0, + }, ) out = hdm.run_house_dedup_merge(object(), run_id=42, params={"dry_run": False}) # type: ignore[arg-type] @@ -541,13 +599,20 @@ def test_run_wrapper_passes_dry_run_param(monkeypatch: pytest.MonkeyPatch) -> No monkeypatch.setattr(runs_mod, "mark_done", lambda *a, **k: None) monkeypatch.setattr(runs_mod, "mark_failed", lambda *a, **k: None) - def _fake_merge(_db: Any, dry_run: bool = False) -> dict[str, int]: + def _fake_merge( + _db: Any, dry_run: bool = False, run_id: int | None = None, initiator: str = "manual" + ) -> dict[str, int]: captured["dry_run"] = dry_run + captured["run_id"] = run_id + captured["initiator"] = initiator return {"dry_run": int(dry_run)} monkeypatch.setattr(hdm, "merge_duplicate_houses", _fake_merge) hdm.run_house_dedup_merge(object(), run_id=1, params={"dry_run": True}) # type: ignore[arg-type] assert captured["dry_run"] is True + # the journal must be able to say WHICH run did it, and that it was not a human (#2690) + assert captured["run_id"] == 1 + assert captured["initiator"] == "schedule" def test_run_wrapper_marks_failed_on_error(monkeypatch: pytest.MonkeyPatch) -> None: @@ -562,7 +627,9 @@ def test_run_wrapper_marks_failed_on_error(monkeypatch: pytest.MonkeyPatch) -> N lambda _db, run_id, err, counters: failed.update(run_id=run_id, err=err), ) - def _boom(_db: Any, dry_run: bool = False) -> dict[str, int]: + def _boom( + _db: Any, dry_run: bool = False, run_id: int | None = None, initiator: str = "manual" + ) -> dict[str, int]: raise RuntimeError("merge exploded") monkeypatch.setattr(hdm, "merge_duplicate_houses", _boom) @@ -642,12 +709,16 @@ def test_real_merge_repoints_dedups_deletes_and_is_idempotent() -> None: db = _live_session() assert db is not None try: - # Two houses at the SAME address. Keeper (geom present) should win. + # Two houses at the SAME address, ~10 m apart (the #2187 canon geo guard needs geom + # on BOTH sides). Keeper = min(id) once geom and listing counts tie. db.execute( _t( - "INSERT INTO houses (id, source, ext_house_id, address, lat, lon) VALUES " - "(900001, 'avito', 'EXT-KEEP', 'тестдом 1772, 1', 56.84, 60.60)," - "(900002, 'cian', 'EXT-LOSE', 'тестдом 1772, 1', NULL, NULL)" + # url is NOT NULL in houses (002_core_tables); nothing here asserts on it, + # so 'u' is a placeholder. These live-DB fixtures self-skip in CI, which is + # how they silently drifted out of sync with the schema in the first place. + "INSERT INTO houses (id, source, ext_house_id, url, address, lat, lon) VALUES " + "(900001, 'avito', 'EXT-KEEP','u', 'тестдом 1772, 1', 56.84, 60.60)," + "(900002, 'cian', 'EXT-LOSE','u', 'тестдом 1772, 1', 56.84009, 60.60)" ) ) # listings pointing at BOTH (the loser's must be re-pointed). source_url, dedup_hash, @@ -755,6 +826,9 @@ def test_real_merge_repoints_dedups_deletes_and_is_idempotent() -> None: db.execute( _t("DELETE FROM house_address_aliases WHERE normalized_address = 'тестдом 1772, 1'") ) + # journal rows have no FK and are never cascaded away — sweep them explicitly, + # or a re-run accumulates them (all live fixtures live in the 9000xx id range). + db.execute(_t("DELETE FROM house_merge_log WHERE loser_id BETWEEN 900000 AND 900299")) db.execute(_t("DELETE FROM houses WHERE id IN (900001,900002)")) db.commit() db.close() @@ -781,16 +855,16 @@ def test_real_canon_clusterkey_and_geo_guard_merge_semantics() -> None: try: db.execute( _t( - "INSERT INTO houses (id, source, ext_house_id, address, lat, lon) VALUES " + "INSERT INTO houses (id, source, ext_house_id, url, address, lat, lon) VALUES " # A — ул/улица spelling variants of the SAME building, ~10 m apart → MERGE - "(900010, 'avito', 'EXT-T-VK', 'улица Тестовая1772, 66', 56.84000, 60.60000)," - "(900011, 'cian', 'EXT-T-VL', 'ул. Тестовая1772, 66', 56.84009, 60.60000)," + "(900010, 'avito', 'EXT-T-VK','u', 'улица Тестовая1772, 66', 56.84000, 60.60000)," + "(900011, 'cian', 'EXT-T-VL','u', 'ул. Тестовая1772, 66', 56.84009, 60.60000)," # B — same canon (ленина-like) but ~5 km apart → geo guard BLOCKS the merge - "(900012, 'avito', 'EXT-T-L1', 'улица Тестовая1772, 5', 56.84000, 60.60000)," - "(900013, 'cian', 'EXT-T-L2', 'улица Тестовая1772, 5', 56.88500, 60.60000)," + "(900012, 'avito', 'EXT-T-L1','u', 'улица Тестовая1772, 5', 56.84000, 60.60000)," + "(900013, 'cian', 'EXT-T-L2','u', 'улица Тестовая1772, 5', 56.88500, 60.60000)," # C — different корпус → different canon, ~10 m apart → NOT merged - "(900014, 'avito', 'EXT-T-M2', 'Тестовая1772, 34к2', 56.84000, 60.60000)," - "(900015, 'cian', 'EXT-T-M4', 'Тестовая1772, 34к4', 56.84009, 60.60000)" + "(900014, 'avito', 'EXT-T-M2','u', 'Тестовая1772, 34к2', 56.84000, 60.60000)," + "(900015, 'cian', 'EXT-T-M4','u', 'Тестовая1772, 34к4', 56.84009, 60.60000)" ) ) db.execute( @@ -840,7 +914,7 @@ def test_real_canon_clusterkey_and_geo_guard_merge_semantics() -> None: db.execute( _t( "DELETE FROM house_sources WHERE ext_id IN " - "('EXT-T-VK','EXT-T-VL','EXT-T-L1','EXT-T-L2','EXT-T-M2','EXT-T-M4')" + "('EXT-T-VK','u','EXT-T-VL','u','EXT-T-L1','u','EXT-T-L2','u','EXT-T-M2','u','EXT-T-M4')" ) ) db.execute( @@ -850,6 +924,9 @@ def test_real_canon_clusterkey_and_geo_guard_merge_semantics() -> None: "'тестовая1772, 34к2','тестовая1772, 34к4')" ) ) + # journal rows have no FK and are never cascaded away — sweep them explicitly, + # or a re-run accumulates them (all live fixtures live in the 9000xx id range). + db.execute(_t("DELETE FROM house_merge_log WHERE loser_id BETWEEN 900000 AND 900299")) db.execute(_t("DELETE FROM houses WHERE id BETWEEN 900010 AND 900015")) db.commit() db.close() @@ -877,21 +954,21 @@ def test_real_fias_pass_cross_guard_and_identity_carryover() -> None: db.execute( _t( "INSERT INTO houses " - "(id, source, ext_house_id, address, lat, lon, house_fias_id, gar_house_guid, " + "(id, source, ext_house_id, url, address, lat, lon, house_fias_id, gar_house_guid, " " dadata_enriched_at) VALUES " # A — same fias, different canon (different streets), ~10 m apart → FIAS-pass merge - "(900020,'avito','EXT-F-K','ФиасОдин1772, 10', 56.84000,60.60000," + "(900020,'avito','EXT-F-K','u','ФиасОдин1772, 10', 56.84000,60.60000," " 'F-SAME-1772',NULL,NULL)," - "(900021,'cian', 'EXT-F-L','СовсемДругая1772, 77',56.84009,60.60000," + "(900021,'cian', 'EXT-F-L','u','СовсемДругая1772, 77',56.84009,60.60000," " 'F-SAME-1772',NULL,NULL)," # B — same canon (slash-collapse), DIFFERENT fias → cross-fias guard BLOCKS - "(900022,'avito','EXT-B-1','Клара1772, 32',56.84000,60.60000," + "(900022,'avito','EXT-B-1','u','Клара1772, 32',56.84000,60.60000," " 'F-B1-1772',NULL,NULL)," - "(900023,'cian', 'EXT-B-2','Клара1772, 3/2',56.84009,60.60000," + "(900023,'cian', 'EXT-B-2','u','Клара1772, 3/2',56.84009,60.60000," " 'F-B2-1772',NULL,NULL)," # C — same canon, fias only on the loser → canon-pass merge + carry-over - "(900024,'avito','EXT-C-K','Донбасс1772, 8',56.84000,60.60000,NULL,NULL,NULL)," - "(900025,'cian', 'EXT-C-L','Донбасс1772, 8',56.84009,60.60000," + "(900024,'avito','EXT-C-K','u','Донбасс1772, 8',56.84000,60.60000,NULL,NULL,NULL)," + "(900025,'cian', 'EXT-C-L','u','Донбасс1772, 8',56.84009,60.60000," " 'F-CARRY-1772','G-CARRY-1772',NOW())" ) ) @@ -944,7 +1021,7 @@ def test_real_fias_pass_cross_guard_and_identity_carryover() -> None: db.execute( _t( "DELETE FROM house_sources WHERE ext_id IN " - "('EXT-F-K','EXT-F-L','EXT-B-1','EXT-B-2','EXT-C-K','EXT-C-L')" + "('EXT-F-K','u','EXT-F-L','u','EXT-B-1','u','EXT-B-2','u','EXT-C-K','u','EXT-C-L')" ) ) db.execute( @@ -954,6 +1031,9 @@ def test_real_fias_pass_cross_guard_and_identity_carryover() -> None: "'донбасс1772, 8')" ) ) + # journal rows have no FK and are never cascaded away — sweep them explicitly, + # or a re-run accumulates them (all live fixtures live in the 9000xx id range). + db.execute(_t("DELETE FROM house_merge_log WHERE loser_id BETWEEN 900000 AND 900299")) db.execute(_t("DELETE FROM houses WHERE id BETWEEN 900020 AND 900025")) db.commit() db.close() @@ -982,16 +1062,16 @@ def test_real_fias_pass_ignores_geo_guard() -> None: db.execute( _t( "INSERT INTO houses " - "(id, source, ext_house_id, address, lat, lon, house_fias_id) VALUES " + "(id, source, ext_house_id, url, address, lat, lon, house_fias_id) VALUES " # A — same fias, loser NULL geom → fias pass merges despite the missing coordinate - "(900030,'avito','EXT-2187-A-K','ФиасГеоA2187, 1', 56.84000,60.60000,'F-A-2187')," - "(900031,'cian', 'EXT-2187-A-L','ФиасГеоAL2187, 2',NULL, NULL, 'F-A-2187')," + "(900030,'avito','EXT-2187-A-K','u','ФиасГеоA2187, 1',56.84,60.6,'F-A-2187')," + "(900031,'cian', 'EXT-2187-A-L','u','ФиасГеоAL2187, 2',NULL,NULL,'F-A-2187')," # B — same fias, ~5 km apart (>250 m) → fias pass merges despite the distance - "(900032,'avito','EXT-2187-B-K','ФиасГеоB2187, 3', 56.84000,60.60000,'F-B-2187')," - "(900033,'cian', 'EXT-2187-B-L','ФиасГеоBL2187, 4',56.88500,60.60000,'F-B-2187')," + "(900032,'avito','EXT-2187-B-K','u','ФиасГеоB2187, 3',56.84,60.6,'F-B-2187')," + "(900033,'cian', 'EXT-2187-B-L','u','ФиасГеоBL2187, 4',56.885,60.6,'F-B-2187')," # C — same canon, NO fias, ~5 km apart → canon pass STILL blocks (guard unchanged) - "(900034,'avito','EXT-2187-C-1','КанонГео2187, 5', 56.84000,60.60000,NULL)," - "(900035,'cian', 'EXT-2187-C-2','КанонГео2187, 5', 56.88500,60.60000,NULL)" + "(900034,'avito','EXT-2187-C-1','u','КанонГео2187, 5', 56.84000,60.60000,NULL)," + "(900035,'cian', 'EXT-2187-C-2','u','КанонГео2187, 5', 56.88500,60.60000,NULL)" ) ) # A loser gets a listing so we prove the re-point still fires with a NULL-geom loser. @@ -1028,6 +1108,199 @@ def test_real_fias_pass_ignores_geo_guard() -> None: db.execute(_t("DELETE FROM listings WHERE id = 910031")) db.execute(_t("DELETE FROM house_sources WHERE house_id BETWEEN 900030 AND 900035")) db.execute(_t("DELETE FROM house_address_aliases WHERE house_id BETWEEN 900030 AND 900035")) + # journal rows have no FK and are never cascaded away — sweep them explicitly, + # or a re-run accumulates them (all live fixtures live in the 9000xx id range). + db.execute(_t("DELETE FROM house_merge_log WHERE loser_id BETWEEN 900000 AND 900299")) db.execute(_t("DELETE FROM houses WHERE id BETWEEN 900030 AND 900035")) db.commit() db.close() + + +# ── Merge journal: reversibility (#2690) ────────────────────────────────────── + + +def test_undo_carryover_fields_match_merge_carryover() -> None: + """Static drift guard: migration 230's undo must un-set EXACTLY the fields the merge carries. + + The undo NULLs the keeper's identity fields that the merge COALESCE-filled from a loser. + If _CARRY_OVER_IDENTITY_SQL ever gains a field and the migration's array does not, the undo + silently leaves that field on the keeper — the restored loser and the keeper would then both + claim the same ФИАС, and the next fias pass would merge them straight back. + """ + migration = (_SQL_DIR / "230_house_merge_log.sql").read_text(encoding="utf-8") + carried = set(re.findall(r"^\s+(\w+)\s*=\s*COALESCE\(k\.", _CARRY_SQL, re.M)) + # slice the ARRAY[...] literal itself — the declaration's own `text[]` also holds a «]» + block = migration[migration.index("c_carry_fields") :] + undone = set(re.findall(r"'(\w+)'", block[block.index("ARRAY[") : block.index("];")])) + assert carried, "could not parse carried fields out of _CARRY_OVER_IDENTITY_SQL" + assert carried == undone, f"carry-over/undo field drift: merge={carried} undo={undone}" + + +def test_journal_written_in_the_same_transaction_as_the_merge() -> None: + """The journal INSERT must sit between mapping and delete, with no commit in between. + + Requirement from #2690: a merge that commits without its journal row is exactly the failure + the journal exists to prevent, so the two must share one transaction. + """ + src = inspect.getsource(hdm._run_merge_pass) + assert "_JOURNAL_INSERT_SQL" in src + assert "db.commit()" not in src, "the pass must not commit — the caller owns the txn" + # phase 1 (snapshots) strictly before the steps mutate anything, delete strictly after. + assert src.index("_JOURNAL_INSERT_SQL") < src.index("for label, sql in _STEPS") + assert src.index("for label, sql in _STEPS") < src.index("_DELETE_LOSERS_SQL") + + +def test_every_step_returns_what_it_touched() -> None: + """Each step must RETURN its rows: ids for survivors, full snapshots for destroyed rows.""" + for label, sql in hdm._STEPS: + assert "RETURNING" in sql, f"{label}: no RETURNING — its rows would go unjournaled" + if label.endswith("(collision-delete)") or label.endswith("(dedup)"): + assert "to_jsonb(" in sql, f"{label}: destroys rows, must snapshot them, not ids" + else: + assert "AS child_id" in sql, f"{label}: re-points rows, must return their ids" + + +@pytest.mark.skipif(_live_session() is None, reason="no reachable Postgres test DB") +def test_real_merge_is_reversible_via_journal() -> None: + """End-to-end on a real DB: merge → journal is sufficient → undo restores the ORIGINAL state. + + The comparison is over `to_jsonb(row.*)` for every row that existed before the merge — all + columns, not a chosen pair — for houses and for every FK child touched. + """ + from sqlalchemy import text as _t + + db = _live_session() + assert db is not None + ids = "(900201, 900202)" + try: + # Keeper 900201 and loser 900202: same canon address, ~12 m apart (inside the 250 m + # guard), keeper has the listings so the keeper rule picks it. + db.execute( + _t( + "INSERT INTO houses (id, source, ext_house_id, url, address, lat, lon, geom, " + "year_built, house_fias_id, gar_flat_count, raw_payload) VALUES " + "(900201,'avito','K-2690','http://t/2690/k','улица Журнальная, 7', " + " 56.8400, 60.6000, ST_SetSRID(ST_MakePoint(60.6000,56.8400),4326), " + " 1979, NULL, NULL, '{\"k\":[1,2]}'), " + "(900202,'cian','L-2690','http://t/2690/l','ул. Журнальная,7', " + " 56.8401, 60.6000, ST_SetSRID(ST_MakePoint(60.6000,56.8401),4326), " + # NB: no «:word» inside the literal — SQLAlchemy text() would read it as a bind. + " NULL, 'fias-2690-uuid', 144, '{\"l\":{\"deep\":[3,4]}}')" + ) + ) + db.execute( + _t( + "INSERT INTO listings (id, source, source_url, source_id, dedup_hash, price_rub, " + "house_id_fk) VALUES " + "(910201,'avito','http://t/2690/1','L1','dh-2690-1',5000000,900201)," + "(910202,'avito','http://t/2690/2','L2','dh-2690-2',5100000,900201)," + "(910203,'cian','http://t/2690/3','L3','dh-2690-3',6000000,900202)" + ) + ) + db.execute( + _t( + "INSERT INTO house_sources (house_id, ext_source, ext_id, confidence, " + "matched_method) VALUES (900201,'avito','S-2690-K',1.0,'t')," + "(900202,'cian','S-2690-L',1.0,'t')" + ) + ) + # Colliding child: identical 6-col UNIQUE key on both → the loser's row is DESTROYED by + # the dedup step. Only a full snapshot can bring it back. + db.execute( + _t( + "INSERT INTO houses_price_dynamics (house_id, month_date, source, room_count, " + "prices_type, period, price_per_sqm) VALUES " + "(900201, DATE '2026-02-01','cian','all','priceSqm','allTime',100000)," + "(900202, DATE '2026-02-01','cian','all','priceSqm','allTime',999999)" + ) + ) + db.commit() + + def snapshot() -> dict[tuple[str, int], Any]: + """to_jsonb of every seeded row, keyed by (table, id) — the full-fidelity state.""" + out: dict[tuple[str, int], Any] = {} + for tbl, col in ( + ("houses", "id"), + ("listings", "house_id_fk"), + ("house_sources", "house_id"), + ("houses_price_dynamics", "house_id"), + ): + where = f"id IN {ids}" if tbl == "houses" else f"{col} IN {ids}" + for r in db.execute( + _t(f"SELECT id, to_jsonb(t.*) AS j FROM {tbl} t WHERE {where}") + ): + out[(tbl, r.id)] = r.j + return out + + before = snapshot() + assert len(before) == 9, f"fixture should seed 9 rows, got {sorted(before)}" + + # ── merge ── + out = hdm.merge_duplicate_houses(db, dry_run=False, initiator="test") + assert out["losers_deleted"] == 1 + assert db.execute(_t(f"SELECT count(*) FROM houses WHERE id IN {ids}")).scalar() == 1 + + # ── the journal alone must be able to NAME what went into what ── + row = db.execute( + _t("SELECT * FROM house_merge_log WHERE loser_id = 900202 ORDER BY id DESC LIMIT 1") + ).one() + assert (row.loser_id, row.keeper_id) == (900202, 900201) + assert row.merge_pass == "canon" and row.geo_guard is True + assert row.cluster_key.startswith("addr:") + assert 0 < row.distance_m < 250, "distance to the keeper must be recorded, in metres" + assert row.initiator == "test" + # full snapshot of the deleted row, not a reference to it + assert row.loser_row == before[("houses", 900202)] + # keeper as it stood BEFORE the identity carry-over (fias still empty there, filled now) + assert row.keeper_before["house_fias_id"] is None + assert ( + db.execute(_t("SELECT house_fias_id FROM houses WHERE id = 900201")).scalar() + == "fias-2690-uuid" + ), "carry-over should have moved the loser's fias up" + # children: the loser's listing moved by id, the destroyed price row by content + assert row.children_repointed["listings.house_id_fk"] == [910203] + assert [r["price_per_sqm"] for r in row.children_deleted["houses_price_dynamics"]] == [ + 999999 + ] + + # ── undo ── + report = db.execute( + _t("SELECT * FROM house_merge_undo(CAST(:b AS uuid))"), {"b": str(row.batch_id)} + ).all() + assert len(report) == 1 and report[0].out_status.startswith("restored:"), report + db.commit() + + after = snapshot() + # every row that existed before is back, byte-identical, on every column + assert {k: v for k, v in after.items() if k in before} == before + # the ONLY residue is the house_sources row the merge backfilled for the keeper. + # migration 230 documents this: it is built from the keeper's OWN ext_house_id, so + # it would have been true without the merge too. Asserted, not assumed. + residue = [v for k, v in after.items() if k not in before] + assert all(v["matched_method"] == "backfill_dedup_merge" for v in residue), residue + finally: + db.rollback() + db.execute(_t(f"DELETE FROM listings WHERE house_id_fk IN {ids}")) + db.execute(_t("DELETE FROM listings WHERE id IN (910201,910202,910203)")) + db.execute(_t("DELETE FROM house_merge_log WHERE loser_id = 900202")) + db.execute(_t("DELETE FROM house_address_aliases WHERE house_id IN (900201,900202)")) + db.execute(_t(f"DELETE FROM houses WHERE id IN {ids}")) + db.commit() + db.close() + + +def test_merge_refuses_when_an_fk_child_is_unhandled() -> None: + """A new FK child on houses(id) must ABORT the merge, not be CASCADE-deleted unjournaled. + + 9 of the 11 FKs are ON DELETE CASCADE. A child the _STEPS pipeline does not know about is + therefore destroyed by `DELETE FROM houses` — no re-point step touches it, no RETURNING + records it, and the journal would claim a complete snapshot it does not have. Migration 133 + already broke on prod over a missed child; there it failed loudly, here it would be silent. + """ + db = _FakeDB( + mapping_rows=[_Row(2, 1, "ул. ленина, 5")], + fk_children={**_FK_CHILDREN, "house_brand_new_child": "house_id"}, + ) + with pytest.raises(RuntimeError, match="house_brand_new_child"): + hdm.merge_duplicate_houses(db, dry_run=False) # type: ignore[arg-type] + assert db.commits == 0, "an unhandled child must abort before anything is committed"