diff --git a/backend/app/services/site_finder/supply_layers.py b/backend/app/services/site_finder/supply_layers.py index 7f74eaa7..a3e205b2 100644 --- a/backend/app/services/site_finder/supply_layers.py +++ b/backend/app/services/site_finder/supply_layers.py @@ -104,6 +104,10 @@ class SupplyLayerRow: district_name: str | None complex_id: int | None obj_class: str | None + # L3-grain (#970, м.128): бакет future-supply по группе застройщика. ЧАСТЬ логического + # ключа uq_supply_layers_logical → разные dev_group'ы района дают разные ключи и больше + # не схлопываются (~95.6% потерь L3 до фикса). NULL для L1/L2 (у них dev_group'а нет). + dev_group_name: str | None units_estimate: int | None area_estimate: float | None expected_online_date: date | None @@ -118,6 +122,7 @@ class SupplyLayerRow: "district_name": self.district_name, "complex_id": self.complex_id, "obj_class": self.obj_class, + "dev_group_name": self.dev_group_name, "units_estimate": self.units_estimate, "area_estimate": _round_or_none(self.area_estimate, 1), "expected_online_date": self.expected_online_date, @@ -307,6 +312,7 @@ def compute_layer1_open( district_name=r["district"], complex_id=None, obj_class=r["obj_class"], + dev_group_name=None, # L1 (open) — у района-агрегата dev_group'а нет units_estimate=units, area_estimate=area if area > 0 else None, expected_online_date=None, @@ -360,6 +366,7 @@ def compute_layer2_hidden( district_name=r["district_name"], complex_id=None, obj_class=None, + dev_group_name=None, # L2 (hidden) — district-агрегат, dev_group'а нет units_estimate=hidden, area_estimate=None, # площадь скрытого недёшево/ненадёжно вывести — None expected_online_date=None, @@ -415,6 +422,9 @@ def compute_layer3_future( district_name=r["district_name"], complex_id=None, obj_class=None, + # #970: dev_group_name теперь КЛЮЧ-колонка (uq_supply_layers_logical), + # не только текст в method → dev_group'ы района не схлопываются на UPSERT. + dev_group_name=dev_group, units_estimate=units if units > 0 else None, area_estimate=area if area > 0 else None, expected_online_date=r["expected_online_date"], diff --git a/backend/app/workers/tasks/supply_layers_refresh.py b/backend/app/workers/tasks/supply_layers_refresh.py index b51bbf48..eddcd606 100644 --- a/backend/app/workers/tasks/supply_layers_refresh.py +++ b/backend/app/workers/tasks/supply_layers_refresh.py @@ -4,8 +4,9 @@ Step 5. Для каждого района Свердловской обл. вызывает детерминированный сервис ``supply_layers.compute_all_layers`` (L1 open / L2 hidden / L3 future) и UPSERT-ит полученные ``SupplyLayerRow`` в таблицу ``supply_layers`` по логическому ключу -``uq_supply_layers_logical`` (layer, district_name, complex_id, obj_class, source, -snapshot_date). +``uq_supply_layers_logical`` (layer, district_name, complex_id, obj_class, +dev_group_name, source, snapshot_date) — dev_group_name в ключе с м.128 (#970), иначе +dev_group'ы района схлопывались (~95.6% потерь L3). Детерминированно, без LLM — сервис read-only считает агрегаты, этот worker только пишет. ON CONFLICT делает повторные прогоны идемпотентными в рамках одного дня @@ -14,7 +15,7 @@ snapshot_date). Контракт записи (КРИТИЧНО, флаг database-expert): на конфликте обновляем ВСЕ мутабельные колонки (units/area/expected_online_date/confidence/method/computed_at), иначе при ре-прогоне в таблице останутся stale-значения. Ключевые колонки (layer, -district_name, complex_id, obj_class, source, snapshot_date) НЕ трогаем. +district_name, complex_id, obj_class, dev_group_name, source, snapshot_date) НЕ трогаем. Mirror conventions (cbr_macro_sync.py / refresh_quarter_price_index.py): • ``SessionLocal()`` + try/finally close, ``logger`` (не print). @@ -71,15 +72,18 @@ _DISTRICTS_SQL = text( """ ) -# UPSERT в supply_layers (м.125). Конфликт по uq_supply_layers_logical (NULLS NOT -# DISTINCT) → ON CONFLICT по точному column-list ключа. DO UPDATE обновляет ВСЕ +# UPSERT в supply_layers (м.125 + м.128). Конфликт по uq_supply_layers_logical (NULLS +# NOT DISTINCT) → ON CONFLICT по точному column-list ключа. DO UPDATE обновляет ВСЕ # мутабельные колонки + computed_at=now() (database-expert: иначе stale-данные). -# Ключевые колонки (layer/district_name/complex_id/obj_class/source/snapshot_date) -# НЕ обновляются — это и есть ключ. CAST(:x AS type) везде (psycopg v3, никогда ::). +# Ключевые колонки (layer/district_name/complex_id/obj_class/dev_group_name/source/ +# snapshot_date) НЕ обновляются — это и есть ключ. dev_group_name добавлен в ключ м.128 +# (#970): без него все dev_group'ы района схлопывались в один NULL-key (complex_id всегда +# NULL для L3) → UPSERT перезатирал их (~95.6% потерь L3). CAST(:x AS type) везде +# (psycopg v3, никогда ::). _UPSERT_SQL = text( """ INSERT INTO supply_layers ( - layer, district_name, complex_id, obj_class, + layer, district_name, complex_id, obj_class, dev_group_name, units_estimate, area_estimate, expected_online_date, source, confidence, method, snapshot_date ) VALUES ( @@ -87,6 +91,7 @@ _UPSERT_SQL = text( CAST(:district_name AS text), CAST(:complex_id AS bigint), CAST(:obj_class AS text), + CAST(:dev_group_name AS text), CAST(:units_estimate AS integer), CAST(:area_estimate AS numeric), CAST(:expected_online_date AS date), @@ -95,7 +100,9 @@ _UPSERT_SQL = text( CAST(:method AS text), CAST(:snapshot_date AS date) ) - ON CONFLICT (layer, district_name, complex_id, obj_class, source, snapshot_date) + ON CONFLICT ( + layer, district_name, complex_id, obj_class, dev_group_name, source, snapshot_date + ) DO UPDATE SET units_estimate = EXCLUDED.units_estimate, area_estimate = EXCLUDED.area_estimate, diff --git a/backend/tests/services/site_finder/test_supply_layers.py b/backend/tests/services/site_finder/test_supply_layers.py index eba94c60..8fd5b192 100644 --- a/backend/tests/services/site_finder/test_supply_layers.py +++ b/backend/tests/services/site_finder/test_supply_layers.py @@ -113,6 +113,7 @@ class TestSupplyLayerRowAsDict: district_name="Автовокзал", complex_id=None, obj_class=None, + dev_group_name=None, units_estimate=70, area_estimate=1234.5678, expected_online_date=None, @@ -124,18 +125,38 @@ class TestSupplyLayerRowAsDict: d = row.as_dict() assert d["layer"] == 2 assert d["district_name"] == "Автовокзал" + assert d["dev_group_name"] is None assert d["units_estimate"] == 70 assert d["area_estimate"] == 1234.6 assert d["source"] == "domrf_declared" assert d["confidence"] == "medium" assert d["snapshot_date"] == date(2026, 6, 3) + def test_dev_group_name_carried_for_l3(self) -> None: + # #970: L3-строка несёт непустой dev_group_name → попадает в as_dict (ключ UPSERT). + row = SupplyLayerRow( + layer=3, + district_name="Академический", + complex_id=None, + obj_class=None, + dev_group_name="Брусника", + units_estimate=900, + area_estimate=45000.0, + expected_online_date=date(2028, 1, 1), + source="domrf_multiphase", + confidence="low", + method="test", + snapshot_date=date(2026, 6, 3), + ) + assert row.as_dict()["dev_group_name"] == "Брусника" + def test_none_area_survives(self) -> None: row = SupplyLayerRow( layer=2, district_name="X", complex_id=None, obj_class=None, + dev_group_name=None, units_estimate=0, area_estimate=None, expected_online_date=None, @@ -220,6 +241,7 @@ class TestLayer1Open: assert r.confidence in _ALLOWED_CONFIDENCE assert r.expected_online_date is None assert r.complex_id is None + assert r.dev_group_name is None # #970: L1 не имеет dev_group first = out[0] assert first.obj_class == "комфорт" assert first.units_estimate == 42 @@ -321,6 +343,7 @@ class TestLayer2Hidden: assert row.confidence == "medium" assert row.units_estimate == 540 assert row.obj_class is None # district-агрегат класс-agnostic + assert row.dev_group_name is None # #970: L2 не имеет dev_group assert row.area_estimate is None assert row.expected_online_date is None assert row.source in _ALLOWED_SOURCES @@ -426,9 +449,60 @@ class TestLayer3Future: assert row.units_estimate == 900 assert row.area_estimate == 45000.0 assert row.expected_online_date == date(2028, 1, 1) + # #970: dev_group_name из SQL-строки протянут в dataclass (ключ-колонка UPSERT) + assert row.dev_group_name == "Брусника" assert row.source in _ALLOWED_SOURCES assert row.confidence in _ALLOWED_CONFIDENCE + def test_same_district_different_dev_groups_distinct_keys(self) -> None: + # #970 (REGRESSION GUARD): два dev_group'а ОДНОГО района должны давать РАЗНЫЕ + # логические ключи (layer, district_name, complex_id, obj_class, dev_group_name, + # source, snapshot_date), иначе UPSERT воркера схлопнет их в один (был ~95.6% + # потерь L3). complex_id/obj_class у L3 всегда NULL → раньше единственным + # различителем был dev_group_name, но его НЕ было в строке → коллапс. + rows = [ + { + "district_name": "Академический", + "dev_group_name": "Брусника", + "n_objects": 5, + "units_future": 900, + "area_future": 45000.0, + "expected_online_date": date(2028, 1, 1), + }, + { + "district_name": "Академический", + "dev_group_name": "Атомстройкомплекс", + "n_objects": 7, + "units_future": 1200, + "area_future": 60000.0, + "expected_online_date": date(2029, 6, 1), + }, + ] + db = _mock_db(rows) + out = compute_layer3_future(db, district="Академический") + assert len(out) == 2 + # каждый dev_group протянут в свою строку (не только в method-текст) + assert {r.dev_group_name for r in out} == {"Брусника", "Атомстройкомплекс"} + + # ключевой инвариант: UPSERT-ключи двух строк РАЗЛИЧНЫ (не схлопываются) + def _key(r: SupplyLayerRow) -> tuple: + d = r.as_dict() + return ( + d["layer"], + d["district_name"], + d["complex_id"], + d["obj_class"], + d["dev_group_name"], + d["source"], + d["snapshot_date"], + ) + + keys = [_key(r) for r in out] + assert len(set(keys)) == 2, f"dev_group keys collapsed: {keys}" + # всё, КРОМЕ dev_group_name, у них идентично — единственный различитель = dev_group + assert keys[0][:4] == keys[1][:4] # layer/district/complex_id/obj_class совпадают + assert keys[0][4] != keys[1][4] # dev_group_name различается + def test_zero_units_and_area_become_none(self) -> None: rows = [ { diff --git a/backend/tests/workers/test_supply_layers_refresh.py b/backend/tests/workers/test_supply_layers_refresh.py index b82278b6..4981cab7 100644 --- a/backend/tests/workers/test_supply_layers_refresh.py +++ b/backend/tests/workers/test_supply_layers_refresh.py @@ -67,6 +67,7 @@ def _row( units: int | None = 10, area: float | None = 500.0, obj_class: str | None = None, + dev_group_name: str | None = None, expected: date | None = None, snapshot_date: date | None = None, ) -> SupplyLayerRow: @@ -75,6 +76,7 @@ def _row( district_name=district, complex_id=None, obj_class=obj_class, + dev_group_name=dev_group_name, units_estimate=units, area_estimate=area, expected_online_date=expected, @@ -89,18 +91,24 @@ def _row( def test_upsert_sql_conflict_target_and_updated_columns() -> None: - """ON CONFLICT таргетит точный логический ключ; DO UPDATE обновляет ВСЕ - мутабельные колонки + computed_at; ключевые колонки НЕ в SET; нет :x::type.""" + """ON CONFLICT таргетит точный логический ключ (#970: + dev_group_name); DO UPDATE + обновляет ВСЕ мутабельные колонки + computed_at; ключевые колонки НЕ в SET; нет :x::type.""" from app.workers.tasks.supply_layers_refresh import _UPSERT_SQL sql = str(_UPSERT_SQL) upper = sql.upper() + # нормализуем whitespace — ON CONFLICT column-list перенесён на неск. строк (м.128). + flat = " ".join(upper.split()) assert "INSERT INTO SUPPLY_LAYERS" in upper - # точный ON CONFLICT target = uq_supply_layers_logical column-list + # точный ON CONFLICT target = uq_supply_layers_logical column-list (м.128, с dev_group_name) assert ( - "ON CONFLICT (LAYER, DISTRICT_NAME, COMPLEX_ID, OBJ_CLASS, SOURCE, SNAPSHOT_DATE)" in upper - ) + "ON CONFLICT ( LAYER, DISTRICT_NAME, COMPLEX_ID, OBJ_CLASS, " + "DEV_GROUP_NAME, SOURCE, SNAPSHOT_DATE )" in flat + ), f"ON CONFLICT target mismatch (must include dev_group_name): {flat}" + # dev_group_name присутствует в INSERT column-list и в VALUES как CAST(...AS text) + assert "DEV_GROUP_NAME," in upper + assert "CAST(:DEV_GROUP_NAME AS TEXT)" in upper assert "DO UPDATE SET" in upper # ВСЕ мутабельные колонки обновляются (database-expert: иначе stale-данные) for col in ( @@ -112,13 +120,15 @@ def test_upsert_sql_conflict_target_and_updated_columns() -> None: "COMPUTED_AT = NOW()", ): assert col in upper, f"missing updated column: {col}" - # ключевые колонки НЕ обновляются в SET-части (после DO UPDATE SET) + # ключевые колонки НЕ обновляются в SET-части (после DO UPDATE SET) — dev_group_name + # теперь тоже КЛЮЧ (#970) → не должен апдейтиться. set_clause = upper.split("DO UPDATE SET", 1)[1] for key_col in ( "LAYER = EXCLUDED", "DISTRICT_NAME = EXCLUDED", "COMPLEX_ID = EXCLUDED", "OBJ_CLASS = EXCLUDED", + "DEV_GROUP_NAME = EXCLUDED", "SOURCE = EXCLUDED", "SNAPSHOT_DATE = EXCLUDED", ): @@ -239,6 +249,60 @@ def test_upsert_rows_savepoint_isolates_bad_row() -> None: assert [p["district_name"] for p in captured] == ["A", "C"] +def test_upsert_rows_l3_same_district_diff_dev_group_distinct_keys() -> None: + """#970 (REGRESSION GUARD, worker-level): две L3-строки ОДНОГО района с РАЗНЫМИ + dev_group_name проходят как ДВЕ строки с РАЗЛИЧНЫМИ ключами UPSERT — не схлопываются. + + Без dev_group_name в ключе (complex_id/obj_class у L3 = NULL) обе дали бы один ключ + → ON CONFLICT перезатёр бы их (~95.6% потерь L3 на prod). Проверяем, что param-dict + каждой строки несёт свой dev_group_name и ключевые кортежи различны. + """ + from app.workers.tasks import supply_layers_refresh as task_mod + + db, captured, _ = _make_mock_db([]) + db.execute("SELECT FROM supply_layers warmup", None) # пометить districts обслуженными + snap = date(2026, 6, 3) + rows = [ + _row( + layer=3, + district="Академический", + source="domrf_multiphase", + confidence="low", + dev_group_name="Брусника", + snapshot_date=snap, + ), + _row( + layer=3, + district="Академический", + source="domrf_multiphase", + confidence="low", + dev_group_name="Атомстройкомплекс", + snapshot_date=snap, + ), + ] + upserted, skipped = task_mod._upsert_rows(db, rows) + + assert upserted == 2 + assert skipped == 0 + # оба dev_group дошли до UPSERT-params (binding передаёт dev_group_name) + assert {p["dev_group_name"] for p in captured} == {"Брусника", "Атомстройкомплекс"} + + # ключевые кортежи (под uq_supply_layers_logical м.128) различны → нет коллапса + def _key(p: dict[str, Any]) -> tuple: + return ( + p["layer"], + p["district_name"], + p["complex_id"], + p["obj_class"], + p["dev_group_name"], + p["source"], + p["snapshot_date"], + ) + + keys = [_key(p) for p in captured] + assert len(set(keys)) == 2, f"L3 dev_group keys collapsed at worker level: {keys}" + + # ── end-to-end task: coherent snapshot, counters, resilience ───────────────── diff --git a/data/sql/128_supply_layers_dev_group_key.sql b/data/sql/128_supply_layers_dev_group_key.sql new file mode 100644 index 00000000..a6fc2d51 --- /dev/null +++ b/data/sql/128_supply_layers_dev_group_key.sql @@ -0,0 +1,99 @@ +-- 128_supply_layers_dev_group_key.sql +-- #970 (REOPENED, CRITICAL) — Layer-3 future-supply collapse: добавляем dev_group_name в зерно +-- и в логический UNIQUE-ключ supply_layers, чтобы dev_group'ы района перестали схлопываться. +-- +-- БАГ (ground-truth, измерено на prod): +-- Layer-3 (future supply) в supply_layers.py (_L3_FUTURE_SQL) считается per +-- (district_name, dev_group_name) — GROUP BY district_name, dev_group_name — но строки +-- эмитятся с complex_id=NULL и БЕЗ dev_group_name (колонки нет с м.125). Логический ключ +-- uq_supply_layers_logical (м.125) = (layer, district_name, complex_id, obj_class, source, +-- snapshot_date) NULLS NOT DISTINCT — dev_group_name в нём НЕТ. Итог: ВСЕ dev_group'ы района +-- дают ОДИН и тот же ключ (complex_id всегда NULL для L3) → UPSERT воркера +-- (supply_layers_refresh.py, ON CONFLICT) перезатирает их друг другом → массовая потеря. +-- Измерено (Академический, latest snapshot): должно быть 13 808 юнитов в 15 dev_group'ах / +-- 54 объектах, выживает 1 строка / 607 юнитов (~95.6% потерь, недетерминированно per-run). +-- +-- ЧИНИМ (схема): dev_group_name становится частью L3-grain и частью логического ключа. +-- 1. ADD COLUMN dev_group_name TEXT (nullable — у L1/L2 dev_group'а НЕТ, его ставит только L3). +-- 2. Перестроить uq_supply_layers_logical, ДОБАВив dev_group_name: +-- (layer, district_name, complex_id, obj_class, dev_group_name, source, snapshot_date) +-- по-прежнему UNIQUE NULLS NOT DISTINCT (PG15+, как м.125). +-- +-- Почему NULLS NOT DISTINCT сохраняется (КРИТИЧНО — корректность дедупа обоих случаев): +-- В обычном UNIQUE Postgres считает NULL DISTINCT → (а) region-wide / L1/L2 строки +-- (district_name/complex_id/obj_class/dev_group_name IS NULL) дублировались бы, (б) ON CONFLICT +-- их НЕ матчил бы (NULL <> NULL). NULLS NOT DISTINCT сравнивает NULL'ы как РАВНЫЕ. Следствия: +-- • L1/L2 строки: dev_group_name=NULL у ВСЕХ из них → новая колонка для них «прозрачна», +-- они продолжают дедуплиться по ОСТАЛЬНОЙ части ключа ровно как раньше (NULL=NULL по +-- dev_group_name не ломает их уникальность). Семантика L1/L2 НЕ меняется. +-- • L3 строки: теперь несут РАЗНЫЕ непустые dev_group_name → дают РАЗНЫЕ ключи → больше +-- НЕ схлопываются; разные dev_group'ы района сосуществуют (15 строк вместо 1). +-- Прецеденты NULLS NOT DISTINCT в репо: м.125 (исходный ключ), м.110 (uq_infra_dedupe). +-- +-- Зависимости / окружение: +-- • supply_layers (м.125) — ALTER существующей таблицы (ADD COLUMN + rebuild constraint). +-- • PostgreSQL 16 (postgis/postgis:16-3.4) — нужен для UNIQUE NULLS NOT DISTINCT (PG15+). +-- • Unique-constraint сам по себе даёт индекс — отдельный supporting-индекс под этот ключ +-- НЕ нужен (м.125 его и не заводила; read-path индексы idx_supply_layers_* остаются как есть). +-- +-- Idempotent (acceptance — ре-applied чисто ДВАЖДЫ): ADD COLUMN IF NOT EXISTS + guarded DROP/ADD +-- констрейнта через DO-блок с проверкой pg_constraint (мирроринг м.125 / м.40). Re-apply +-- конвергентен. BEGIN/COMMIT — атомарно. НИКАКИХ DROP TABLE/TRUNCATE, никаких INSERT/строк. +-- +-- Deploy order (КРИТИЧНО): эта SQL-миграция применяется ПЕРВОЙ (auto-apply через deploy.yml / +-- _schema_migrations tracking, ровно один раз по NN). ЗАТЕМ — отдельным backend-изменением +-- (НЕ в этом PR) воркер backend/app/workers/tasks/supply_layers_refresh.py: +-- • _UPSERT_SQL INSERT column-list + VALUES должны ДОБАВИТЬ dev_group_name +-- (CAST(:dev_group_name AS text)), +-- • ON CONFLICT (...) должен стать +-- (layer, district_name, complex_id, obj_class, dev_group_name, source, snapshot_date) — +-- ровно под новый ключ, +-- • а supply_layers.py L3-emit (SupplyLayerRow, layer=3) должен ПОПУЛИРОВАТЬ dev_group_name +-- (сейчас он только пишет dev_group в method-строку, в саму строку — нет). +-- Никогда наоборот: воркер с новым ON CONFLICT упадёт без этой колонки/ключа. +-- Существующие схлопнутые данные САМО-залечиваются на следующем supply_layers_refresh — +-- воркер перезапишет район строками по ИСПРАВЛЕННОМУ ключу (15 dev_group'ов вместо 1). + +BEGIN; + +-- ── 1. Новая колонка dev_group_name (L3-grain) ───────────────────────────────── +-- Nullable: L1/L2 dev_group'а не имеют (ставит только L3 = бакет future-supply по группе +-- застройщика). ADD COLUMN IF NOT EXISTS → re-apply пропускает добавление (safe). +ALTER TABLE supply_layers + ADD COLUMN IF NOT EXISTS dev_group_name TEXT; + +COMMENT ON COLUMN supply_layers.dev_group_name IS + 'L3-grain (#970): бакет future-supply по группе застройщика (dev_group_name, м.113). ' + 'Часть логического ключа uq_supply_layers_logical для L3 (domrf_multiphase). NULL для ' + 'L1/L2 (у них dev_group''а нет). Без этой колонки в ключе все dev_group''ы района ' + 'схлопывались в один NULL-key и UPSERT их перезатирал (~95.6% потерь L3).'; + +-- ── 2. Перестройка логического UNIQUE-ключа: ДОБАВляем dev_group_name ─────────── +-- Цель ON CONFLICT воркера (#950 Step5 / #970 fix). NULLS NOT DISTINCT сохраняется (PG15+): +-- L1/L2 (dev_group_name=NULL) дедуплятся по остальной части ключа как раньше; L3 с разными +-- непустыми dev_group_name дают разные ключи и больше не схлопываются. Guarded DROP+ADD через +-- pg_constraint → idempotent re-apply (констрейнт нельзя добавить дважды; имя то же — +-- uq_supply_layers_logical, чтобы ON CONFLICT таргетил его по column-list). Паттерн зеркалит +-- м.125 (constraint-add guard) + м.40 (FK guard). DROP с тем же именем безопасен: на «чистой» +-- м.125-таблице он снимает старый 6-колоночный ключ, затем ставит новый 7-колоночный; на +-- повторном прогоне старого имени уже нет → DROP no-op (IF EXISTS), ADD пропускается guard'ом. +DO $$ +BEGIN + -- Снять старый ключ (м.125: без dev_group_name), если он есть. IF EXISTS → re-apply safe. + ALTER TABLE supply_layers + DROP CONSTRAINT IF EXISTS uq_supply_layers_logical; + + -- Поставить новый ключ (с dev_group_name) только если его ещё нет → re-apply конвергентен. + IF NOT EXISTS ( + SELECT 1 FROM pg_constraint WHERE conname = 'uq_supply_layers_logical' + ) THEN + ALTER TABLE supply_layers + ADD CONSTRAINT uq_supply_layers_logical + UNIQUE NULLS NOT DISTINCT + (layer, district_name, complex_id, obj_class, + dev_group_name, source, snapshot_date); + END IF; +END +$$; + +COMMIT;