compose.yaml

@@ -310,7 +310,10 @@ services:
       - ./infra/airflow/init.sh:/opt/airflow/init.sh:ro
       - airflow_logs:/opt/airflow/logs
       - airflow_auth:/opt/airflow/auth
-    mem_limit: 640m
+    # LocalExecutor запускает здесь до четырех процессов задач одновременно.
+    # 640 МиБ дали OOM, а 1 ГиБ - давление на memory.max при штатном dq_check;
+    # 1,5 ГиБ оставляет запас при AIRFLOW__CORE__PARALLELISM=4.
+    mem_limit: 1536m
     healthcheck:
       test: ["CMD-SHELL", "curl -sf http://127.0.0.1:8974/health >/dev/null"]
       interval: 10s

dags/dds_transform.py

@@ -279,11 +279,6 @@ def replacements(scope: dict[str, object]) -> list[dict[str, object]]:
     # Два прогона разом собирали бы один и тот же двойник: второй затёр бы
     # чужую сборку между вставкой и заменой партиций.
     max_active_runs=1,
-    # Потолок на задачи в прогоне. Размноженная замена норовит выстрелить
-    # всеми днями сразу, а каждая задача — свой процесс исполнителя, и
-    # планировщику отведено 640 МиБ (compose.yaml). Замерено: тринадцать дней
-    # двух сущностей без потолка кончаются SIGKILL от контрольной группы.
-    max_active_tasks=3,
     template_searchpath=str(SQL_ROOT),
     tags=["dds"],
 )

dags/dm_transform.py

@@ -76,9 +76,6 @@ def replacements(scope: dict[str, object]) -> list[dict[str, object]]:
     start_date=START_DATE,
     is_paused_upon_creation=False,
     max_active_runs=1,
-    # Каждая задача LocalExecutor - отдельный процесс. Четыре процесса уже
-    # дали SIGKILL при лимите scheduler 640 МиБ; три - измеренный потолок DDS.
-    max_active_tasks=3,
     template_searchpath=str(SQL_ROOT),
     tags=["dm"],
 )
@@ -136,11 +133,6 @@ def dm_transform():
             conn_id=CLICKHOUSE_CONNECTION,
             sql="dm/revenue_daily_replace.sql",
             do_xcom_push=False,
-            # Airflow 3 применяет этот лимит и к размноженным экземплярам
-            # (сверено через Context7). Каждая замена - отдельный процесс
-            # LocalExecutor; три разом исчерпали память контейнера scheduler,
-            # а самим дням параллельность ничего не дает.
-            max_active_tis_per_dag=1,
         ).expand_kwargs(replacements(days))
 
         rebuild >> replace
@@ -198,8 +190,6 @@ def dm_transform():
             conn_id=CLICKHOUSE_CONNECTION,
             sql="dm/purchase_vs_orders_replace.sql",
             do_xcom_push=False,
-            # Причина единицы разобрана у revenue_daily.replace выше.
-            max_active_tis_per_dag=1,
         ).expand_kwargs(replacements(days))
 
         rebuild >> replace

dags/dq_check.py

@@ -0,0 +1,313 @@
+"""Проверки качества стендовых моделей и витрин.
+
+Пять проверок сначала собирают общий донор. После его полной сборки дневные
+партиции публикуются одной серией замен, и только затем именованные задачи
+утверждают результат. Расхождение поэтому остаётся в ``dm.dq_summary_v`` до
+красного завершения дага; ошибка сборки прежнюю сводку не меняет.
+"""
+
+from __future__ import annotations
+
+import datetime
+import json
+import logging
+from pathlib import Path
+
+from airflow.exceptions import AirflowException
+from airflow.providers.clickhousedb.hooks.clickhouse import ClickHouseHook
+from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator
+from airflow.sdk import Variable, dag, get_current_context, task, task_group
+
+START_DATE = datetime.datetime(2026, 1, 1, tzinfo=datetime.UTC)
+
+SQL_ROOT = Path("/opt/airflow/sql")
+INVENTORY_PATH = Path("/opt/airflow/world/data/world-inventory.json")
+CLICKHOUSE_CONNECTION = "clickhouse_default"
+WORLD_POSITION = "world_position"
+DETAIL_LIMIT = 20
+
+SUMMARY_COLUMNS = [
+    "data_date",
+    "check_name",
+    "reference_rows",
+    "actual_rows",
+    "failed_rows",
+    "status",
+    "_load_id",
+    "_load_ts",
+]
+
+
+def _clickhouse_hook() -> ClickHouseHook:
+    """Подключение, которое дожидается записи в Distributed-таблицу."""
+    return ClickHouseHook(
+        clickhouse_conn_id=CLICKHOUSE_CONNECTION,
+        session_settings={"distributed_foreground_insert": 1},
+    )
+
+
+def _world_position() -> int:
+    """Первый несыгранный день оси мира."""
+    raw_position = Variable.get(WORLD_POSITION, default=None)
+    if raw_position is None:
+        raise AirflowException(
+            "мир еще не создан: выполните make up и дождитесь world_initialize"
+        )
+    try:
+        position = int(raw_position)
+    except (TypeError, ValueError) as error:
+        raise AirflowException(
+            f"world_position={raw_position!r} не является номером дня; "
+            "выполните make rebuild-storage"
+        ) from error
+    if position < 1:
+        raise AirflowException(
+            f"мир не готов: world_position={position}; "
+            "выполните make up или make rebuild-storage"
+        )
+    return position
+
+
+def _publish_summary(
+    check_name: str,
+    summaries: list[tuple[object, ...]],
+) -> list[str]:
+    """Записать дневную сводку и вернуть малый срез готовой диагностики."""
+    if not summaries:
+        raise AirflowException(f"{check_name}: область проверки пуста")
+
+    diagnostics: list[str] = []
+    context = get_current_context()
+    load_id = context["run_id"]
+    load_ts = datetime.datetime.now(datetime.UTC)
+    rows = []
+    for (
+        data_date,
+        reference_rows,
+        actual_rows,
+        failed_rows,
+        details,
+    ) in summaries:
+        rows.append(
+            (
+                data_date,
+                check_name,
+                reference_rows,
+                actual_rows,
+                failed_rows,
+                "pass" if failed_rows == 0 else "fail",
+                load_id,
+                load_ts,
+            )
+        )
+        diagnostics.extend(details[: DETAIL_LIMIT - len(diagnostics)])
+
+    _clickhouse_hook().bulk_insert_rows(
+        "dm.dq_summary_stage_dist",
+        rows,
+        column_names=SUMMARY_COLUMNS,
+    )
+    return diagnostics
+
+
+def _build_sql_check(
+    check_name: str,
+    sql_name: str,
+    parameters: dict[str, object] | None = None,
+) -> list[str]:
+    """Выполнить SQL-сверку и записать ее дневные итоги в донор."""
+    query = (SQL_ROOT / "dq" / sql_name).read_text(encoding="utf-8")
+    summaries = _clickhouse_hook().get_records(query, parameters=parameters)
+    return _publish_summary(check_name, summaries)
+
+
+@task
+def build_sql_check(check_name: str, sql_name: str) -> list[str]:
+    """Собрать проверку, которой не нужны координаты модельного мира."""
+    return _build_sql_check(check_name, sql_name)
+
+
+@task
+def build_sessions_vs_reference() -> list[str]:
+    """Сверить сессии только в пределах прожитых дней мира."""
+    return _build_sql_check(
+        "sessions_vs_reference",
+        "sessions_vs_reference.sql",
+        parameters={"position": _world_position()},
+    )
+
+
+@task
+def build_classes_vs_inventory() -> list[str]:
+    """Сопоставить классы закрытых дней с техническим оракулом генератора."""
+    inventory = json.loads(INVENTORY_PATH.read_text(encoding="utf-8"))
+    expected = {
+        (datetime.date.fromisoformat(day["date"]), mismatch_class): count
+        for day in inventory["days"]
+        if "orders" in day
+        for mismatch_class, count in day["orders"].items()
+    }
+    if not expected:
+        raise AirflowException("classes_vs_inventory: в описи нет закрытых дней")
+
+    query = (SQL_ROOT / "dq" / "classes_vs_inventory.sql").read_text(encoding="utf-8")
+    actual = {
+        (data_date, mismatch_class): count
+        for data_date, mismatch_class, count in _clickhouse_hook().get_records(query)
+        if (data_date, mismatch_class) in expected
+    }
+    summaries = []
+    for data_date in sorted({day for day, _ in expected}):
+        details = []
+        reference_rows = 0
+        actual_rows = 0
+        failed_rows = 0
+        for (day, mismatch_class), reference_count in sorted(expected.items()):
+            if day != data_date:
+                continue
+            actual_key = (day, mismatch_class)
+            actual_count = actual.get(actual_key)
+            reference_rows += 1
+            actual_rows += int(actual_key in actual)
+            if reference_count != actual_count:
+                failed_rows += 1
+                actual_text = actual_count if actual_count is not None else "нет ключа"
+                details.append(
+                    f"{day} key={mismatch_class}: "
+                    f"эталон={reference_count}, "
+                    f"объект={actual_text}"
+                )
+        summaries.append((data_date, reference_rows, actual_rows, failed_rows, details))
+    return _publish_summary("classes_vs_inventory", summaries)
+
+
+@task
+def publication_scope() -> list[dict[str, dict[str, str]]]:
+    """Вернуть дни полностью собранного донора для размноженной замены."""
+    rows = _clickhouse_hook().get_records(
+        "SELECT DISTINCT data_date FROM dm.dq_summary_stage_dist ORDER BY data_date"
+    )
+    if not rows:
+        raise AirflowException("донор dm.dq_summary пуст: публиковать нечего")
+    return [{"params": {"day": data_date.isoformat()}} for (data_date,) in rows]
+
+
+@task
+def drop_stale_partitions() -> None:
+    """Убрать дни, исчезнувшие после починки с обеих сторон сверки."""
+    hook = _clickhouse_hook()
+    stale_days = hook.get_records(
+        """
+        SELECT DISTINCT data_date
+        FROM dm.dq_summary_dist
+        WHERE data_date GLOBAL NOT IN
+            (SELECT data_date FROM dm.dq_summary_stage_dist)
+        ORDER BY data_date
+        """
+    )
+    for (data_date,) in stale_days:
+        # ALTER PARTITION не принимает параметр ClickHouse. Значение пришло
+        # из колонки Date самого хранилища и форматируется без внешнего ввода.
+        hook.run(
+            "ALTER TABLE dm.dq_summary_rep ON CLUSTER clickstream_cluster "
+            f"DROP PARTITION '{data_date.isoformat()}'"
+        )
+
+
+@task
+def assert_check(check_name: str, diagnostics: list[str]) -> None:
+    """Покрасить именованную задачу по уже опубликованной сводке."""
+    failed_days = _clickhouse_hook().get_records(
+        """
+        SELECT data_date, reference_rows, actual_rows, failed_rows
+        FROM dm.dq_summary_v
+        WHERE check_name = {check_name:String} AND failed_rows > 0
+        ORDER BY data_date
+        """,
+        parameters={"check_name": check_name},
+    )
+    if not failed_days:
+        return
+
+    for detail in diagnostics:
+        logging.error("%s: %s", check_name, detail)
+    summary = "; ".join(
+        f"{data_date}: эталонных ключей {reference_rows}, "
+        f"фактических {actual_rows}, нарушено {failed_rows}"
+        for data_date, reference_rows, actual_rows, failed_rows in failed_days
+    )
+    raise AirflowException(f"{check_name}: {summary}")
+
+
+@dag(
+    dag_id="dq_check",
+    schedule=None,
+    start_date=START_DATE,
+    is_paused_upon_creation=False,
+    max_active_runs=1,
+    template_searchpath=str(SQL_ROOT),
+    tags=["dq"],
+)
+def dq_check():
+    """Собрать, опубликовать и утвердить текущее качество данных."""
+    prepare = SQLExecuteQueryOperator(
+        task_id="prepare",
+        conn_id=CLICKHOUSE_CONNECTION,
+        sql="dq/prepare.sql",
+        do_xcom_push=False,
+    )
+
+    sessions = build_sessions_vs_reference()
+    classes = build_classes_vs_inventory.override(
+        task_id="build_classes_vs_inventory"
+    )()
+    revenue = build_sql_check.override(task_id="build_revenue_daily_vs_dds")(
+        "revenue_daily_vs_dds", "revenue_daily_vs_dds.sql"
+    )
+    purchases = build_sql_check.override(task_id="build_purchase_vs_orders_vs_dds")(
+        "purchase_vs_orders_vs_dds", "purchase_vs_orders_vs_dds.sql"
+    )
+    traffic = build_sql_check.override(task_id="build_daily_traffic_vs_dds")(
+        "daily_traffic_vs_dds", "daily_traffic_vs_dds.sql"
+    )
+    builders = [sessions, classes, revenue, purchases, traffic]
+    prepare >> builders
+
+    days = publication_scope()
+    for builder in builders:
+        builder >> days
+
+    replace = SQLExecuteQueryOperator.partial(
+        task_id="publish",
+        conn_id=CLICKHOUSE_CONNECTION,
+        sql="dq/replace.sql",
+        do_xcom_push=False,
+    ).expand_kwargs(days)
+
+    sessions_assertion = assert_check.override(task_id="sessions_vs_reference")(
+        "sessions_vs_reference", sessions
+    )
+    classes_assertion = assert_check.override(task_id="classes_vs_inventory")(
+        "classes_vs_inventory", classes
+    )
+
+    @task_group(group_id="marts_vs_dds")
+    def assert_marts():
+        return [
+            assert_check.override(task_id="revenue_daily_vs_dds")(
+                "revenue_daily_vs_dds", revenue
+            ),
+            assert_check.override(task_id="purchase_vs_orders_vs_dds")(
+                "purchase_vs_orders_vs_dds", purchases
+            ),
+            assert_check.override(task_id="daily_traffic_vs_dds")(
+                "daily_traffic_vs_dds", traffic
+            ),
+        ]
+
+    assertions = [sessions_assertion, classes_assertion, *assert_marts()]
+    remove_stale = drop_stale_partitions()
+    replace >> remove_stale >> assertions
+
+
+dq_check()

dags/etl_pipeline.py

@@ -1,4 +1,4 @@
-"""Дневной конвейер: сначала DDS, затем DM.
+"""Дневной конвейер: DDS, DM и проверка качества.
 
 Своих преобразований у дага нет. Ждущие триггеры связывают состояния:
 если слой краснеет, конвейер не запускает следующий слой и краснеет сам.
@@ -36,8 +36,18 @@ def etl_pipeline():
         wait_for_completion=True,
         poke_interval=10,
     )
+    # Airflow 3.3: оператор провайдера standard с wait_for_completion=True
+    # поднимает AirflowException, если запущенный даг завершился красным.
+    # Поведение и путь импорта повторно сверены через Context7 27 августа
+    # 2026 года.
+    dq = TriggerDagRunOperator(
+        task_id="trigger_dq",
+        trigger_dag_id="dq_check",
+        wait_for_completion=True,
+        poke_interval=10,
+    )
 
-    dds >> dm
+    dds >> dm >> dq
 
 
 etl_pipeline()

docs/adr/0004-resource-limits.md

@@ -19,16 +19,23 @@
 
 Нодам ClickHouse — по 4 ГиБ вместо гигабайта. Дальше в этом тексте предел
 памяти контейнера (`mem_limit`) называется коробкой: сервер живёт внутри неё и
-о самой машине ничего не знает. Остальные `mem_limit` не меняются.
+о самой машине ничего не знает. На 1 августа остальные `mem_limit` не менялись.
 
 Числа, выставленные ради прежнего бюджета, остаются в силе, но лишаются
 статуса решения: `KAFKA_HEAP_OPTS`, `shared_buffers`, `max_connections`,
-`AIRFLOW__CORE__PARALLELISM` и девять неизменённых `mem_limit`. Основания,
+`AIRFLOW__CORE__PARALLELISM` и восемь оставшихся `mem_limit`. Основания,
 кроме отменённого, у них нет. Кто в них упрётся — меняет по первому
 свидетельству и этот ADR не пересматривает. Срок хранения метрик Prometheus
 сюда не входит: неделя истории — осознанный учебный выбор, и цена у него
 дисковая.
 
+27 августа таким пределом впервые стала коробка `airflow-scheduler`:
+LocalExecutor исполняет в ней до четырех процессов задач одновременно, а
+640 МиБ этого не вместили. Коробка поднята до 1,5 ГиБ. Ресурсные ограничения
+из отдельных дагов удалены. Общий предел параллелизма уже задает Airflow.
+`max_active_runs=1` остается там, где два запуска делили бы один донор и могли
+повредить публикацию.
+
 Ресурсный довод ADR 0001 отозван.
 
 ## Почему
@@ -94,8 +101,8 @@ ClickHouse сам понижает каждый до 512 МиБ, а на сте
 заранее, и в покое стенд занимает около 3,2 ГБ. Восемь гигабайт нужны не сумме
 пределов, а работе с запасом.
 
-Остальные девять лимитов не меняются, потому что ни один из них ни разу не
-сработал. Снять их скопом — то же изменение без свидетельств, каким они были
+На 1 августа остальные девять лимитов не менялись, потому что ни один из них
+не сработал. Снять их скопом — то же изменение без свидетельств, каким они были
 выставлены, только в обратную сторону. Лечение здесь не в том, чтобы стереть
 числа, а в том, чтобы лишить их статуса закона: число без основания меняют,
 когда оно мешает, и не защищают как принятое решение.
@@ -151,3 +158,13 @@ to 2.00 GiB because the system has limited RAM` и такие же строки
 Значение `max_server_memory_usage_to_ram_ratio` = 0,9 и то, что в cgroup предел
 считается от коробки, а не от машины, сверены по документации ClickHouse через
 MCP Context7.
+
+`airflow-scheduler` проверен отдельно 27 августа 2026 года. На свежем
+контейнере две попытки запустить четыре DQ-сверки параллельно в коробке
+640 МиБ дошли ровно до `memory.max` и дали `oom_kill=1`: сначала при выгрузке
+57 567 ключей сессий в Python, затем и после переноса дневной агрегации в
+ClickHouse. В коробке 1 ГиБ сам `dq_check` прошел, но снова коснулся верхней
+границы (`memory.events.max=57`). Полный `etl_pipeline` без локальных
+ограничений задач в коробке 1,5 ГиБ завершился успешно: пик - 1 013 760 000
+байт, `memory.events.max=0`, `oom_kill=0`. Docker не резервирует разницу между
+пиком и пределом, поэтому этот запас не увеличивает расход памяти в покое.

docs/architecture/dm.md

@@ -149,31 +149,66 @@ Distributed-таблица вернула бы каждый непарный з
 
 ## `dm.dq_summary`
 
-Зерно — «день данных × проверка»: строка отвечает, каково качество дня
-сейчас, а не сколько раз его проверяли, — история прогонов и так живёт в
-Airflow. Партиция — день данных; прогон звена DQ заменяет партиции
-проверенных дней через `REPLACE PARTITION`. Обещанный спекой перевод с
-`TRUNCATE+INSERT` выполняется той же механикой замены, что и у витрин, —
-ничего отдельного учить не надо.
+Зерно публичного `dm.dq_summary_v` — «день данных × атомарная проверка».
+Строка отвечает, каково качество дня сейчас; историю запусков хранит Airflow.
+Партиция совпадает с днём данных и заменяется через `REPLACE PARTITION`.
 
-## Будущее звено DQ конвейера
+Деловые колонки сводки:
 
-Встанет в `etl_pipeline` после DM ([конвейер](etl.md)) и проверит только
-объекты стенда:
+- `data_date` — проверяемый день;
+- `check_name` — устойчивое имя инварианта;
+- `reference_rows` — число деловых ключей на эталонной стороне;
+- `actual_rows` — число деловых ключей в проверяемом объекте;
+- `failed_rows` — число отсутствующих или различающихся ключей;
+- `status` — `pass` при нуле нарушений, иначе `fail`.
 
-- счётная сверка сессий с эталоном `VisitID` — кука × день, эталон читается
-  из `ods.event_v` ([модель сессии](dds/session.md));
-- счётчики классов расхождений против описи мира — по дням с закрытым
-  окном. Проверка техническая, не урок DQ: её оракул — опись генератора,
-  а у настоящего магазина оракула нет. Обещание узкое и честное: до
-  витрин доехал тот мир, который родился, — одна растяжка через весь
-  путь вместо сторожей на каждом шаге. Верно ли сам мир рождён, она
-  не видит: опись считает тот же генератор;
-- счётные сверки витрин стенда с их DDS-источниками.
+Служебные `_load_id` и `_load_ts` называют собравший запуск. Представление
+перечисляет колонки явно. Порогов и промежуточного состояния нет.
 
-Результаты ложатся в `dm.dq_summary`. Звено — не только сторож конвейера,
-но и экспонат: менти видит, как контроль качества устроен в бою — по
-расписанию, с историей, рядом с трансформациями.
+## Звено DQ конвейера
+
+Даг `dq_check` стоит в `etl_pipeline` после `dm_transform` и проверяет пять
+атомарных договоров:
+
+- `sessions_vs_reference`: ключ — `client_id` внутри дня, эталон —
+  `uniqExact(VisitID)` из `ods.event_v`, объект — число строк
+  `dds.session_v`;
+- `classes_vs_inventory`: ключ — `mismatch_class`, эталон — счётчик из
+  `orders` в описи мира, объект — число заказов класса в
+  `dm.purchase_vs_orders_v`. Проверяются `match`, `cancelled` и
+  `amount_delta` только для покрытых описью закрытых дней;
+- `revenue_daily_vs_dds`: ключ — `product_category` внутри дня. Сравниваются
+  `orders`, `units`, `revenue` и `aov` с прямым агрегатом `dds.order_v`;
+- `purchase_vs_orders_vs_dds`: ключ — `order_id` внутри дня. Все деловые
+  поля сверки сравниваются с независимой сборкой из `dds.event_v` и
+  `dds.order_v`;
+- `daily_traffic_vs_dds`: ключ — день. `visitors` и `known_users`
+  сравниваются с прямыми агрегатами `dds.session_v` и
+  `dds.identity_map_v`.
+
+Область каждой проверки выводится из объединения сторон. Полностью потерянный
+день или ключ поэтому не исчезает из результата. Разные ошибки тоже не могут
+скомпенсировать друг друга общим итогом: `failed_rows` считается после
+сопоставления деловых ключей.
+
+`classes_vs_inventory` — техническая растяжка пути от генератора до DM. Она
+подтверждает, что до витрины доехал записанный в описи мир, но не проверяет
+сам генератор: у настоящего магазина такого оракула нет.
+
+Четыре SQL-сверки считают дневные итоги в ClickHouse и возвращают Airflow
+только сводку и малый срез диагностики. Техническая сверка с описью читает
+небольшой файл JSON в Python. Все пять сборок пишут в общий донор. После их
+успеха даг заменяет проверенные дневные партиции. Затем именованные задачи
+читают опубликованную сводку и падают при `failed_rows > 0`. Расхождение успевает
+сохраниться до красного завершения, а ошибка сборки не затирает прежний
+результат. Красная задача печатает не больше двадцати нарушенных ключей за
+проверку со значениями сторон. Для полного списка итоговую агрегацию SQL-сверки
+заменяют на `SELECT * FROM comparison WHERE failed ORDER BY data_date,
+business_key`.
+
+Если после починки день исчез с обеих сторон, его больше нет и в новом доноре.
+Такой день удаляется из сводки отдельно: старая красная строка не должна
+превращаться в историю попыток.
 
 ## Даг самопроверки
 
@@ -188,12 +223,10 @@ Airflow. Партиция — день данных; прогон звена DQ
 валидационный даг с группами проверок по слоям давал менти обратную связь
 по каждому шагу домашки, не трогая рабочий конвейер.
 
-## Остаётся нарезке
+## Остается нарезке
 
-- Имена звена DQ и дага самопроверки.
-- Ключи шардирования остальных материализованных витрин — вместе с их
-  DDL; правило партиции-донора — в [доке хранилища](storage.md).
-- Тексты учебных заданий и точный список проверок обоих контуров.
+- Имя и состав дага самопроверки домашних заданий.
+- Тексты учебных заданий и точный список проверок учебного контура.
 
 ## Отклонённые варианты
 

docs/architecture/etl.md

@@ -13,8 +13,8 @@
 
 Дневной конвейер на стенде один — даг `etl_pipeline`. Своей работы
 у него нет: он дёргает слойные даги последовательно и ждёт каждый —
-`dds_transform`, затем `dm_transform`. Будущее звено DQ встанет после DM.
-Его состав описан в [доке слоя DM](dm.md).
+`dds_transform`, затем `dm_transform` и `dq_check`. Состав проверок описан в
+[доке слоя DM](dm.md).
 
 Ждущие вызовы сделаны `TriggerDagRunOperator`. Проверка через Context7 по
 ближайшей доступной документации Airflow 3.x (версия 3.1.6) подтвердила, что
@@ -28,7 +28,7 @@
 
 ```text
 world_initialize ─┐
-world_live_day ───┼─> etl_pipeline ─> dds_transform ─> dm_transform
+world_live_day ───┼─> etl_pipeline ─> dds_transform ─> dm_transform ─> dq_check
 запуск рукой ─────┘
 
 world_next_day ─────> источники без ETL
@@ -140,8 +140,8 @@ ODS и ещё не собранный в DDS, затем при необходи
   диапазон, включая форму необязательных параметров того же дага, — этап 5
   (#7); туда же стык #111 про дату модельного мира и `logical_date`.
 - Инкрементальный ETL вместо пересборки — техдолг #2.
-- Состав витрин `dm_transform` и звена DQ — [дока слоя DM](dm.md);
-  имена задач — нарезка этапа 4.
+- Состав витрин `dm_transform` и проверок `dq_check` —
+  [дока слоя DM](dm.md).
 
 ## Отклонённые варианты
 

docs/architecture/storage.md

@@ -606,6 +606,9 @@ SQL-файлы исполняет штатный `SQLExecuteQueryOperator`. Пр
 | `dm` | `dm.purchase_vs_orders_v` | публичный договор сверки с `NULL` для отсутствующей стороны |
 | `dm` | `dm.daily_traffic_rep` / `_dist` | версии дневных счетчиков трафика |
 | `dm` | `dm.daily_traffic_v` | точные текущие счетчики трафика с `FINAL` |
+| `dm` | `dm.dq_summary_rep` / `_dist` | текущее качество дня по атомарным проверкам |
+| `dm` | `dm.dq_summary_stage_rep` / `_dist` | общий донор дневных партиций DQ-сводки |
+| `dm` | `dm.dq_summary_v` | публичная сводка качества дней |
 | `dic` | `dic.products_file` | чтец CSV-каталога, общего с генератором |
 | `dic` | `dic.products` | словарь товаров поверх подложки |
 

docs/specs/2026-07-30-stand-v2-realism.md

@@ -514,8 +514,8 @@ Kafka день переигрывается генератором заново:
 - `events_enriched_v`, `top_pages_daily_v`, `session_overview_v`,
   `dq_errors_daily_v` — переезжают на новую модель без смены роли: источник —
   `dds.event_v`, не `ods.event`.
-- `dm.dq_summary` переводится с TRUNCATE+INSERT на партиционную замену
-  (политика «без TRUNCATE»).
+- `dm.dq_summary` хранит текущее качество дня и обновляется партиционной
+  заменой (политика «без TRUNCATE»).
 
 Дашборд Superset получает три новых сюжета: выручка по дням и категориям,
 таблица сверки с классами расхождений, пара посетители/известные.

sql/ddl/55-dm-tables.sql

@@ -121,3 +121,50 @@ ENGINE = Distributed(
     'daily_traffic_rep',
     cityHash64(report_date)
 );
+
+-- Сводка качества хранит текущее состояние атомарной проверки одного дня.
+-- История запусков остаётся в Airflow, поэтому версия строки не нужна:
+-- следующий успешный прогон заменяет дневную партицию целиком.
+CREATE TABLE IF NOT EXISTS dm.dq_summary_rep ON CLUSTER clickstream_cluster
+(
+    data_date Date,
+    check_name LowCardinality(String),
+    reference_rows UInt64,
+    actual_rows UInt64,
+    failed_rows UInt64,
+    status LowCardinality(String),
+    _load_id String,
+    _load_ts DateTime64(3, 'UTC')
+)
+ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/{database}/{table}', '{replica}')
+PARTITION BY data_date
+ORDER BY (data_date, check_name);
+
+-- Все строки одной проверки лежат на одном шарде. Таблица мала, а ключ
+-- check_name делает раскладку донора и цели одинаковой без нового измерения.
+CREATE TABLE IF NOT EXISTS dm.dq_summary_dist ON CLUSTER clickstream_cluster
+AS dm.dq_summary_rep
+ENGINE = Distributed(
+    'clickstream_cluster',
+    'dm',
+    'dq_summary_rep',
+    cityHash64(check_name)
+);
+
+-- Источник и цель REPLACE PARTITION обязаны совпадать по структуре, ключам и
+-- политике хранения. Требование повторно сверено по документации ClickHouse
+-- через Context7 27 августа 2026 года; AS не даёт формам разойтись.
+CREATE TABLE IF NOT EXISTS dm.dq_summary_stage_rep ON CLUSTER clickstream_cluster
+AS dm.dq_summary_rep
+ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/{database}/{table}', '{replica}')
+PARTITION BY data_date
+ORDER BY (data_date, check_name);
+
+CREATE TABLE IF NOT EXISTS dm.dq_summary_stage_dist ON CLUSTER clickstream_cluster
+AS dm.dq_summary_stage_rep
+ENGINE = Distributed(
+    'clickstream_cluster',
+    'dm',
+    'dq_summary_stage_rep',
+    cityHash64(check_name)
+);

sql/ddl/60-dm-views.sql

@@ -42,3 +42,18 @@ SELECT
     _load_id,
     _load_ts
 FROM dm.daily_traffic_dist FINAL;
+
+-- Сводка отвечает, какие атомарные договоры дня выполняются сейчас.
+-- Явный список не пропускает в публичный договор будущие физические колонки.
+CREATE VIEW IF NOT EXISTS dm.dq_summary_v ON CLUSTER clickstream_cluster
+AS
+SELECT
+    data_date,
+    check_name,
+    reference_rows,
+    actual_rows,
+    failed_rows,
+    status,
+    _load_id,
+    _load_ts
+FROM dm.dq_summary_dist;

sql/dq/classes_vs_inventory.sql

@@ -0,0 +1,17 @@
+-- DQ: стендовая сторона технической растяжки от описи мира.
+--
+-- Эталонные счётчики читает Airflow из world-inventory.json: у настоящего
+-- магазина такого оракула нет. Запрос возвращает только наблюдаемую сторону;
+-- Python-задача сопоставляет ее с тремя счетчиками закрытого дня из описи.
+SELECT
+    order_day AS data_date,
+    mismatch_class,
+    count() AS orders
+FROM dm.purchase_vs_orders_v
+WHERE mismatch_class IN ('match', 'cancelled', 'amount_delta')
+GROUP BY
+    data_date,
+    mismatch_class
+ORDER BY
+    data_date,
+    mismatch_class

sql/dq/daily_traffic_vs_dds.sql

@@ -0,0 +1,71 @@
+-- DQ: дневной трафик против прямого агрегата сессий и карты идентичностей.
+--
+-- Публичные представления скрывают физическую ко-локацию, поэтому GLOBAL
+-- назван явно, как в самой витрине. Полное объединение не прячет день,
+-- исчезнувший с любой стороны.
+WITH
+    reference AS
+    (
+        SELECT
+            sessions.session_date AS report_date,
+            uniqExact(sessions.client_id) AS visitors,
+            uniqExactIf(identities.user_id, identities.user_id != 0) AS known_users
+        FROM dds.session_v AS sessions
+        GLOBAL LEFT JOIN dds.identity_map_v AS identities USING (client_id)
+        GROUP BY report_date
+    ),
+    actual AS
+    (
+        SELECT
+            report_date,
+            count() AS physical_rows,
+            any(visitors) AS visitors,
+            any(known_users) AS known_users
+        FROM dm.daily_traffic_v
+        GROUP BY report_date
+    ),
+    comparison AS
+    (
+        SELECT
+            coalesce(reference.report_date, actual.report_date) AS data_date,
+            toString(coalesce(reference.report_date, actual.report_date))
+                AS business_key,
+            isNotNull(reference.report_date) AS reference_present,
+            isNotNull(actual.report_date) AS actual_present,
+            toString(tuple(reference.visitors, reference.known_users))
+                AS reference_value,
+            concat(
+                toString(tuple(actual.visitors, actual.known_users)),
+                ', rows=',
+                toString(actual.physical_rows)
+            ) AS actual_value,
+            isNull(reference.report_date)
+                OR isNull(actual.report_date)
+                OR actual.physical_rows != 1
+                OR NOT isNotDistinctFrom(reference.visitors, actual.visitors)
+                OR NOT isNotDistinctFrom(reference.known_users, actual.known_users)
+                AS failed
+        FROM reference
+        GLOBAL FULL OUTER JOIN actual USING (report_date)
+    )
+SELECT
+    data_date,
+    countIf(reference_present) AS reference_rows,
+    countIf(actual_present) AS actual_rows,
+    countIf(failed) AS failed_rows,
+    groupArrayIf(20)(
+        concat(
+            toString(data_date),
+            ' key=',
+            business_key,
+            ': эталон=',
+            if(reference_present, reference_value, 'нет ключа'),
+            ', объект=',
+            if(actual_present, actual_value, 'нет ключа')
+        ),
+        failed
+    ) AS diagnostics
+FROM comparison
+GROUP BY data_date
+ORDER BY data_date
+SETTINGS join_use_nulls = 1

sql/dq/prepare.sql

@@ -0,0 +1,7 @@
+-- DQ: общий донор очищается перед сборкой всех проверок.
+--
+-- Если одна сборка упадет, публикация не начнется. Следующий запуск удалит
+-- неполный результат здесь и соберет донор заново.
+
+ALTER TABLE dm.dq_summary_stage_rep ON CLUSTER clickstream_cluster
+DROP PARTITION ALL;

sql/dq/purchase_vs_orders_vs_dds.sql

@@ -0,0 +1,120 @@
+-- DQ: строка сверки покупок и заказов против независимой сборки из DDS.
+--
+-- Проверяются все деловые поля. Счёт физических строк отдельно ловит дубль
+-- ключа, даже если обе копии несут одинаковые значения.
+WITH
+    purchases AS
+    (
+        SELECT
+            purchase_id[1] AS purchase_order_id,
+            any(event_date) AS purchase_day,
+            any(purchase_revenue[1]) AS declared_revenue,
+            count() AS purchase_count
+        FROM dds.event_v
+        WHERE event_type = 'purchase'
+        GROUP BY purchase_order_id
+    ),
+    orders AS
+    (
+        SELECT
+            order_id AS backend_order_id,
+            order_date,
+            items_total,
+            status
+        FROM dds.order_v
+    ),
+    reference AS
+    (
+        SELECT
+            coalesce(order_date, purchase_day) AS order_day,
+            coalesce(purchase_order_id, backend_order_id) AS order_id,
+            declared_revenue,
+            items_total,
+            status,
+            multiIf(
+                backend_order_id IS NULL, 'awaiting_order',
+                status = 'cancelled', 'cancelled',
+                purchase_order_id IS NULL, 'lost_event',
+                purchase_count > 1, 'duplicate_event',
+                declared_revenue != items_total, 'amount_delta',
+                'match'
+            ) AS mismatch_class
+        FROM purchases
+        GLOBAL FULL OUTER JOIN orders
+            ON purchase_order_id = backend_order_id
+    ),
+    actual AS
+    (
+        SELECT
+            order_day,
+            order_id,
+            count() AS physical_rows,
+            any(declared_revenue) AS declared_revenue,
+            any(items_total) AS items_total,
+            any(status) AS status,
+            any(mismatch_class) AS mismatch_class
+        FROM dm.purchase_vs_orders_v
+        GROUP BY
+            order_day,
+            order_id
+    ),
+    comparison AS
+    (
+        SELECT
+            coalesce(reference.order_day, actual.order_day) AS data_date,
+            coalesce(reference.order_id, actual.order_id) AS business_key,
+            isNotNull(reference.order_id) AS reference_present,
+            isNotNull(actual.order_id) AS actual_present,
+            toString(tuple(
+                reference.declared_revenue,
+                reference.items_total,
+                reference.status,
+                reference.mismatch_class
+            )) AS reference_value,
+            concat(
+                toString(tuple(
+                    actual.declared_revenue,
+                    actual.items_total,
+                    actual.status,
+                    actual.mismatch_class
+                )),
+                ', rows=',
+                toString(actual.physical_rows)
+            ) AS actual_value,
+            isNull(reference.order_id)
+                OR isNull(actual.order_id)
+                OR actual.physical_rows != 1
+                OR NOT isNotDistinctFrom(
+                    reference.declared_revenue,
+                    actual.declared_revenue
+                )
+                OR NOT isNotDistinctFrom(reference.items_total, actual.items_total)
+                OR NOT isNotDistinctFrom(reference.status, actual.status)
+                OR NOT isNotDistinctFrom(
+                    reference.mismatch_class,
+                    actual.mismatch_class
+                ) AS failed
+        FROM reference
+        GLOBAL FULL OUTER JOIN actual USING (order_day, order_id)
+    )
+SELECT
+    data_date,
+    countIf(reference_present) AS reference_rows,
+    countIf(actual_present) AS actual_rows,
+    countIf(failed) AS failed_rows,
+    groupArrayIf(20)(
+        concat(
+            toString(data_date),
+            ' key=',
+            business_key,
+            ': эталон=',
+            if(reference_present, reference_value, 'нет ключа'),
+            ', объект=',
+            if(actual_present, actual_value, 'нет ключа')
+        ),
+        failed
+    ) AS diagnostics
+FROM comparison
+GROUP BY data_date
+ORDER BY data_date
+SETTINGS join_use_nulls = 1

sql/dq/replace.sql

@@ -0,0 +1,7 @@
+-- DQ: публикация одного полностью собранного дня общей сводки.
+--
+-- День приходит из самого донора после успеха всех пяти сборок. REPLACE
+-- PARTITION работает с локальными таблицами одинаковой структуры; это
+-- ограничение сверено по документации ClickHouse 27 августа 2026 года.
+ALTER TABLE dm.dq_summary_rep ON CLUSTER clickstream_cluster
+REPLACE PARTITION '{{ params.day }}' FROM dm.dq_summary_stage_rep;

sql/dq/revenue_daily_vs_dds.sql

@@ -0,0 +1,107 @@
+-- DQ: дневная выручка против прямого агрегата заказов DDS.
+--
+-- День берётся из объединения сторон. Поэтому исчезнувший целиком день или
+-- категория остаются видимы. Денежный расчет повторяет публичный договор
+-- витрины и сохраняет Decimal до округления AOV.
+WITH
+    reference AS
+    (
+        SELECT
+            report_date,
+            product_category,
+            orders,
+            units,
+            CAST(revenue, 'Decimal(18, 2)') AS revenue,
+            CAST(
+                round(divideDecimal(revenue, toDecimal128(orders, 0), 6), 2),
+                'Decimal(18, 2)'
+            ) AS aov
+        FROM
+        (
+            SELECT
+                order_date AS report_date,
+                dictGet('dic.products', 'category', item_sku) AS product_category,
+                uniqExact(order_id) AS orders,
+                sum(item_quantity) AS units,
+                sum(item_price * item_quantity) AS revenue
+            FROM dds.order_v
+            ARRAY JOIN
+                item_sku,
+                item_quantity,
+                item_price
+            WHERE status = 'paid'
+            GROUP BY
+                report_date,
+                product_category
+        )
+    ),
+    actual AS
+    (
+        SELECT
+            report_date,
+            product_category,
+            count() AS physical_rows,
+            any(orders) AS orders,
+            any(units) AS units,
+            any(revenue) AS revenue,
+            any(aov) AS aov
+        FROM dm.revenue_daily_v
+        GROUP BY
+            report_date,
+            product_category
+    ),
+    comparison AS
+    (
+        SELECT
+            coalesce(reference.report_date, actual.report_date) AS data_date,
+            coalesce(reference.product_category, actual.product_category)
+                AS business_key,
+            isNotNull(reference.product_category) AS reference_present,
+            isNotNull(actual.product_category) AS actual_present,
+            toString(tuple(
+                reference.orders,
+                reference.units,
+                reference.revenue,
+                reference.aov
+            )) AS reference_value,
+            concat(
+                toString(tuple(
+                    actual.orders,
+                    actual.units,
+                    actual.revenue,
+                    actual.aov
+                )),
+                ', rows=',
+                toString(actual.physical_rows)
+            ) AS actual_value,
+            isNull(reference.product_category)
+                OR isNull(actual.product_category)
+                OR actual.physical_rows != 1
+                OR NOT isNotDistinctFrom(reference.orders, actual.orders)
+                OR NOT isNotDistinctFrom(reference.units, actual.units)
+                OR NOT isNotDistinctFrom(reference.revenue, actual.revenue)
+                OR NOT isNotDistinctFrom(reference.aov, actual.aov) AS failed
+        FROM reference
+        GLOBAL FULL OUTER JOIN actual USING (report_date, product_category)
+    )
+SELECT
+    data_date,
+    countIf(reference_present) AS reference_rows,
+    countIf(actual_present) AS actual_rows,
+    countIf(failed) AS failed_rows,
+    groupArrayIf(20)(
+        concat(
+            toString(data_date),
+            ' key=',
+            business_key,
+            ': эталон=',
+            if(reference_present, reference_value, 'нет ключа'),
+            ', объект=',
+            if(actual_present, actual_value, 'нет ключа')
+        ),
+        failed
+    ) AS diagnostics
+FROM comparison
+GROUP BY data_date
+ORDER BY data_date
+SETTINGS join_use_nulls = 1

sql/dq/sessions_vs_reference.sql

@@ -0,0 +1,65 @@
+-- DQ: эталонная сессия Метрики против сессии, нарезанной слоем DDS.
+--
+-- Сопоставление идет по куке внутри завершенного дня: лишняя сессия одной
+-- куки не скрывает потерянную сессию другой. Результат содержит итог дня.
+WITH
+    (SELECT min(EventDate) FROM ods.event_v) AS first_source_day,
+    first_source_day + ({position:UInt16} - 1) AS last_lived_day,
+    reference AS
+    (
+        SELECT
+            EventDate AS data_date,
+            ClientID AS client_id,
+            uniqExact(VisitID) AS sessions
+        FROM ods.event_v
+        WHERE EventDate <= last_lived_day
+        GROUP BY
+            data_date,
+            client_id
+    ),
+    actual AS
+    (
+        SELECT
+            session_date AS data_date,
+            client_id,
+            count() AS sessions
+        FROM dds.session_v
+        WHERE session_date <= last_lived_day
+        GROUP BY
+            data_date,
+            client_id
+    ),
+    comparison AS
+    (
+        SELECT
+            coalesce(reference.data_date, actual.data_date) AS data_date,
+            toString(coalesce(reference.client_id, actual.client_id)) AS business_key,
+            isNotNull(reference.client_id) AS reference_present,
+            isNotNull(actual.client_id) AS actual_present,
+            toString(reference.sessions) AS reference_value,
+            toString(actual.sessions) AS actual_value,
+            NOT isNotDistinctFrom(reference.sessions, actual.sessions) AS failed
+        FROM reference
+        GLOBAL FULL OUTER JOIN actual USING (data_date, client_id)
+    )
+SELECT
+    data_date,
+    countIf(reference_present) AS reference_rows,
+    countIf(actual_present) AS actual_rows,
+    countIf(failed) AS failed_rows,
+    groupArrayIf(20)(
+        concat(
+            toString(data_date),
+            ' key=',
+            business_key,
+            ': эталон=',
+            if(reference_present, reference_value, 'нет ключа'),
+            ', объект=',
+            if(actual_present, actual_value, 'нет ключа')
+        ),
+        failed
+    ) AS diagnostics
+FROM comparison
+GROUP BY data_date
+ORDER BY data_date
+SETTINGS join_use_nulls = 1