lekss361/gendesign
1
0
Fork 0
Code Issues 143 Pull requests Projects Releases Packages Wiki Activity Actions

feat(tradein/houses): журнал слияний домов — слияние стало обратимым (#2740)
All checks were successful
Deploy Trade-In / changes (push) Successful in 11s
Details
Deploy Trade-In / build-frontend (push) Has been skipped
Details
Deploy Trade-In / build-browser (push) Has been skipped
Details
Deploy Trade-In / test (push) Successful in 3m6s
Details
Deploy Trade-In / build-backend (push) Successful in 1m2s
Details
Deploy Trade-In / deploy (push) Successful in 1m31s
Details

Browse source
This commit is contained in:
bot-backend 2026-08-06 15:41:20 +00:00
parent 0dc6f12630
commit c86a5378ef
3 changed files with 822 additions and 67 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

272
tradein-mvp/backend/app/services/house_dedup_merge.py
View file

@ -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",

262
tradein-mvp/backend/data/sql/230_house_merge_log.sql Normal file
View file

@ -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:<uuid>»), '
'а не имя ключа — по нему видно, какое именно совпадение сработало.';
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(''<batch_id>''); -- прочитать статусы -- COMMIT;';
COMMIT;

355
tradein-mvp/backend/tests/test_house_dedup_merge.py
View file

@ -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"