feat(dq): добавлен контроль качества конвейера

- Зачем: - конвейеру нужна публикуемая сводка текущего качества с красным завершением при нарушении договоров. - Что: - добавлены DDL, публичное представление и пять сверок на естественном деловом зерне. - даг dq_check встроен после DM и публикует полные дневные партиции до именованных утверждений. - параллелизм DDS, DM и DQ приведён к измеренному лимиту памяти планировщика. - Проверка: - make lint; make config-test; make rebuild-storage. - make smoke; make check-clickhouse; make check-services. - три семейства проверок покрашены искусственными расхождениями и восстановлены повторным полным прогоном.
2026-08-27 07:55:28 -04:00
1 parent 4e60aafc31
commit 26b2e5fea1
@@ -355,6 +355,10 @@ def dds_transform():
            conn_id=CLICKHOUSE_CONNECTION,
            sql="dds/order_replace.sql",
            do_xcom_push=False,
+            # Замены не зависят друг от друга, но каждая — отдельный процесс
+            # LocalExecutor. Три разом получили SIGKILL при лимите scheduler
+            # 640 МиБ; последовательность не меняет деловой результат.
+            max_active_tis_per_dag=1,
        ).expand_kwargs(replacements(days))

        rebuild >> replace
@@ -416,6 +420,8 @@ def dds_transform():
            conn_id=CLICKHOUSE_CONNECTION,
            sql="dds/session_replace.sql",
            do_xcom_push=False,
+            # Причина единицы разобрана у order.replace выше.
+            max_active_tis_per_dag=1,
        ).expand_kwargs(replacements(days))

        rebuild >> replace
@@ -76,9 +76,10 @@ 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,
+    # Каждая задача LocalExecutor — отдельный процесс. После появления
+    # dq_check три одновременные сборки DM получили SIGKILL при лимите
+    # scheduler 640 МиБ; две оставляют запас и сохраняют параллельность.
+    max_active_tasks=2,
    template_searchpath=str(SQL_ROOT),
    tags=["dm"],
 )
@@ -0,0 +1,314 @@
+"""Проверки качества стендовых моделей и витрин.
+
+Пять проверок сначала собирают общий донор. После его полной сборки дневные
+партиции публикуются одной серией замен, и только затем именованные задачи
+утверждают результат. Расхождение поэтому остаётся в ``dm.dq_summary_v`` до
+красного завершения дага; ошибка сборки прежнюю сводку не меняет.
+"""
+
+from __future__ import annotations
+
+import datetime
+import json
+import logging
+from collections import defaultdict
+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 _write_summary(
+    check_name: str,
+    comparisons: list[tuple[object, ...]],
+) -> list[str]:
+    """Свернуть строки деловых ключей в дневную сводку и малый срез ошибок."""
+    if not comparisons:
+        raise AirflowException(f"{check_name}: область проверки пуста")
+
+    totals: dict[datetime.date, list[int]] = defaultdict(lambda: [0, 0, 0])
+    diagnostics: list[str] = []
+    for (
+        data_date,
+        business_key,
+        reference_present,
+        actual_present,
+        reference_value,
+        actual_value,
+        failed,
+    ) in comparisons:
+        counts = totals[data_date]
+        counts[0] += int(reference_present)
+        counts[1] += int(actual_present)
+        counts[2] += int(failed)
+        if failed and len(diagnostics) < DETAIL_LIMIT:
+            reference_text = reference_value if reference_present else "нет ключа"
+            actual_text = actual_value if actual_present else "нет ключа"
+            diagnostics.append(
+                f"{data_date} key={business_key}: "
+                f"эталон={reference_text}, объект={actual_text}"
+            )
+
+    context = get_current_context()
+    load_id = context["run_id"]
+    load_ts = datetime.datetime.now(datetime.UTC)
+    rows = [
+        (
+            data_date,
+            check_name,
+            reference_rows,
+            actual_rows,
+            failed_rows,
+            "pass" if failed_rows == 0 else "fail",
+            load_id,
+            load_ts,
+        )
+        for data_date, (reference_rows, actual_rows, failed_rows) in sorted(
+            totals.items()
+        )
+    ]
+    _clickhouse_hook().bulk_insert_rows(
+        "dm.dq_summary_stage_dist",
+        rows,
+        column_names=SUMMARY_COLUMNS,
+    )
+    return diagnostics
+
+
+@task
+def build_sql_check(check_name: str, sql_name: str) -> list[str]:
+    """Выполнить одну SQL-сверку и записать ее дневные итоги в донор."""
+    query = (SQL_ROOT / "dq" / sql_name).read_text(encoding="utf-8")
+    parameters = (
+        {"position": _world_position()}
+        if check_name == "sessions_vs_reference"
+        else None
+    )
+    comparisons = _clickhouse_hook().get_records(query, parameters=parameters)
+    return _write_summary(check_name, comparisons)
+
+
+@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
+    }
+    comparisons = [
+        (
+            data_date,
+            mismatch_class,
+            1,
+            int((data_date, mismatch_class) in actual),
+            reference_count,
+            actual.get((data_date, mismatch_class)),
+            reference_count != actual.get((data_date, mismatch_class)),
+        )
+        for (data_date, mismatch_class), reference_count in sorted(expected.items())
+    ]
+    return _write_summary("classes_vs_inventory", comparisons)
+
+
+@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,
+    # Три одновременные сверки на живом мире исчерпали 640 МиБ scheduler и
+    # получили SIGKILL. Последовательность здесь дешева: весь даг укладывается
+    # в один учебный прогон, а общему донору параллельность ничего не даёт.
+    max_active_tasks=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_sql_check.override(task_id="build_sessions_vs_reference")(
+        "sessions_vs_reference", "sessions_vs_reference.sql"
+    )
+    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,
+        # Каждая замена — отдельный процесс LocalExecutor. Последовательность
+        # держит даг в измеренном лимите памяти планировщика.
+        max_active_tis_per_dag=1,
+    ).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()
@@ -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()
@@ -149,31 +149,58 @@ 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. Она
+подтверждает, что до витрины доехал записанный в описи мир, но не проверяет
+сам генератор: у настоящего магазина такого оракула нет.
+
+Пять сборок пишут в общий донор. Только после их успеха даг заменяет все
+проверенные дневные партиции, затем именованные задачи читают опубликованную
+сводку и падают при `failed_rows > 0`. Расхождение успевает сохраниться до
+красного завершения, а ошибка сборки не затирает прежний результат. Красная
+задача печатает не больше двадцати нарушенных ключей со значениями сторон;
+полный список воспроизводится запросом проверки.

 ## Даг самопроверки

@@ -188,12 +215,10 @@ Airflow. Партиция — день данных; прогон звена DQ
 валидационный даг с группами проверок по слоям давал менти обратную связь
 по каждому шагу домашки, не трогая рабочий конвейер.

-## Остаётся нарезке
+## Остается нарезке

- Имена звена DQ и дага самопроверки.
- Ключи шардирования остальных материализованных витрин — вместе с их
-  DDL; правило партиции-донора — в [доке хранилища](storage.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).

 ## Отклонённые варианты

@@ -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` | словарь товаров поверх подложки |

@@ -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 получает три новых сюжета: выручка по дням и категориям,
 таблица сверки с классами расхождений, пара посетители/известные.
@@ -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)
+);
@@ -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;
@@ -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
@@ -0,0 +1,46 @@
+-- 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
+    )
+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)
+ORDER BY data_date
+SETTINGS join_use_nulls = 1
@@ -0,0 +1,7 @@
+-- DQ: общий донор очищается перед сборкой всех проверок.
+--
+-- Если одна сборка упадет, публикация не начнется. Следующий запуск удалит
+-- неполный результат здесь и соберет донор заново.
+
+ALTER TABLE dm.dq_summary_stage_rep ON CLUSTER clickstream_cluster
+DROP PARTITION ALL;
@@ -0,0 +1,95 @@
+-- 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
+    )
+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)
+ORDER BY
+    data_date,
+    business_key
+SETTINGS join_use_nulls = 1
@@ -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;
@@ -0,0 +1,77 @@
+-- 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
+    )
+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)
+ORDER BY
+    data_date,
+    business_key
+SETTINGS join_use_nulls = 1
@@ -0,0 +1,45 @@
+-- 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
+    )
+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)
+ORDER BY
+    data_date,
+    business_key
+SETTINGS join_use_nulls = 1