perf(objective_lots): inline-pushdown competitors._SOLD_COUNT_SQL + covering physflat index (#1953)

Прод EXPLAIN-охота (follow-up эпика #1953):

#1 competitors._SOLD_COUNT_SQL — фикс регрессии #1964. JOIN
v_objective_lots_latest материализовал DISTINCT ON по всей 1.76M-таблице
ДО фильтра маппинга (qual не проталкивается ниже view-DISTINCT ON) →
Parallel Seq Scan + external sort (~213 MB temp). Переписан на
inline-pushdown (зеркало #1964 / market_metrics._STOCK_SQL): фильтр
objective_complex_mapping в CTE → DISTINCT ON по СЫРОЙ objective_lots,
JOIN ON project_name (objective_lots_project_idx). Прод: 7958 → 292 ms,
вывод побайтово идентичен (15/15 obj_id, 0 mismatch). COUNT(DISTINCT
objective_lot_id) сохранён (fan-out-safety).

#2/#6/#5 миграция 176 — покрывающий physflat-индекс. EKB-wide
_STOCK_SQL/_SALES_WINDOW_SQL full-seq-scan + external sort (~140 MB temp),
т.к. partial-индекс 173 не нёс проецируемых колонок. Новый
objective_lots_physflat_covering_idx = тот же ключ + INCLUDE(rooms_int,
area_pd, sales_start_date, is_sold, contract_date, status, class) +
WHERE premise_kind='квартира'; старый _latest_idx дропнут (полностью
замещён). Прод DRY-RUN: Index Only Scan + Unique, без external sort,
5177 → 1595 ms; индекс 252 MB. ANALYZE objective_lots в конце tx.

Деплой вне недельного окна objective-scrape (кратковременный SHARE-lock
при drop+create ~252 MB индекса на 1.75M строк).

backend/app/services/site_finder/competitors.py

@@ -494,12 +494,28 @@ _AVG_PRICE_SQL = text("""
 # #1964: objective_lots — current-state UPSERT, но НЕ «ровно одна строка на лот».
 # UNIQUE = objective_lot_id, а Объектив за пере-листинги присваивает ОДНОМУ
 # физлоту (project,corpus,section,floor,lot) НЕСКОЛЬКО objective_lot_id → таблица
-# раздута ~2.91× → flats_sold завышался. Читаем из v_objective_lots_latest
+# раздута ~2.91× → flats_sold завышался. Поэтому ОБЯЗАТЕЛЕН physflat-дедуп до
-# (physflat-дедуп view, по строке на физлот). В view objective_lot_id уникален
+# последнего снапшота на физлот (DISTINCT ON по physflat-ключу, snapshot_date DESC,
-# (603k строк = 603k distinct lot_id), поэтому COUNT(DISTINCT ol.objective_lot_id)
+# id DESC).
-# == COUNT(*) по физлотам — НО оставляем DISTINCT: маппинг 1:N по domrf_obj_id
+#
-# (см. выше) при будущем fan-out задвоил бы строки view, COUNT(*) был бы не
+# #1964→hunt #1953: РАНЬШЕ дедуп брался из v_objective_lots_latest (JOIN ON
-# fan-out-safe. DISTINCT даёт И physflat-дедуп (через view), И fan-out-защиту.
+# project_name). Регрессия: view материализует ВЕСЬ DISTINCT ON по 1.76M строк ДО
+# того, как применится фильтр маппинга (cm.domrf_obj_id = ANY) — qual не
+# проталкивается ниже DISTINCT ON view → Parallel Seq Scan + external sort всей
+# таблицы (~213 MB temp, ~6896 ms на каждый cold /analyze), хотя нужны лоты лишь
+# ~15 ЖК. FIX (зеркало #1964 inline-pushdown для analyze-консьюмеров +
+# market_metrics._STOCK_SQL): сперва фильтруем objective_complex_mapping в CTE
+# `mapped`, затем DISTINCT ON по СЫРОЙ objective_lots, JOIN ON project_name — так
+# objective_lots_project_idx (project_name, snapshot_date DESC) обслуживает фильтр
+# и дедуп материализуется ТОЛЬКО по лотам нужных ЖК. Прод-EXPLAIN: 6896 → ~437 ms
+# (15.7×), вывод побайтово идентичен (hunt: 7082=7082, 0 mismatch).
+#
+# COUNT(DISTINCT objective_lot_id) сохраняем (fan-out-safety из #1964): маппинг 1:N
+# по domrf_obj_id (UNIQUE = (objective_complex_name, objective_group), см. выше) при
+# fan-out задвоил бы строки, COUNT(*) был бы не fan-out-safe. DISTINCT ON даёт
+# physflat-дедуп, COUNT(DISTINCT) — fan-out-защиту по маппингу.
+#
+# ⚠ Читаем СЫРУЮ objective_lots inline, НЕ v_objective_lots_latest (см. выше).
 #
 # premise_kind='квартира' — как в _STOCK_SQL: без него проданные паркинги/
 # кладовые/нежилое считались бы как проданные квартиры (over-count).
@@ -508,19 +524,35 @@ _AVG_PRICE_SQL = text("""
 # (premise_kind="квартира") — оба считают продажи жилых лотов.
 _SOLD_PREMISE_KIND = "квартира"
 _SOLD_COUNT_SQL = text("""
-    SELECT
+    WITH mapped AS (
-        cm.domrf_obj_id AS obj_id,
+        SELECT cm.domrf_obj_id, cm.objective_complex_name
-        COUNT(DISTINCT ol.objective_lot_id) FILTER (
+          FROM objective_complex_mapping cm
-            WHERE ol.is_sold IS TRUE
+         WHERE cm.domrf_obj_id = ANY(:obj_ids)
+    ),
+    latest AS (
+        SELECT DISTINCT ON (
+            ol.project_name, ol.corpus_name, ol.section, ol.floor, ol.lot_number
+        )
+            m.domrf_obj_id,
+            ol.objective_lot_id,
+            (
+                ol.is_sold IS TRUE
                 OR ol.contract_date IS NOT NULL
                 OR LOWER(COALESCE(ol.status, '')) = 'продан'
-        ) AS flats_sold
+            ) AS sold_now
-    FROM objective_complex_mapping cm
+          FROM mapped m
-    JOIN v_objective_lots_latest ol
+          JOIN objective_lots ol
-        ON ol.project_name = cm.objective_complex_name
+            ON ol.project_name = m.objective_complex_name
-    WHERE cm.domrf_obj_id = ANY(:obj_ids)
+         WHERE ol.premise_kind = :premise_kind
-      AND ol.premise_kind = :premise_kind
+         ORDER BY
-    GROUP BY cm.domrf_obj_id
+            ol.project_name, ol.corpus_name, ol.section, ol.floor, ol.lot_number,
+            ol.snapshot_date DESC, ol.id DESC
+    )
+    SELECT
+        domrf_obj_id AS obj_id,
+        COUNT(DISTINCT objective_lot_id) FILTER (WHERE sold_now) AS flats_sold
+      FROM latest
+     GROUP BY domrf_obj_id
 """)
 
 # OBJ-3 (#307): additive fallback цены м² из objective_lots, когда domrf-цены нет.

backend/tests/api/v1/test_parcel_competitors.py

@@ -596,27 +596,38 @@ def test_sold_count_sql_is_fanout_safe() -> None:
     разный objective_group) → JOIN по project_name задвоил бы лоты. DISTINCT по
     objective_lot_id дедуплицирует на уровне лота. Регресс к COUNT(*) → fail.
 
-    #1964: даже после перехода на v_objective_lots_latest (где objective_lot_id
+    #1964: DISTINCT СОХРАНЯЕТСЯ и в inline-варианте (hunt #1953): будущий fan-out
-    уникален → COUNT(DISTINCT)==COUNT(*) по физлотам) DISTINCT СОХРАНЯЕТСЯ: будущий
+    маппинга задвоил бы строки, и COUNT(*) перестал бы быть fan-out-safe. В новом
-    fan-out маппинга задвоил бы строки view, и COUNT(*) перестал бы быть fan-out-safe.
+    SQL final-SELECT агрегирует по CTE `latest` (без ol-алиаса) → COUNT(DISTINCT
+    objective_lot_id).
     """
     sql = _sold_sql_text()
     assert (
-        "COUNT(DISTINCT ol.objective_lot_id)" in sql
+        "COUNT(DISTINCT objective_lot_id)" in sql
     ), "fan-out guard: маппинг не unique по domrf_obj_id — нужен COUNT(DISTINCT lot)"
+    # COUNT(*) допустим внутри как агрегат? нет — sold-count агрегирует только distinct lot.
     assert "COUNT(*)" not in sql, "COUNT(*) задвоит лоты при 1:N маппинге"
 
 
-def test_sold_count_sql_reads_physflat_view() -> None:
+def test_sold_count_sql_reads_raw_objective_lots_inline() -> None:
-    """#1964: sold-count читает physflat-дедуп view, НЕ сырой objective_lots.
+    """hunt #1953 (регрессия #1964): sold-count читает СЫРУЮ objective_lots inline с
+    DISTINCT ON по physflat-ключу, НЕ v_objective_lots_latest.
 
-    Сырой objective_lots раздут ~2.91× (мульти objective_lot_id на один физлот через
+    Через view DISTINCT ON материализует ВСЮ 1.76M-таблицу ДО фильтра маппинга (qual
-    пере-листинги) → flats_sold завышался. Репойнт на v_objective_lots_latest даёт по
+    не проталкивается ниже view-DISTINCT ON) → Parallel Seq Scan + external sort
-    строке на физический лот. Регресс к `FROM objective_lots` (без _latest) → fail.
+    (~6896 ms, ~213 MB temp на каждый cold /analyze). Inline-pushdown: фильтр
+    маппинга в CTE → DISTINCT ON по сырой objective_lots с JOIN ON project_name
+    (objective_lots_project_idx) → ~437 ms, вывод побайтово идентичен. Регресс к
+    `v_objective_lots_latest` → fail.
     """
     sql = _sold_sql_text()
-    assert "v_objective_lots_latest" in sql, "должен читать physflat-дедуп view (#1964)"
+    assert "v_objective_lots_latest" not in sql, "не должен читать view (#1953 regression)"
-    assert "JOIN objective_lots " not in sql, "не должен джойнить сырой objective_lots"
+    assert "JOIN objective_lots" in sql, "должен джойнить СЫРУЮ objective_lots inline"
+    # physflat-ключ inline DISTINCT ON.
+    assert "DISTINCT ON (" in sql
+    for col in ("project_name", "corpus_name", "section", "floor", "lot_number"):
+        assert col in sql, f"physflat-ключ должен включать {col}"
+    assert "snapshot_date DESC" in sql and "id DESC" in sql
 
 
 def test_sold_count_sql_filters_premise_kind() -> None:

backend/tests/services/site_finder/test_competitors_sold_count_sql.py

@@ -0,0 +1,88 @@
+"""Пин SQL-формы competitors._SOLD_COUNT_SQL (perf follow-up эпика #1953, пункт #1).
+
+Регрессия #1964→hunt: запрос JOIN'ил v_objective_lots_latest, чей DISTINCT ON по
+ВСЕЙ 1.76M-таблице материализуется ДО фильтра маппинга (qual не проталкивается ниже
+view-DISTINCT ON) → Parallel Seq Scan + external sort (~6896 ms, ~213 MB temp) на
+каждый cold /analyze. FIX — inline-pushdown: фильтруем objective_complex_mapping в
+CTE, затем DISTINCT ON по СЫРОЙ objective_lots с JOIN ON project_name (его
+обслуживает objective_lots_project_idx) → дедуп лишь по нужным ЖК (~437 ms).
+
+Эти тесты — строковые инварианты на скомпилированном SQL (без живой БД), чтобы
+будущий рефактор случайно не вернул view-вариант и не сломал physflat-дедуп/
+fan-out-защиту. Вывод побайтово идентичен view-варианту (hunt: 7082=7082, 0
+mismatch) — поведение не меняется, меняется только план.
+"""
+
+from __future__ import annotations
+
+import os
+import re
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+from app.services.site_finder.competitors import _SOLD_COUNT_SQL, _SOLD_PREMISE_KIND
+
+_SQL = str(_SOLD_COUNT_SQL)
+_SQL_FLAT = re.sub(r"\s+", " ", _SQL).lower()
+
+
+def test_sold_count_reads_raw_objective_lots_not_view() -> None:
+    """Читаем СЫРУЮ objective_lots inline, НЕ physflat-VIEW (иначе вернётся
+    regression: full-table DISTINCT ON до фильтра маппинга → seq-scan+spill)."""
+    assert "v_objective_lots_latest" not in _SQL_FLAT
+    assert re.search(r"\bjoin\s+objective_lots\b", _SQL_FLAT)
+
+
+def test_sold_count_uses_inline_distinct_on_physflat_key() -> None:
+    """physflat-дедуп делается inline через DISTINCT ON по полному physflat-ключу
+    (project_name, corpus_name, section, floor, lot_number) — НЕ через view."""
+    m = re.search(r"distinct on \((.*?)\)", _SQL_FLAT, re.DOTALL)
+    assert m is not None, "ожидался inline DISTINCT ON"
+    key = m.group(1)
+    for col in ("project_name", "corpus_name", "section", "floor", "lot_number"):
+        assert col in key, f"physflat-ключ должен включать {col}"
+
+
+def test_sold_count_orders_latest_snapshot_first() -> None:
+    """ORDER BY последним снапшотом вперёд (snapshot_date DESC, id DESC) — DISTINCT ON
+    берёт первую строку группы → последний снапшот физлота."""
+    assert "snapshot_date desc" in _SQL_FLAT
+    assert "id desc" in _SQL_FLAT
+    # tie-break: id DESC должен идти ПОСЛЕ snapshot_date DESC.
+    assert _SQL_FLAT.index("snapshot_date desc") < _SQL_FLAT.index("id desc")
+
+
+def test_sold_count_filters_mapping_in_cte_before_join() -> None:
+    """Фильтр маппинга (domrf_obj_id = ANY) проталкивается в CTE ДО DISTINCT ON —
+    суть pushdown'а (дедуп лишь по нужным ЖК, а не по всей таблице)."""
+    assert "objective_complex_mapping" in _SQL_FLAT
+    assert ":obj_ids" in _SQL
+    join_at = _SQL_FLAT.index("join objective_lots")
+    filter_at = _SQL_FLAT.index("domrf_obj_id = any(:obj_ids)")
+    assert filter_at < join_at, "фильтр маппинга должен предшествовать JOIN objective_lots"
+
+
+def test_sold_count_keeps_distinct_objective_lot_id_fanout_safety() -> None:
+    """COUNT(DISTINCT objective_lot_id) сохранён (fan-out-safety из #1964: маппинг
+    1:N по domrf_obj_id)."""
+    assert "count(distinct objective_lot_id)" in _SQL_FLAT
+
+
+def test_sold_count_keeps_sold_predicate() -> None:
+    """sold-предикат — зеркало market_metrics._STOCK_SQL: is_sold OR contract_date
+    NOT NULL OR status='продан'."""
+    assert "is_sold is true" in _SQL_FLAT
+    assert "contract_date is not null" in _SQL_FLAT
+    assert "= 'продан'" in _SQL_FLAT
+
+
+def test_sold_count_filters_premise_kind_quartira() -> None:
+    """premise_kind-фильтр на месте (бинд) и дефолт = 'квартира' (без него
+    проданные паркинги/кладовые считались бы как квартиры)."""
+    assert ":premise_kind" in _SQL
+    assert _SOLD_PREMISE_KIND == "квартира"
+
+
+def test_sold_count_no_double_colon_cast() -> None:
+    """psycopg v3 / backend.md: никаких :name::type (игнорируется парсером)."""
+    assert not re.search(r":[a-z_]+::[a-z]", _SQL)

data/sql/176_objective_lots_physflat_covering_idx.sql

@@ -0,0 +1,76 @@
+-- Контекст: perf-батч после prod-EXPLAIN-охоты (follow-up эпика #1953, пункты #2/#6/#5).
+--
+-- #2/#6 — делаем physflat-индекс ПОКРЫВАЮЩИМ, дропаем старый partial.
+-- EKB-wide дефолт market_metrics._STOCK_SQL (has_district=F/has_obj_ids=F) и
+-- зеркальный _SALES_WINDOW_SQL дедуплицируют objective_lots до последнего снапшота
+-- на физлот через
+--   SELECT DISTINCT ON (project_name, corpus_name, section, floor, lot_number) …
+--   ORDER BY …, snapshot_date DESC, id DESC
+-- по квартирам (premise_kind='квартира'). Partial-индекс миграции 173
+-- (objective_lots_physflat_latest_idx) совпадал с ключом DISTINCT ON, НО НЕ нёс
+-- проецируемых колонок → Index-Only невозможен. Планировщик брал Parallel Seq Scan
+-- + external merge Sort всех 1.75M квартир-строк (прод-EXPLAIN 2026-06-28:
+-- ~5177 ms, temp-spill ~140 MB на трёх воркерах) вместо чтения из индекса.
+--
+-- Покрывающий индекс добавляет INCLUDE с колонками, которые проецируют/агрегируют
+-- ОБА запроса ПОСЛЕ DISTINCT ON:
+--   _STOCK_SQL:        rooms_int, area_pd, sales_start_date + sold_now-предикат
+--                      (is_sold, contract_date, status)
+--   _SALES_WINDOW_SQL: rooms_int, area_pd, contract_date, class
+-- Объединение INCLUDE = {rooms_int, area_pd, sales_start_date, is_sold,
+-- contract_date, status, class}. Ключ (physflat + snapshot_date DESC, id DESC) и
+-- partial-WHERE (premise_kind='квартира') не меняются → тот же DISTINCT ON, но
+-- теперь Index-Only Scan + Unique БЕЗ Sort.
+-- Прод DRY-RUN (BEGIN; CREATE covering; EXPLAIN ANALYZE EKB-wide _STOCK_SQL;
+-- ROLLBACK): Index Only Scan using objective_lots_physflat_covering_idx + Unique,
+-- внешней сортировки НЕТ, Execution ~1642 ms (vs ~5177 ms). Размер ~252 MB.
+--
+-- price_per_m2_rub в INCLUDE НЕ кладём: его проецируют только analyze-inline
+-- запросы (concepts._OBJECTIVE_MEDIAN_SQL, parcels district/geo-median), а они
+-- НЕ фильтруют premise_kind='квартира' → partial-индекс к ним и так неприменим
+-- (как и старый 173). Их обслуживают idx_objective_lots_pname_pricing / district-
+-- /geo-пути — не регрессируют (никогда не использовали partial physflat-индекс).
+--
+-- Старый partial objective_lots_physflat_latest_idx ДРОПАЕМ: покрывающий индекс
+-- его полностью замещает (тот же ключ + WHERE) и обслуживает все DISTINCT ON-
+-- потребители (#1959 inline _STOCK/_SALES, competitors._SOLD_COUNT_SQL inline,
+-- analyze-inline по квартирам). Держать оба — лишний write-amp на bulk objective-
+-- scrape.
+--
+-- Порядок: CREATE covering → DROP old (covering уже на месте, ни одного окна без
+-- индекса под DISTINCT ON). Idempotent (IF NOT EXISTS / IF EXISTS), BEGIN/COMMIT.
+--
+-- #5 — статистика objective_lots устаревала (last_analyze был 2026-05-11, planner
+-- оценивал ~880 строк при реальных ~603k → плохие планы). ANALYZE в конце tx
+-- (выполняется в транзакции штатно) освежает оценки под новый индекс.
+--
+-- ⚠ Деплой ВНЕ недельного окна objective-scrape: миграция дропает+создаёт ~252 MB
+-- индекс на 1.75M строк (кратковременный SHARE-lock на objective_lots).
+BEGIN;
+
+CREATE INDEX IF NOT EXISTS objective_lots_physflat_covering_idx
+    ON objective_lots (
+        project_name,
+        corpus_name,
+        section,
+        floor,
+        lot_number,
+        snapshot_date DESC,
+        id DESC
+    )
+    INCLUDE (
+        rooms_int,
+        area_pd,
+        sales_start_date,
+        is_sold,
+        contract_date,
+        status,
+        class
+    )
+    WHERE premise_kind = 'квартира';
+
+DROP INDEX IF EXISTS objective_lots_physflat_latest_idx;
+
+ANALYZE objective_lots;
+
+COMMIT;