dags/dm_transform.py

@@ -1,8 +1,9 @@
 """Трансформации слоя DM: сборка стендовых витрин.
 
-Выручка наследует ритм заказов и заменяет дневные партиции изменяемого окна.
-Трафик наследует ретроспективность карты идентичностей и пересчитывает всю
-историю новой версией. Объем каждого прогона выводится из состояния слоев.
+Выручка наследует ритм заказов. Сверка соединяет окно заказов с прожитым
+хвостом событий. Обе заменяют дневные партиции. Трафик наследует
+ретроспективность карты идентичностей и пересчитывает всю историю новой
+версией. Объем каждого прогона выводится из состояния слоев.
 """
 
 from __future__ import annotations
@@ -14,12 +15,41 @@ 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 dag, task, task_group
+from airflow.sdk import Variable, dag, task, task_group
 
 START_DATE = datetime.datetime(2026, 1, 1, tzinfo=datetime.UTC)
 
 SQL_ROOT = Path("/opt/airflow/sql")
 CLICKHOUSE_CONNECTION = "clickhouse_default"
+# Та же ось модельного времени, что у DDS. Общего модуля у дагов нет
+# намеренно: связь держит переменная Airflow, а не импорт одного дага другим.
+WORLD_POSITION = "world_position"
+
+
+def _world_position() -> int:
+    """Позиция мира для границы прожитых событий сверки.
+
+    Причина отдельного чтения переменной разобрана у одноименной функции в
+    ``dds_transform.py``.
+    """
+    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 _days(first_day: datetime.date, last_day: datetime.date) -> list[str]:
@@ -46,6 +76,9 @@ 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"],
 )
@@ -114,6 +147,65 @@ def dm_transform():
 
     revenue_daily()
 
+    @task_group(group_id="purchase_vs_orders")
+    def purchase_vs_orders():
+        """Сверка: окно заказов, прожитый хвост и предварительные дни."""
+
+        @task
+        def scope() -> dict[str, object]:
+            """Спросить у обоих источников, какие дни пересобрать."""
+            query = (SQL_ROOT / "dm" / "purchase_vs_orders_scope.sql").read_text(
+                encoding="utf-8"
+            )
+            hook = ClickHouseHook(clickhouse_conn_id=CLICKHOUSE_CONNECTION)
+            row = hook.get_first(query, parameters={"position": _world_position()})
+            if row is None:
+                raise AirflowException(
+                    "нет принятого слепка заказов или прожитых покупок; "
+                    "дождитесь источников и сначала выполните dds_transform"
+                )
+            first_day, last_day = row
+            days = _days(first_day, last_day)
+            logging.info(
+                "объем пересборки: %s … %s, дней %d",
+                first_day,
+                last_day,
+                len(days),
+            )
+            return {
+                "first_day": first_day.isoformat(),
+                "last_day": last_day.isoformat(),
+                "days": days,
+            }
+
+        days = scope()
+
+        rebuild = SQLExecuteQueryOperator(
+            task_id="rebuild",
+            conn_id=CLICKHOUSE_CONNECTION,
+            sql="dm/purchase_vs_orders_rebuild.sql",
+            parameters={
+                "load_id": "{{ run_id }}",
+                "first_day": days["first_day"],
+                "last_day": days["last_day"],
+            },
+            split_statements=True,
+            do_xcom_push=False,
+        )
+
+        replace = SQLExecuteQueryOperator.partial(
+            task_id="replace",
+            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
+
+    purchase_vs_orders()
+
     SQLExecuteQueryOperator(
         task_id="daily_traffic",
         conn_id=CLICKHOUSE_CONNECTION,

docs/architecture/dds/event.md

@@ -45,6 +45,11 @@
 своё имя здесь создало бы третий синоним. `client_timezone` тоже остаётся —
 это атрибут посетителя, а не линза времени.
 
+`purchase_revenue` сохраняет смысл поля источника. Тип денег нормализуется в
+ODS; точный тип и проверка описаны в разделе
+[«Актуальные события»](../storage.md#актуальные-события). DDS только
+переименовывает уже нормализованную колонку.
+
 ## Расшифровки
 
 Одна: `device_category` — коды 1–4 в строки `desktop` / `mobile` /

docs/architecture/dm.md

@@ -70,9 +70,17 @@
 знать, какие связки новые, а полный пересчёт карты этот след затирает —
 восстановление следа оставлено лабе повышенной сложности (issue #149).
 
+У `purchase_vs_orders` два источника и одна непрерывная пересборка. Заказная
+сторона задает изменяемое окно последним принятым слепком ODS, событийная —
+последний прожитый день с `purchase`. Левая граница возвращает первую дыру и
+самый ранний день с `awaiting_order`: предварительная строка уже есть в
+витрине, поэтому поиск отсутствующего дня ее не заметит. Правая граница
+доходит до более позднего из двух хвостов. Так один прогон догоняет пакетный
+разгон длиннее K и не оставляет старое ожидание замороженным.
+
 ## Витрины стенда
 
-Оба публичных представления перечисляют деловые колонки и координаты запуска
+Публичные представления перечисляют деловые колонки и координаты запуска
 `_load_id`, `_load_ts` явно. Новая колонка физической таблицы сама в договор не
 попадет.
 
@@ -98,6 +106,29 @@
 шесть категорий стартового мира близки по объему. Более сложный составной ключ
 распределил бы эту маленькую таблицу без нового урока.
 
+### `dm.purchase_vs_orders_v`
+
+Зерно — заказ. Покупки из `dds.event_v` сначала сворачиваются по
+`purchase_id[1]`: контракт источника обещает один номер и одну объявленную
+сумму, а отдельный счет событий оставляет будущий класс `duplicate_event`
+наблюдаемым. Заказы приходят из `dds.order_v`. Обе стороны записаны
+подзапросами, затем соединены `FULL OUTER GLOBAL JOIN`: голая левая
+Distributed-таблица вернула бы каждый непарный заказ по разу с каждого шарда
+([исследование](../research/2026-08-22-distributed-join.md)).
+
+Непарная сторона остается `NULL` благодаря `join_use_nulls = 1`. При
+`join_use_nulls = 0` ClickHouse заполнил бы ее нулем или пустой строкой и скрыл
+отсутствие источника. Клиентская `declared_revenue` и серверная `items_total`
+сравниваются как `Decimal(18, 2)`. Классы проверяются по порядку:
+`awaiting_order` для покупки без заказа, затем `cancelled`, `lost_event`,
+`duplicate_event`, `amount_delta`, `match`. Поэтому отмененный заказ без
+события остается отмененным, а не потерянным.
+
+Витрина нарезана по `order_day`, цель и донор шардируются по
+`cityHash64(order_id)`. Диапазон собирается в доноре одной вставкой и
+заменяется по дням. Первичная сборка охватывает всю историю обоих источников;
+следующие прогоны идут по двустороннему ритму, описанному выше.
+
 ### `dm.daily_traffic_v`
 
 В строке один день. `visitors` считает разные `client_id` из
@@ -163,7 +194,6 @@ Airflow. Партиция — день данных; прогон звена DQ
 - Ключи шардирования остальных материализованных витрин — вместе с их
   DDL; правило партиции-донора — в [доке хранилища](storage.md).
 - Тексты учебных заданий и точный список проверок обоих контуров.
-- Механика сверки A+C в SQL — рамка в спеке, разделы 4 и 7.
 
 ## Отклонённые варианты
 

docs/architecture/storage.md

@@ -221,6 +221,13 @@ ODS и в таблицу ошибок. Нужен ли он выше ODS, реш
 `dds.event_v` отвечает за другое: даёт событию язык бизнес-модели, в том
 числе snake_case-имена и расшифровку кодов.
 
+Типы ODS обычно повторяют контракт выгрузки. Денежный `purchaseRevenue` —
+исключение: источник объявляет `Array(Float64)`, а материализованное
+представление ODS разбирает числа сырого JSON сразу в
+`Array(Decimal(18, 2))`. DDS наследует этот тип без нового преобразования.
+Прямой разбор подтвержден документацией ClickHouse через Context7 и запросом
+к ClickHouse 26.3; сравнение с сырьем подтвердило: копейки всех покупок целы.
+
 ## Версии заказов
 
 `ods.order_rep` хранит принятые версии в
@@ -483,9 +490,9 @@ kafka_offset)`: смотрят такую таблицу от класса, а
 административная операция, роль `etl` права на неё не получает.
 
 Деньги каталога живут в трёх единицах: в CSV — целые копейки, словарь отдаёт
-`Decimal(18, 2)`, `productPrice` события — целые рубли. Разрыв намеренный, на
-нём стоит урок про `Float64` ([описание
-выгрузки](../formats/clickstream-event.md)).
+`Decimal(18, 2)`, `productPrice` события — целые рубли. Поэтому точную сумму
+заказа берут из `purchaseRevenue`, а не складывают округленные цены товаров
+([описание выгрузки](../formats/clickstream-event.md)).
 
 Устройство обоих объектов — почему такой движок, такая форма пути, такой
 пользователь и такое окно — расписано в
@@ -523,7 +530,7 @@ SQL-файлы исполняет штатный `SQLExecuteQueryOperator`. Пр
 | `40-stg-views.sql` | матвью приёма: чтец в сырьё |
 | `45-dds-tables.sql` | физические модели заказа, сессии и карты идентичностей, доноры партиций |
 | `50-dds-views.sql` | публичные договоры сущностей DDS |
-| `55-dm-tables.sql` | физические витрины выручки и трафика, донор партиций выручки |
+| `55-dm-tables.sql` | физические витрины выручки, сверки и трафика, доноры дневных партиций |
 | `60-dm-views.sql` | публичные договоры витрин DM |
 
 Справочники идут сразу за базами: зона `dic` не зависит ни от одного слоя, и
@@ -594,6 +601,9 @@ SQL-файлы исполняет штатный `SQLExecuteQueryOperator`. Пр
 | `dm` | `dm.revenue_daily_rep` / `_dist` | дневная выручка по категориям |
 | `dm` | `dm.revenue_daily_stage_rep` / `_dist` | таблица-двойник, донор дневных партиций выручки |
 | `dm` | `dm.revenue_daily_v` | публичный договор дневной выручки |
+| `dm` | `dm.purchase_vs_orders_rep` / `_dist` | сверка покупок с заказами, партиция — день заказа |
+| `dm` | `dm.purchase_vs_orders_stage_rep` / `_dist` | донор дневных партиций сверки |
+| `dm` | `dm.purchase_vs_orders_v` | публичный договор сверки с `NULL` для отсутствующей стороны |
 | `dm` | `dm.daily_traffic_rep` / `_dist` | версии дневных счетчиков трафика |
 | `dm` | `dm.daily_traffic_v` | точные текущие счетчики трафика с `FINAL` |
 | `dic` | `dic.products_file` | чтец CSV-каталога, общего с генератором |
@@ -605,8 +615,8 @@ SQL-файлы исполняет штатный `SQLExecuteQueryOperator`. Пр
 
 В слое DDS построены событие, заказ, сессия и карта идентичностей. Событие
 остаётся представлением над ODS; заказ и сессия гасят повтор заменой партиции;
-карта пишет полный набор пар в версионную таблицу. В DM выручка заменяет
-партиции изменяемых дней, а трафик пишет полную историю новой версией.
+карта пишет полный набор пар в версионную таблицу. В DM выручка и сверка
+заменяют партиции изменяемых дней, а трафик пишет полную историю новой версией.
 
 ## Что проверено
 
@@ -849,6 +859,10 @@ Kafka с пустым значением (ноль байт) и запись-н
   промежуточного `toDecimal64`. Проверено на живом слепке; честность разбора
   сторожит инвариант бэкенда — сумма «цена × количество» совпала с `items_total`
   у всех 1662 заказов.
+- **`JSONExtract` читает `Decimal` и из числа JSON.** Массив
+  `purchaseRevenue`, объявленный источником как `Array(Float64)`, разбирается
+  сразу в `Array(Decimal(18, 2))`. Синтаксис подтвержден через Context7 и на
+  ClickHouse 26.3; все суммы после ODS совпали с сырьем до копейки.
 
 **Сказано по памяти, проверки нет.** Группа пуста: оба утверждения, ждавшие
 матвью разбора, закрыты опытами при исполнении #43 и переехали выше.

docs/formats/clickstream-event.md

@@ -11,10 +11,12 @@
 по элементу на товар. Отдельной сущности «визит» в выгрузке нет — визиты
 собирают на стороне хранилища, а `VisitID` дан как эталон для самопроверки.
 
-Имена и типы колонок — стороны источника. Хранилище принимает их как есть и
-нормализует у себя: то же имя в нашем стиле ждёт в столбце «Нормализованное
-имя». Это имя источника, приведённое к snake_case, а не имя атрибута в
-модели данных: слой DDS складывает свою модель и называет атрибуты по ней.
+Имена и типы колонок описывают сторону источника. Обычно ODS принимает их без
+изменений. Исключение — `purchaseRevenue`: источник объявляет Float64, а ODS
+хранит деньги в Decimal. Точный локальный тип назван в документации
+хранилища. Имя в нашем стиле ждет в столбце «Нормализованное имя». Это имя
+источника, приведенное к snake_case, а не имя атрибута в модели данных: слой
+DDS складывает свою модель и называет атрибуты по ней.
 Столбец «Тип numpy» показывает, чем колонка представлена внутри генератора;
 у массивов это тип элемента. Номер — место колонки в выгрузке: порядок задан
 контрактом.
@@ -90,7 +92,7 @@
 | № | Колонка | Тип ClickHouse | Тип numpy | Нормализованное имя | Комментарий |
 |---|---|---|---|---|---|
 | 37 | `purchaseID` | `Array(String)` | `object` | `purchase_id` | номер заказа; у события purchase — один элемент |
-| 38 | `purchaseRevenue` | `Array(Float64)` | `float64` | `purchase_revenue` | выручка заказа глазами клиента; Float64, как у Метрики — на этом держится урок о расхождениях с бэкендом |
+| 38 | `purchaseRevenue` | `Array(Float64)` | `float64` | `purchase_revenue` | выручка заказа глазами клиента; источник объявляет Float64, ODS хранит точный Decimal |
 | 39 | `purchaseCurrency` | `Array(String)` | `object` | `purchase_currency` | валюта заказа |
 | 40 | `purchaseCoupon` | `Array(String)` | `object` | `purchase_coupon` | купон заказа, если был применён |
 | 41 | `productID` | `Array(String)` | `object` | `product_id` | id товаров события |

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

@@ -138,8 +138,9 @@ Ecommerce (заполнены только у торговых событий):
 | `ecommerce` | String — сырой JSON события, как отдаёт Метрика (кандидат будущей лабы: сырое против разобранного) |
 
 `add_to_cart` несёт массивы `product*` с одним товаром; `purchase` — состав
-заказа и блок `purchase*`. Выручка у клиента — во Float64, как у Метрики:
-это не недосмотр, а часть урока о расхождениях (см. раздел 4).
+заказа и блок `purchase*`. Выручка приходит JSON-числом Float64, как у
+Метрики. ODS один раз нормализует деньги в `Array(Decimal(18, 2))`, не меняя
+формат источника; DDS наследует тип хранения без нового вычисления.
 
 ### 1.3 Ключи и движки
 
@@ -210,9 +211,9 @@ Ecommerce (заполнены только у торговых событий):
   ключей совпадает с порядком полей в таблице контракта, у позиции — `sku`,
   `qty`, `price`: так байты воспроизводимы без сортировки ключей. Разница с
   кликстримом намеренна: там значения `purchaseRevenue` приезжают JSON-числами
-  и разбираются как `Array(Float64)`, а бэкенд передаёт деньги строками для
-  точного `Decimal`. Это показывает расхождение представлений денег в двух
-  источниках.
+  Float64, а бэкенд передает деньги строками. На границе ODS оба источника
+  становятся `Decimal(18, 2)`: клиентское число разбирается сразу, без
+  промежуточного Float64 в хранилище.
   Запись собирает явная функция существующего канонического сериализатора:
   один словарь с вложенным списком и один `orjson.dumps`, без универсального
   слоя кодеков.
@@ -285,7 +286,7 @@ CSV в репозитории (`data/catalog/products.csv`: `sku`, `name`, `cate
 под именем `declared_*` там, где атрибуция без трекера невозможна (UTM);
 слово `declared` в имени — сигнал «это заявка клиента, не деньги
 отчётности». Оно выучивается на конфликте: суммы не сойдутся, менти сам
-раскопает почему (Float64 против Decimal, промокод, доставка, отмены).
+раскопает почему (состав позиций, промокод, доставка, отмены).
 
 ## 5. Анонимность и склейка идентичностей
 

generator/src/clickstream_generator/schema.py

@@ -12,12 +12,12 @@
 
 - `name` — имя источника, как в облачной выгрузке Метрики; сырой слой хранит
   его без изменений.
-- `clickhouse_type` — тип в хранилище ровно в той записи, в какой его вернёт
-  `system.columns`: параметры входят в имя типа целиком, без сокращений
-  (`LowCardinality(String)`, `Array(Float64)`). Сверено 2026-08-01 —
-  по документации ClickHouse через Context7 и запросом к узлу стенда
-  (26.3.17.56). Запись важна потому, что по ней человек пишет DDL: тип,
-  сокращённый здесь, приедет в таблицу сокращённым же.
+- `clickhouse_type` — тип колонки у источника в той записи, которой пользуется
+  ClickHouse: параметры входят в имя целиком, без сокращений
+  (`LowCardinality(String)`, `Array(Float64)`). Обычно ODS берет его без
+  изменений. Исключение одно: денежный `purchaseRevenue` приезжает как
+  Float64, а на границе хранения разбирается прямо в Decimal. Поэтому поле
+  описывает выгрузку, а локальный тип ODS назван в документации хранилища.
 - `numpy_dtype` — чем колонка представлена внутри генератора; у массивов это
   тип элемента. Строки живут в `object`-массивах: numpy-строки фиксированной
   длины стенду ничего не дают.
@@ -371,8 +371,8 @@ COLUMNS: tuple[Column, ...] = (
         numpy_dtype="float64",
         normalized_name="purchase_revenue",
         group=ColumnGroup.ECOMMERCE,
-        comment="выручка заказа глазами клиента; Float64, как у Метрики —"
-        " на этом держится урок о расхождениях с бэкендом",
+        comment="выручка заказа глазами клиента; источник объявляет Float64,"
+        " ODS хранит точный Decimal",
     ),
     Column(
         name="purchaseCurrency",

generator/src/clickstream_generator/schema_doc.py

@@ -27,10 +27,12 @@ PREAMBLE = """# Описание выгрузки: событие кликстр
 по элементу на товар. Отдельной сущности «визит» в выгрузке нет — визиты
 собирают на стороне хранилища, а `VisitID` дан как эталон для самопроверки.
 
-Имена и типы колонок — стороны источника. Хранилище принимает их как есть и
-нормализует у себя: то же имя в нашем стиле ждёт в столбце «Нормализованное
-имя». Это имя источника, приведённое к snake_case, а не имя атрибута в
-модели данных: слой DDS складывает свою модель и называет атрибуты по ней.
+Имена и типы колонок описывают сторону источника. Обычно ODS принимает их без
+изменений. Исключение — `purchaseRevenue`: источник объявляет Float64, а ODS
+хранит деньги в Decimal. Точный локальный тип назван в документации
+хранилища. Имя в нашем стиле ждет в столбце «Нормализованное имя». Это имя
+источника, приведенное к snake_case, а не имя атрибута в модели данных: слой
+DDS складывает свою модель и называет атрибуты по ней.
 Столбец «Тип numpy» показывает, чем колонка представлена внутри генератора;
 у массивов это тип элемента. Номер — место колонки в выгрузке: порядок задан
 контрактом.

sql/ddl/20-ods-tables.sql

@@ -83,7 +83,7 @@ CREATE TABLE IF NOT EXISTS ods.event_rep ON CLUSTER clickstream_cluster
     GoalsReached Array(UInt32),
     ParsedParamsKey1 Array(String),
     purchaseID Array(String),
-    purchaseRevenue Array(Float64),
+    purchaseRevenue Array(Decimal(18, 2)),
     purchaseCurrency Array(String),
     purchaseCoupon Array(String),
     productID Array(String),

sql/ddl/30-ods-views.sql

@@ -154,7 +154,11 @@ SELECT
     JSONExtract(raw, 'GoalsReached', 'Array(UInt32)') AS GoalsReached,
     JSONExtract(raw, 'ParsedParamsKey1', 'Array(String)') AS ParsedParamsKey1,
     JSONExtract(raw, 'purchaseID', 'Array(String)') AS purchaseID,
-    JSONExtract(raw, 'purchaseRevenue', 'Array(Float64)') AS purchaseRevenue,
+    -- Выгрузка несет JSON-числа типа Float64, но деньги хранятся в Decimal.
+    -- JSONExtract разбирает их сразу, без промежуточного двоичного числа:
+    -- синтаксис подтвержден документацией ClickHouse через Context7 и
+    -- запросом к стендовому ClickHouse 26.3 при исполнении #155.
+    JSONExtract(raw, 'purchaseRevenue', 'Array(Decimal(18, 2))') AS purchaseRevenue,
     JSONExtract(raw, 'purchaseCurrency', 'Array(String)') AS purchaseCurrency,
     JSONExtract(raw, 'purchaseCoupon', 'Array(String)') AS purchaseCoupon,
     JSONExtract(raw, 'productID', 'Array(String)') AS productID,

sql/ddl/55-dm-tables.sql

@@ -1,4 +1,4 @@
--- DM: физические дневные витрины и донор партиций выручки.
+-- DM: физические дневные витрины и доноры их партиций.
 
 -- Выручка хранится в зерне «день заказа × категория товара». День служит
 -- единицей пересборки, поэтому совпадает с ключом партиции.
@@ -48,6 +48,51 @@ ENGINE = Distributed(
     cityHash64(product_category)
 );
 
+-- Сверка хранит одну строку на заказ. День — единица пересборки: пока рядом
+-- живут два источника, строка может перейти из awaiting_order в обычный класс.
+CREATE TABLE IF NOT EXISTS dm.purchase_vs_orders_rep ON CLUSTER clickstream_cluster
+(
+    order_day Date,
+    order_id String,
+    declared_revenue Nullable(Decimal(18, 2)),
+    items_total Nullable(Decimal(18, 2)),
+    status LowCardinality(Nullable(String)),
+    mismatch_class LowCardinality(String),
+    _load_id String,
+    _load_ts DateTime64(3, 'UTC')
+)
+ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/{database}/{table}', '{replica}')
+PARTITION BY order_day
+ORDER BY (order_day, order_id);
+
+-- Ключ совпадает с DDS, поэтому строка заказа на всех слоях попадает на один
+-- шард.
+CREATE TABLE IF NOT EXISTS dm.purchase_vs_orders_dist ON CLUSTER clickstream_cluster
+AS dm.purchase_vs_orders_rep
+ENGINE = Distributed(
+    'clickstream_cluster',
+    'dm',
+    'purchase_vs_orders_rep',
+    cityHash64(order_id)
+);
+
+-- Донор повторяет структуру и раскладку цели по правилу, разобранному у
+-- revenue_daily_stage_rep и в docs/architecture/storage.md.
+CREATE TABLE IF NOT EXISTS dm.purchase_vs_orders_stage_rep ON CLUSTER clickstream_cluster
+AS dm.purchase_vs_orders_rep
+ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/{database}/{table}', '{replica}')
+PARTITION BY order_day
+ORDER BY (order_day, order_id);
+
+CREATE TABLE IF NOT EXISTS dm.purchase_vs_orders_stage_dist ON CLUSTER clickstream_cluster
+AS dm.purchase_vs_orders_stage_rep
+ENGINE = Distributed(
+    'clickstream_cluster',
+    'dm',
+    'purchase_vs_orders_stage_rep',
+    cityHash64(order_id)
+);
+
 -- Трафик пересчитывается целиком: новая связь куки с пользователем меняет
 -- прошлые дни. Версия запуска позволяет точному представлению скрыть строки
 -- прежних пересчетов без TRUNCATE и без партиций.

sql/ddl/60-dm-views.sql

@@ -15,6 +15,22 @@ SELECT
     _load_ts
 FROM dm.revenue_daily_dist;
 
+-- Непарные стороны остаются NULL. При join_use_nulls = 0 ClickHouse заполнил
+-- бы их нулями и пустыми строками, скрыв отсутствие источника.
+-- Служебные колонки называют запуск, который последним заменил день.
+CREATE VIEW IF NOT EXISTS dm.purchase_vs_orders_v ON CLUSTER clickstream_cluster
+AS
+SELECT
+    order_day,
+    order_id,
+    declared_revenue,
+    items_total,
+    status,
+    mismatch_class,
+    _load_id,
+    _load_ts
+FROM dm.purchase_vs_orders_dist;
+
 -- Полный пересчет пишет новую версию каждого дня. FINAL оставляет текущие
 -- счетчики и прячет повторы запусков от потребителя.
 CREATE VIEW IF NOT EXISTS dm.daily_traffic_v ON CLUSTER clickstream_cluster

sql/dm/purchase_vs_orders_rebuild.sql

@@ -0,0 +1,71 @@
+-- DM: сверка покупок трекера с текущими заказами бэкенда.
+--
+-- Обе стороны FULL OUTER JOIN — подзапросы. С голой левой Distributed
+-- ClickHouse вернул бы непарный заказ по разу с каждого шарда; форма измерена
+-- в docs/research/2026-08-22-distributed-join.md.
+-- Выбор DROP PARTITION ALL вместо TRUNCATE разобран в соседнем
+-- sql/dds/order_rebuild.sql.
+
+ALTER TABLE dm.purchase_vs_orders_stage_rep ON CLUSTER clickstream_cluster
+DROP PARTITION ALL;
+
+INSERT INTO dm.purchase_vs_orders_stage_dist
+(
+    order_day,
+    order_id,
+    declared_revenue,
+    items_total,
+    status,
+    mismatch_class,
+    _load_id,
+    _load_ts
+)
+SELECT
+    -- У парного заказа день берется у бэкенда; трекер дает его только
+    -- непарной покупке.
+    coalesce(order_date, purchase_day),
+    coalesce(purchase_order_id, backend_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'
+    ),
+    {load_id:String},
+    now64(3, 'UTC')
+FROM
+(
+    -- Контракт purchase несет один номер и одну объявленную сумму в массивах.
+    -- Бизнес-дубль повторяет оба значения, поэтому count хранится отдельно,
+    -- а сумма берется одна — складывать дубли значило бы выдумать дельту.
+    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'
+        AND event_date BETWEEN {first_day:Date} AND {last_day:Date}
+    GROUP BY purchase_order_id
+) AS purchases
+GLOBAL FULL OUTER JOIN
+(
+    SELECT
+        order_id AS backend_order_id,
+        order_date,
+        items_total,
+        status
+    FROM dds.order_v
+    WHERE order_date BETWEEN {first_day:Date} AND {last_day:Date}
+) AS orders ON purchase_order_id = backend_order_id
+-- NULL обозначает отсутствующую сторону. При join_use_nulls = 0 ClickHouse
+-- заполнил бы ее нулем или пустой строкой и скрыл отсутствие источника.
+-- Вставка идет через ключ цели: источники разложены по другим ключам.
+SETTINGS join_use_nulls = 1,
+    distributed_foreground_insert = 1,
+    parallel_distributed_insert_select = 0;

sql/dm/purchase_vs_orders_replace.sql

@@ -0,0 +1,9 @@
+-- DM: подмена одного дня сверки собранной партицией донора.
+--
+-- День для PARTITION берет purchase_vs_orders_scope.sql. Выбор Jinja,
+-- атомарность и повтор разобраны в sql/dds/order_replace.sql.
+--
+-- Пустая партиция законна: если источник больше не показывает строк дня,
+-- замена убирает этот день из витрины.
+ALTER TABLE dm.purchase_vs_orders_rep ON CLUSTER clickstream_cluster
+REPLACE PARTITION '{{ params.day }}' FROM dm.purchase_vs_orders_stage_rep;

sql/dm/purchase_vs_orders_scope.sql

@@ -0,0 +1,62 @@
+-- DM: объем пересборки сверки покупок с заказами.
+--
+-- У источников разные правые границы. Заказы называют изменяемое окно ярлыком
+-- последнего принятого слепка. События заканчиваются последним прожитым днем,
+-- который выводится из позиции мира. Одна граница другую не подменяет.
+--
+-- Левая граница возвращает не только пропуски, но и самый ранний
+-- awaiting_order. Такой день уже есть в витрине, поэтому обычный поиск дыры
+-- его не увидит. Иначе пакетный разгон длиннее K навсегда оставил бы прежний
+-- хвост предварительным.
+--
+-- Файл заканчивается без точки с запятой: почему — sql/dds/order_scope.sql.
+WITH
+    (SELECT min(event_date) FROM dds.event_v) AS first_event_day,
+    first_event_day + ({position:UInt16} - 1) AS last_lived_day,
+    (SELECT max(snapshot_date) FROM ods.order_dist) AS last_snapshot,
+    (
+        SELECT min(toDate(toTimeZone(created_at, 'Europe/Samara')))
+        FROM ods.order_dist
+        WHERE snapshot_date = last_snapshot
+    ) AS window_start,
+    (
+        SELECT max(toDate(toTimeZone(created_at, 'Europe/Samara')))
+        FROM ods.order_dist
+        WHERE snapshot_date = last_snapshot
+    ) AS window_end,
+    (
+        SELECT maxOrNull(event_date)
+        FROM dds.event_v
+        WHERE event_type = 'purchase' AND event_date <= last_lived_day
+    ) AS last_purchase_day,
+    (
+        SELECT minOrNull(day)
+        FROM (
+            SELECT order_date AS day
+            FROM dds.order_v
+            UNION DISTINCT
+            SELECT event_date AS day
+            FROM dds.event_v
+            WHERE event_type = 'purchase' AND event_date <= last_lived_day
+        )
+        WHERE day GLOBAL NOT IN (
+            SELECT order_day FROM dm.purchase_vs_orders_dist
+        )
+    ) AS first_missing_day,
+    (
+        SELECT minOrNull(order_day)
+        FROM dm.purchase_vs_orders_dist
+        WHERE mismatch_class = 'awaiting_order'
+    ) AS first_provisional_day
+SELECT
+    least(
+        window_start,
+        coalesce(first_missing_day, window_start),
+        coalesce(first_provisional_day, window_start)
+    ) AS first_day,
+    greatest(window_end, last_purchase_day) AS last_day
+-- Без принятого слепка неизвестно окно заказов. Без готового DDS или прожитой
+-- покупки донор был бы пуст, поэтому подменять партиции еще нельзя.
+WHERE (SELECT count() FROM ods.order_dist) > 0
+    AND (SELECT count() FROM dds.order_v) > 0
+    AND last_purchase_day IS NOT NULL