dags/dm_transform.py

@@ -0,0 +1,126 @@
+"""Трансформации слоя DM: сборка стендовых витрин.
+
+Выручка наследует ритм заказов и заменяет дневные партиции изменяемого окна.
+Трафик наследует ретроспективность карты идентичностей и пересчитывает всю
+историю новой версией. Объем каждого прогона выводится из состояния слоев.
+"""
+
+from __future__ import annotations
+
+import datetime
+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 dag, task, task_group
+
+START_DATE = datetime.datetime(2026, 1, 1, tzinfo=datetime.UTC)
+
+SQL_ROOT = Path("/opt/airflow/sql")
+CLICKHOUSE_CONNECTION = "clickhouse_default"
+
+
+def _days(first_day: datetime.date, last_day: datetime.date) -> list[str]:
+    """Дни отрезка строками - в таком виде их ждет замена партиции."""
+    return [
+        (first_day + datetime.timedelta(days=shift)).isoformat()
+        for shift in range((last_day - first_day).days + 1)
+    ]
+
+
+@task
+def replacements(scope: dict[str, object]) -> list[dict[str, object]]:
+    """Разложить дни в аргументы размноженной задачи замены.
+
+    Отдельный словарь нужен каждому экземпляру ``expand_kwargs``; тот же
+    прием разобран в ``dds_transform.py`` на замене партиций заказов.
+    """
+    return [{"params": {"day": day}} for day in scope["days"]]
+
+
+@dag(
+    dag_id="dm_transform",
+    schedule=None,
+    start_date=START_DATE,
+    is_paused_upon_creation=False,
+    max_active_runs=1,
+    template_searchpath=str(SQL_ROOT),
+    tags=["dm"],
+)
+def dm_transform():
+    """Пересобрать стендовые витрины по текущему состоянию DDS."""
+
+    @task_group(group_id="revenue_daily")
+    def revenue_daily():
+        """Выручка: изменяемое окно заказов и пропущенная история."""
+
+        @task
+        def scope() -> dict[str, object]:
+            """Спросить у хранилища, какие дни пересобирает запуск."""
+            query = (SQL_ROOT / "dm" / "revenue_daily_scope.sql").read_text(
+                encoding="utf-8"
+            )
+            hook = ClickHouseHook(clickhouse_conn_id=CLICKHOUSE_CONNECTION)
+            row = hook.get_first(query)
+            if row is None:
+                raise AirflowException(
+                    "нет слепка заказов в ODS или готовых заказов в DDS; "
+                    "дождитесь загрузки слепка и сначала выполните 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/revenue_daily_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/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
+
+    revenue_daily()
+
+    SQLExecuteQueryOperator(
+        task_id="daily_traffic",
+        conn_id=CLICKHOUSE_CONNECTION,
+        sql="dm/daily_traffic_rebuild.sql",
+        parameters={"load_id": "{{ run_id }}"},
+        do_xcom_push=False,
+    )
+
+
+dm_transform()

docs/architecture/dds/event.md

@@ -89,10 +89,8 @@
 - Сборка сессий: ключи нарезки (`client_id`, `event_time`,
   `event_date`), тип события, страница, UTM, паспорт куки — категория
   устройства приходит уже расшифрованной.
-- Витрины `dm.events_enriched_v`, `dm.top_pages_daily_v`,
-  `dm.utm_effectiveness_v` (`declared_revenue` — из `purchase_revenue`),
-  `dm.daily_traffic_v` (пара «посетители / известные» через
-  `dds.identity_map_v`).
+- Витрины `dm.events_enriched_v`, `dm.top_pages_daily_v` и
+  `dm.utm_effectiveness_v` (`declared_revenue` берется из `purchase_revenue`).
 - Событийная сторона сверки `dm.purchase_vs_orders_v`: `purchase_id`,
   `purchase_revenue`, `event_date`.
 - DQ-сверка сессий с эталоном читает не его, а `ods.event_v` —

docs/architecture/dm.md

@@ -31,8 +31,8 @@
   витрины, на него ссылаются учебные задания;
 - `purchase_vs_orders` — сверка A+C с классами расхождений; зерно — заказ,
   шардирование сохраняет ко-локацию заказов по `cityHash64(order_id)`;
-- `daily_traffic` — пара «посетители» / «известные пользователи», локальное
-  обогащение [картой идентичностей](dds/identity.md);
+- `daily_traffic` — пара «посетители» / «известные пользователи», обогащение
+  [картой идентичностей](dds/identity.md);
 - `dq_summary` — сводка проверок качества (ниже).
 
 Остальные витрины спеки — учебные задания: менти строит их сам по ТЗ с
@@ -55,8 +55,8 @@
 
 - **от событий и сессий** — день замерзает сразу; пересборка догоняет
   хвост несобранных дней заменой дневных партиций;
-- **от заказов** — день дышит окном K и закрывается; пересборка заменяет
-  партиции окна, дни за ним закрыты по построению;
+- **от заказов** — день дышит, пока входит в последний слепок; пересборка
+  заменяет партиции этого отрезка, дни за ним закрыты по построению;
 - **от карты идентичностей** — не замерзает никогда: склейка
   ретроспективна, покупка сегодня делает куку известной во всех её прошлых
   днях, а карта пересчитывается целиком.
@@ -64,12 +64,58 @@
 Поэтому `daily_traffic` наследует у своего источника и ритм, и механику:
 полный пересчёт всех дней каждый прогон в `ReplacingMergeTree(_load_ts)`,
 как у самой карты. Дневных партиций у неё нет: партиция — единица замены,
-а эта витрина заменяется только целиком. Пересчёт дешёв:
-витрина читает сессии и карту локальным соединением по ключу ко-локации.
+а эта витрина заменяется только целиком. На стенде такая работа остается
+быстрой; витрина читает сессии и карту через их публичные договоры.
 Пересчитывать точечно только затронутые дни не из чего: для этого надо
 знать, какие связки новые, а полный пересчёт карты этот след затирает —
 восстановление следа оставлено лабе повышенной сложности (issue #149).
 
+## Витрины стенда
+
+Оба публичных представления перечисляют деловые колонки и координаты запуска
+`_load_id`, `_load_ts` явно. Новая колонка физической таблицы сама в договор не
+попадет.
+
+### `dm.revenue_daily_v`
+
+Зерно: день заказа × категория товара. Витрина читает только оплаченные
+заказы из `dds.order_v`, синхронно разворачивает три массива позиций и узнает
+категорию артикула через `dic.products`. Один заказ с несколькими позициями
+категории считается в `orders` один раз. `units` складывает количество единиц,
+`revenue` складывает цену позиции × количество. Скидка, доставка и итог заказа
+между категориями не делятся. `aov` равен `revenue / orders` в том же срезе и
+хранит два знака после запятой.
+
+Первый запуск строит всю доступную историю. Следующие берут границы дней из
+последнего слепка ODS: только источник знает, какие даты еще может изменить.
+Пропущенный оплаченный день расширяет левую границу, чтобы частичный отказ
+не оставил дыру после следующего прогона. Агрегат при этом читает DDS; ODS
+отвечает только на вопрос об объеме.
+
+`dm.revenue_daily_rep` нарезана по `report_date`; день собирается в
+таблице-доноре и заменяется через `REPLACE PARTITION`. Цель и донор шардируются
+по `cityHash64(product_category)`. Кроме дня, у витрины только категория, а
+шесть категорий стартового мира близки по объему. Более сложный составной ключ
+распределил бы эту маленькую таблицу без нового урока.
+
+### `dm.daily_traffic_v`
+
+В строке один день. `visitors` считает разные `client_id` из
+`dds.session_v`; `known_users` считает разные `user_id` после левого соединения
+с полной текущей `dds.identity_map_v`. Анонимная сессия остается в витрине.
+Несколько кук одного человека увеличивают `visitors`, но дают одного
+`known_user`, поэтому `known_users <= visitors`.
+
+Сессии и карта лежат по `client_id`, но публичные `_v` скрывают
+`Distributed`-таблицы от локальной подстановки. На ClickHouse 26.3
+`EXPLAIN PLAN` показал чтение обоих шардов на инициаторе: настройка
+`distributed_product_mode = 'local'` локального соединения не дала. Поэтому
+витрина использует явный `GLOBAL LEFT JOIN` через публичные договоры, а не
+подменяет их физическими таблицами. Это дороже по сети, зато граница слоев
+остается явной. Новая связь куки меняет прошлые дни, поэтому каждый запуск
+пишет полную историю новой версией в `ReplacingMergeTree(_load_ts)`, а
+публичное представление выбирает текущие строки через `FINAL`.
+
 ## `dm.dq_summary`
 
 Зерно — «день данных × проверка»: строка отвечает, каково качество дня
@@ -113,7 +159,7 @@ Airflow. Партиция — день данных; прогон звена DQ
 
 ## Остаётся нарезке
 
-- Имена задач `dm_transform`, звена DQ и дага самопроверки.
+- Имена звена DQ и дага самопроверки.
 - Ключи шардирования остальных материализованных витрин — вместе с их
   DDL; правило партиции-донора — в [доке хранилища](storage.md).
 - Тексты учебных заданий и точный список проверок обоих контуров.
@@ -136,7 +182,7 @@ Airflow. Партиция — день данных; прогон звена DQ
   расшифровывать. Красное «ещё не построено» честнее.
 - **Самопроверка пишет в `dq_summary`.** Попытки менти смешались бы с
   историей качества конвейера.
-- **Окно K для `daily_traffic` — прошлое замерзает.** Дни за окном хранили
+- **Окно заказов для `daily_traffic` — прошлое замерзает.** Дни за окном хранили
   бы число известных на момент закрытия — витрина тихо расходится с картой
   и протаскивает обратно «знание на дату», отвергнутое развилкой карты
   (#133).

docs/architecture/storage.md

@@ -521,8 +521,10 @@ SQL-файлы исполняет штатный `SQLExecuteQueryOperator`. Пр
 | `20-ods-tables.sql` | типизированное событие, версии заказа и обе таблицы ошибок |
 | `30-ods-views.sql` | актуальные события, текущие заказы и матвью разбора в ODS |
 | `40-stg-views.sql` | матвью приёма: чтец в сырьё |
-| `45-dds-tables.sql` | физические модели заказа и сессии, доноры их партиций |
-| `50-dds-views.sql` | публичные договоры события, заказа и сессии на языке модели DDS |
+| `45-dds-tables.sql` | физические модели заказа, сессии и карты идентичностей, доноры партиций |
+| `50-dds-views.sql` | публичные договоры сущностей DDS |
+| `55-dm-tables.sql` | физические витрины выручки и трафика, донор партиций выручки |
+| `60-dm-views.sql` | публичные договоры витрин DM |
 
 Справочники идут сразу за базами: зона `dic` не зависит ни от одного слоя, и
 правила порядка слоёв к ней не применяются.
@@ -589,6 +591,11 @@ SQL-файлы исполняет штатный `SQLExecuteQueryOperator`. Пр
 | `dds` | `dds.session_v` | публичный договор сессии |
 | `dds` | `dds.identity_map_rep` / `_dist` | известная пара куки с пользователем; таблицы полного пересчёта |
 | `dds` | `dds.identity_map_v` | точный публичный договор карты идентичностей с `FINAL` |
+| `dm` | `dm.revenue_daily_rep` / `_dist` | дневная выручка по категориям |
+| `dm` | `dm.revenue_daily_stage_rep` / `_dist` | таблица-двойник, донор дневных партиций выручки |
+| `dm` | `dm.revenue_daily_v` | публичный договор дневной выручки |
+| `dm` | `dm.daily_traffic_rep` / `_dist` | версии дневных счетчиков трафика |
+| `dm` | `dm.daily_traffic_v` | точные текущие счетчики трафика с `FINAL` |
 | `dic` | `dic.products_file` | чтец CSV-каталога, общего с генератором |
 | `dic` | `dic.products` | словарь товаров поверх подложки |
 
@@ -598,9 +605,8 @@ SQL-файлы исполняет штатный `SQLExecuteQueryOperator`. Пр
 
 В слое DDS построены событие, заказ, сессия и карта идентичностей. Событие
 остаётся представлением над ODS; заказ и сессия гасят повтор заменой партиции;
-карта пишет полный набор пар в версионную таблицу. Слой DM принесут следующие
-тикеты этапа. Его состав задан разделом 7 мастер-спеки и переносится сюда по
-мере постройки.
+карта пишет полный набор пар в версионную таблицу. В DM выручка заменяет
+партиции изменяемых дней, а трафик пишет полную историю новой версией.
 
 ## Что проверено
 

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

@@ -367,8 +367,10 @@ CSV в репозитории (`data/catalog/products.csv`: `sku`, `name`, `cate
   ошибка в `*_errors`, а не молчаливые нули: имена CamelCase регистрозависимы,
   опечатка иначе не падает.
 - **Политика соединений**: по ключу ко-локации — обычное соединение с
-  комментарием, почему локальный результат корректен; по любому другому ключу
-  — только явный GLOBAL; `NOT IN` — только `GLOBAL NOT IN`. Сверка
+  комментарием, почему локальный результат корректен; если публичные
+  представления скрывают ко-локацию — решение записывается рядом с витриной.
+  По любому другому ключу — только явный GLOBAL; `NOT IN` — только
+  `GLOBAL NOT IN`. Сверка
   `purchase`↔заказ —
   легитимная GLOBAL-витрина (заказы малы).
 - **Конвейер без TRUNCATE**: поток версий — в ReplacingMergeTree, точное чтение
@@ -505,8 +507,9 @@ Kafka день переигрывается генератором заново:
   счётчики `purchases`/`add_to_carts` оживают из таксономии, добавляется
   `declared_revenue` по UTM.
 - **`daily_traffic_v`** — расширяется парой «посетители» / «известные
-  пользователи» (обогащение через `dds.identity_map_v`, локальное соединение
-  по ключу ко-локации).
+  пользователи». Обогащение идет явным `GLOBAL LEFT JOIN` через
+  `dds.identity_map_v`: опыт на ClickHouse 26.3 показал, что публичные `_v`
+  скрывают ко-локацию от локальной подстановки.
 - `events_enriched_v`, `top_pages_daily_v`, `session_overview_v`,
   `dq_errors_daily_v` — переезжают на новую модель без смены роли: источник —
   `dds.event_v`, не `ods.event`.
@@ -684,10 +687,6 @@ Kafka Engine на двух нодах снят с этого списка при
 `ReplacingMergeTree`. Проверять то, что заведомо случается и заведомо
 обезврежено, — работа без ответа на конце.
 
-- Точную форму локального обогащения `daily_traffic_v` через публичные `_v` —
-  эмпирически при проектировании витрин #135. Общее поведение соединения двух
-  Distributed-таблиц уже проверено в
-  [исследовании](../research/2026-08-22-distributed-join.md).
 - Размер артефакта эталонного мира после пересборки.
 - Спорные API (Airflow Datasets/сенсоры, ClickHouse DDL) — перед кодом
   сверять через MCP Context7 (правило AGENTS.md).

sql/ddl/00-databases.sql

@@ -17,6 +17,9 @@ CREATE DATABASE IF NOT EXISTS ods ON CLUSTER clickstream_cluster;
 -- База DDS заводится здесь же по той же причине, что и ODS.
 CREATE DATABASE IF NOT EXISTS dds ON CLUSTER clickstream_cluster;
 
+-- База витрин завершает цепочку STG → ODS → DDS → DM.
+CREATE DATABASE IF NOT EXISTS dm ON CLUSTER clickstream_cluster;
+
 -- Справочники стоят вне цепочки STG → ODS → DDS → DM: в хранилище их никто не
 -- производит, а читают их несколько слоёв — ADR 0012. Порядок слоёв к зоне не
 -- применяется, её файл идёт сразу за этим.

sql/ddl/55-dm-tables.sql

@@ -0,0 +1,78 @@
+-- DM: физические дневные витрины и донор партиций выручки.
+
+-- Выручка хранится в зерне «день заказа × категория товара». День служит
+-- единицей пересборки, поэтому совпадает с ключом партиции.
+CREATE TABLE IF NOT EXISTS dm.revenue_daily_rep ON CLUSTER clickstream_cluster
+(
+    report_date Date,
+    product_category LowCardinality(String),
+    orders UInt64,
+    units UInt64,
+    revenue Decimal(18, 2),
+    aov Decimal(18, 2),
+    _load_id String,
+    _load_ts DateTime64(3, 'UTC')
+)
+ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/{database}/{table}', '{replica}')
+PARTITION BY report_date
+ORDER BY (report_date, product_category);
+
+-- Категория — единственное измерение кроме дня. Шесть категорий стартового
+-- мира близки по объему, поэтому простой ключ раскладывает строки достаточно
+-- ровно и не добавляет в учебный пример составной хеш без читателя.
+CREATE TABLE IF NOT EXISTS dm.revenue_daily_dist ON CLUSTER clickstream_cluster
+AS dm.revenue_daily_rep
+ENGINE = Distributed(
+    'clickstream_cluster',
+    'dm',
+    'revenue_daily_rep',
+    cityHash64(product_category)
+);
+
+-- Донор объявлен через AS и шардируется тем же ключом, что цель. REPLACE
+-- PARTITION идет по локальным таблицам, поэтому структура и раскладка донора
+-- обязаны совпадать с целью (docs/architecture/storage.md, «Раскладка по
+-- шардам»).
+CREATE TABLE IF NOT EXISTS dm.revenue_daily_stage_rep ON CLUSTER clickstream_cluster
+AS dm.revenue_daily_rep
+ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/{database}/{table}', '{replica}')
+PARTITION BY report_date
+ORDER BY (report_date, product_category);
+
+CREATE TABLE IF NOT EXISTS dm.revenue_daily_stage_dist ON CLUSTER clickstream_cluster
+AS dm.revenue_daily_stage_rep
+ENGINE = Distributed(
+    'clickstream_cluster',
+    'dm',
+    'revenue_daily_stage_rep',
+    cityHash64(product_category)
+);
+
+-- Трафик пересчитывается целиком: новая связь куки с пользователем меняет
+-- прошлые дни. Версия запуска позволяет точному представлению скрыть строки
+-- прежних пересчетов без TRUNCATE и без партиций.
+CREATE TABLE IF NOT EXISTS dm.daily_traffic_rep ON CLUSTER clickstream_cluster
+(
+    report_date Date,
+    visitors UInt64,
+    known_users UInt64,
+    _load_id String,
+    _load_ts DateTime64(3, 'UTC')
+)
+ENGINE = ReplicatedReplacingMergeTree(
+    '/clickhouse/tables/{shard}/{database}/{table}',
+    '{replica}',
+    _load_ts
+)
+ORDER BY report_date;
+
+-- Все версии одного дня должны попадать на один шард, иначе FINAL не сможет
+-- выбрать последнюю через границу шардов.
+CREATE TABLE IF NOT EXISTS dm.daily_traffic_dist ON CLUSTER clickstream_cluster
+AS dm.daily_traffic_rep
+ENGINE = Distributed(
+    'clickstream_cluster',
+    'dm',
+    'daily_traffic_rep',
+    cityHash64(report_date)
+);

sql/ddl/60-dm-views.sql

@@ -0,0 +1,28 @@
+-- DM: публичные договоры стендовых витрин.
+
+-- День и категория — зерно выручки. Служебные колонки показывают запуск,
+-- который последним заменил дневную партицию.
+CREATE VIEW IF NOT EXISTS dm.revenue_daily_v ON CLUSTER clickstream_cluster
+AS
+SELECT
+    report_date,
+    product_category,
+    orders,
+    units,
+    revenue,
+    aov,
+    _load_id,
+    _load_ts
+FROM dm.revenue_daily_dist;
+
+-- Полный пересчет пишет новую версию каждого дня. FINAL оставляет текущие
+-- счетчики и прячет повторы запусков от потребителя.
+CREATE VIEW IF NOT EXISTS dm.daily_traffic_v ON CLUSTER clickstream_cluster
+AS
+SELECT
+    report_date,
+    visitors,
+    known_users,
+    _load_id,
+    _load_ts
+FROM dm.daily_traffic_dist FINAL;

sql/dm/daily_traffic_rebuild.sql

@@ -0,0 +1,29 @@
+-- DM: полный пересчет дневного трафика.
+--
+-- Публичные _v скрывают Distributed-таблицы от локальной подстановки:
+-- EXPLAIN на ClickHouse 26.3 показал чтение обоих шардов на инициаторе.
+-- Поэтому GLOBAL назван явно, а физические таблицы не подставлены. LEFT JOIN
+-- сохраняет анонимные сессии.
+INSERT INTO dm.daily_traffic_dist
+(
+    report_date,
+    visitors,
+    known_users,
+    _load_id,
+    _load_ts
+)
+SELECT
+    sessions.session_date,
+    uniqExact(sessions.client_id),
+    -- При стендовом join_use_nulls=0 непарный UInt64 получает 0. Условие
+    -- не дает анонимной сессии превратиться в известного пользователя.
+    uniqExactIf(identities.user_id, identities.user_id != 0),
+    {load_id:String},
+    now64(3, 'UTC')
+FROM dds.session_v AS sessions
+GLOBAL LEFT JOIN dds.identity_map_v AS identities USING (client_id)
+GROUP BY sessions.session_date
+-- Ноль оставляет итоговую группировку на инициаторе: иначе каждый шард записал
+-- бы свою строку дня, а FINAL через границу шардов их не объединяет.
+SETTINGS distributed_foreground_insert = 1,
+    parallel_distributed_insert_select = 0;

sql/dm/revenue_daily_rebuild.sql

@@ -0,0 +1,55 @@
+-- DM: сборка выручки в донор перед заменой дневных партиций.
+-- DROP PARTITION ALL очищает локальные таблицы на каждом узле; причина не
+-- применять TRUNCATE разобрана в соседнем sql/dds/order_rebuild.sql.
+
+ALTER TABLE dm.revenue_daily_stage_rep ON CLUSTER clickstream_cluster
+DROP PARTITION ALL;
+
+INSERT INTO dm.revenue_daily_stage_dist
+(
+    report_date,
+    product_category,
+    orders,
+    units,
+    revenue,
+    aov,
+    _load_id,
+    _load_ts
+)
+SELECT
+    report_date,
+    product_category,
+    orders,
+    units,
+    CAST(revenue, 'Decimal(18, 2)'),
+    -- Обычное деление Decimal усекает доли копейки вместо округления;
+    -- запас масштаба сохраняет их до денежного результата.
+    CAST(
+        round(divideDecimal(revenue, toDecimal128(orders, 0), 6), 2),
+        'Decimal(18, 2)'
+    ),
+    {load_id:String},
+    now64(3, 'UTC')
+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'
+        AND order_date BETWEEN {first_day:Date} AND {last_day:Date}
+    GROUP BY
+        report_date,
+        product_category
+)
+-- Источник лежит по заказу, цель — по категории. Ноль заставляет инициатор
+-- слить частичные агрегаты и разложить готовые строки по ключу цели.
+SETTINGS distributed_foreground_insert = 1,
+    parallel_distributed_insert_select = 0;

sql/dm/revenue_daily_replace.sql

@@ -0,0 +1,8 @@
+-- DM: подмена одного дня выручки готовой партицией донора.
+--
+-- День приходит из результата revenue_daily_scope.sql, а не от пользователя.
+-- Jinja нужна потому, что ClickHouse не подставляет параметр в PARTITION.
+-- Замена идет на локальной таблице ON CLUSTER: граница атомарности и повтор
+-- разобраны на таком же шаге в sql/dds/order_replace.sql.
+ALTER TABLE dm.revenue_daily_rep ON CLUSTER clickstream_cluster
+REPLACE PARTITION '{{ params.day }}' FROM dm.revenue_daily_stage_rep;

sql/dm/revenue_daily_scope.sql

@@ -0,0 +1,39 @@
+-- DM: объем пересборки выручки.
+--
+-- Последний слепок ODS называет дни, состояние которых источник еще может
+-- изменить. Сам агрегат читает только DDS; ODS здесь нужен лишь для границ.
+-- Первый запуск и повтор после частичного отказа расширяют левую границу до
+-- первого дня оплаченных заказов, которого еще нет в витрине.
+-- Почему хвост начинается с первого пропуска и чем за это платит ODS,
+-- разобрано в sql/dds/order_scope.sql.
+--
+-- Файл кончается без точки с запятой: даг читает его одним запросом и ждет
+-- одну строку с границами.
+WITH
+    (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 minOrNull(report_date)
+        FROM (SELECT DISTINCT order_date AS report_date FROM dds.order_v WHERE status = 'paid')
+        -- Правая сторона тоже распределенная: GLOBAL собирает множество дней
+        -- на инициаторе и не превращает запрос в запрещенное двойное чтение.
+        WHERE report_date GLOBAL NOT IN (
+            SELECT report_date FROM dm.revenue_daily_dist
+        )
+    ) AS first_missing_day
+SELECT
+    least(window_start, coalesce(first_missing_day, window_start)) AS first_day,
+    window_end AS last_day
+-- Без ODS неизвестно окно. Без готового DDS донор останется пустым, а замена
+-- сотрет уже собранные дни выручки. Поэтому для прогона нужны оба источника.
+WHERE (SELECT count() FROM dds.order_v) > 0
+    AND (SELECT count() FROM ods.order_dist) > 0