feat(dds): добавлена модель заказа и пересборка окна заменой партиций

Зачем:
- Дать слою первую физическую сущность и показать идемпотентность, устроенную
  иначе, чем в приёме: заменой партиции, а не колонкой версии.
- Родить механику дневной пересборки на кластере — образец для сессий и витрин.

Что:
- Добавлены dds.order_rep/_dist, двойник-донор dds.order_stage_rep/_dist и
  публичный договор dds.order_v с разбором позиций в три массива.
- Добавлены запросы объёма прогона, сборки в двойник и замены партиции по дню.
- Добавлен даг dds_transform: группа задач order, замена — размноженной задачей.
- Окно изменяемости K пришло на сторону стенда якорем compose; владелец числа
  остаётся у генератора.
- Объём считается от позднейшего дня заказа и первого недостающего дня: ярлыки
  доставки ODS для этого непригодны, а максимум собранного дня прячет дыры.
- Записано, чего конвейер о себе не знает, и адресовано контуру DQ.

Проверка:
- make lint
- make smoke, make check-clickhouse, make check-services
- Прогон по стартовому миру: 1917 заказов, счёт равен ods.order_v, каждый заказ
  на шарде своего cityHash64(order_id), сумма позиций равна items_total.
- Повтор без нового слепка: содержимое то же, _load_id новый; после слепка
  пересобрано окно, день за окном не тронут.
- Снесённая руками партиция 2026-06-01 восстановлена побайтово.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

compose.yaml

@@ -48,6 +48,16 @@ x-kafka-target: &kafka-target
 
 x-world-starting-days: &world-starting-days "8"
 
+# Окно изменяемости заказа: слепок бэкенда несёт заказы последних K дней, за
+# окном заказ замерзает. Число принадлежит модельному миру, и владелец у него
+# один — generator/world.py (ORDER_WINDOW_DAYS): сделать окно настройкой
+# генератора значило бы разрешить менять физику мира снаружи, а опись мира
+# держится на её неизменности. Стенду копия нужна затем, что по этому же числу
+# трансформация DDS решает, какие дни пересобирать. Разойдутся — пересборка
+# промахнётся по краю окна, и заметит это не проверка, а менти со вчерашней
+# выручкой.
+x-order-window-days: &order-window-days "7"
+
 x-airflow-common: &airflow-common
   image: clickstream-airflow:local
   build:
@@ -77,6 +87,7 @@ x-airflow-common: &airflow-common
     STAND_NETWORK: ${COMPOSE_PROJECT_NAME}_default
     WORLD_STARTING_DAYS: *world-starting-days
     KAFKA_ORDERS_TOPIC: *orders-topic
+    ORDER_WINDOW_DAYS: *order-window-days
   volumes:
     - ./dags:/opt/airflow/dags:ro
     # Запросы дагов: даг называет файл, а обработчик DAG при разборе загружает

dags/dds_transform.py

@@ -0,0 +1,142 @@
+"""Трансформации слоя DDS: пересборка дней, которые ещё могут измениться.
+
+Даг собирает модель поверх принятого ODS. Первая его сущность — заказ; сессии
+и карта идентичностей встанут соседними группами задач.
+
+Расписания у дага нет и параметров тоже: объём каждого прогона он выводит сам
+из состояния слоёв, поэтому один и тот же запуск одинаково верен из триггера
+дневного конвейера, с кнопки менти и после падения
+(docs/architecture/etl.md, «Объём прогона»).
+
+Идемпотентность здесь устроена не так, как в приёме заказов, и в этом весь
+урок сущности. Приём гасит повтор колонкой версии: ReplacingMergeTree
+оставляет позднюю строку. Трансформация версий не знает — она собирает день
+заново в таблицу-двойник и атомарно подменяет им партицию цели. Поэтому
+повтор прогона не меняет делового содержимого слоя, а строки пересобранных
+дней получают новый _load_id: видно, каким запуском собран каждый день.
+
+Замена партиции идёт по локальным таблицам с ON CLUSTER: по Distributed
+операции с партициями отвечают кодом 48 (docs/architecture/storage.md, «Что
+проверено»). Форма двойника и цена ошибки в ключе шардирования разобраны в
+sql/ddl/45-dds-tables.sql.
+"""
+
+from __future__ import annotations
+
+import datetime
+import logging
+import os
+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"
+
+# Окно изменяемости заказа, модельные дни. Число мира, а не настройка стенда:
+# владелец — generator/world.py (ORDER_WINDOW_DAYS), который этим окном
+# собирает слепок. Здесь оно читается, чтобы хранилище знало, какие дни ещё
+# могут измениться. Разойдутся два объявления — пересборка промахнётся по краю
+# окна, поэтому оба названы в compose.yaml рядом и с этой оговоркой.
+ORDER_WINDOW_DAYS = int(os.environ["ORDER_WINDOW_DAYS"])
+
+
+@dag(
+    dag_id="dds_transform",
+    schedule=None,
+    start_date=START_DATE,
+    is_paused_upon_creation=False,
+    # Два прогона разом собирали бы один и тот же двойник: второй затёр бы
+    # чужую сборку между вставкой и заменой партиций.
+    max_active_runs=1,
+    template_searchpath=str(SQL_ROOT),
+    tags=["dds"],
+)
+def dds_transform():
+    """Пересобрать сущности DDS по текущему состоянию источников."""
+
+    @task_group(group_id="order")
+    def order():
+        """Заказ: дышащее окно плюс дни, которых в слое ещё нет."""
+
+        @task
+        def scope() -> dict[str, object]:
+            """Спросить у хранилища, какие дни собирает этот запуск."""
+            query = (SQL_ROOT / "dds" / "order_scope.sql").read_text(encoding="utf-8")
+            hook = ClickHouseHook(clickhouse_conn_id=CLICKHOUSE_CONNECTION)
+            row = hook.get_first(query, parameters={"window_days": ORDER_WINDOW_DAYS})
+            # Пустой ответ означает пустой источник: собирать нечего, и
+            # зеленеть на этом нельзя — прогон встаёт с внятной причиной.
+            if row is None:
+                raise AirflowException(
+                    "в ods.order нет ни одного заказа: собирать нечего; "
+                    "проверьте, дошёл ли слепок до хранилища"
+                )
+            first_day, last_day = row
+            days = [
+                (first_day + datetime.timedelta(days=shift)).isoformat()
+                for shift in range((last_day - first_day).days + 1)
+            ]
+            logging.info(
+                "объём пересборки: %s … %s, дней %d",
+                first_day,
+                last_day,
+                len(days),
+            )
+            return {
+                "first_day": first_day.isoformat(),
+                "last_day": last_day.isoformat(),
+                "days": days,
+            }
+
+        @task
+        def replacements(scope: dict[str, object]) -> list[dict[str, object]]:
+            """Разложить дни в аргументы размноженной задачи замены.
+
+            Шаг существует ради ограничения Airflow: размножать задачу можно
+            только по целому возврату другой задачи — ни по ключу словаря, ни
+            по карте поверх него. Отсюда отдельный возврат под один список.
+
+            День едет в запрос подстановкой Jinja, а не параметром ClickHouse:
+            в ALTER … PARTITION сервер параметр не подставляет — замерено,
+            отвечает синтаксической ошибкой. Значение при этом не внешнее, его
+            вернул сам ClickHouse запросом объёма.
+            """
+            return [{"params": {"day": day}} for day in scope["days"]]
+
+        days = scope()
+
+        rebuild = SQLExecuteQueryOperator(
+            task_id="rebuild",
+            conn_id=CLICKHOUSE_CONNECTION,
+            sql="dds/order_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="dds/order_replace.sql",
+            do_xcom_push=False,
+        ).expand_kwargs(replacements(days))
+
+        rebuild >> replace
+
+    order()
+
+
+dds_transform()

docs/architecture/etl.md

@@ -41,12 +41,30 @@
 
 Параметров у конвейера нет: объём он выводит сам из `world_position` и
 состояния слоёв. Сессии догоняют хвост — от последнего собранного дня до
-последнего прожитого; заказы пересобирают дышащее окно K от позиции; карта
+последнего прожитого; заказы пересобирают дышащее окно K и тот же хвост; карта
 идентичностей пересчитывается целиком; правила витрин — у
 [слоя DM](dm.md).
 Поэтому один и тот же запуск одинаково верен из триггера, с кнопки менти и
 после падения, а разгону на семь дней хватает одного прогона.
 
+Окно заказов отсчитывается от **последнего дня, о котором рассказал сам
+источник**, — от позднейшего дня заказа в ODS. Рядом напрашиваются ещё два
+способа, и оба отвергнуты по одной причине: они отвечают не на тот вопрос.
+
+Позиция мира говорит, какой день прожит, а не какой доехал: прогон дня D везёт
+слепок дня D − 1, да и позиция ставится раньше трансформации, так что после
+неудачного приёма она обгоняет хранилище. Слой обязан считать объём по тому,
+что у него есть, а не по тому, что мир о себе заявляет.
+
+Ярлыки доставки — `snapshot_date`, `_load_id`, `_load_ts` — не отвечают тоже, и
+это свойство ODS, а не недосмотр: слой схлопывает доставки, сохраняя деловые
+версии, поэтому у заказа, который не менялся, ярлык выжившей строки достаётся
+жребием ([приём заказов](orders/ingestion.md), «Роль ODS»). Замер на стартовом
+мире: последний слепок нёс все 1662 заказа, а его ярлык сохранили 1188 —
+отбор «дни, у которых `snapshot_date` равен последнему» потерял бы четверть
+заказов и молча пропустил бы дальний край окна, где заказы давно замёрзли.
+К времени создания это не относится: у всех версий заказа оно одно.
+
 ## Готовность источника
 
 Первый шаг `dds_transform` — дождаться, пока хвост топика `hits` дожёван:
@@ -67,6 +85,31 @@
 замены: полная перезапись в `ReplacingMergeTree` — так живут карта
 идентичностей и витрина `daily_traffic` ([слой DM](dm.md)).
 
+## Чего конвейер о себе не знает
+
+Трансформация собирает слой по правилу, а не сверяет его с источником. Отсюда
+четыре вещи, которых она не заметит, и у всех четырёх один адрес — контур DQ.
+
+- **Окно K разъехалось с источником.** Число живёт двумя копиями: у генератора
+  в `world.py` и у стенда в `compose.yaml`. Разойдутся — слой начнёт
+  недобирать край окна или переписывать замёрзший день, и оба прогона будут
+  зелёными.
+- **Пустой свежий день сдвигает край окна.** Если в позднейший день источника
+  не родилось ни одного заказа, окно съезжает на день назад
+  ([`sql/dds/order_scope.sql`](../../sql/dds/order_scope.sql)).
+- **Замена партиции прошла на одном шарде из двух.** День в слое есть — значит,
+  объём следующего прогона его не тронет, а слой останется собранным наполовину
+  из старого, наполовину из нового.
+- **Брак внутри `items` становится нулями.** Строгий приём внутрь позиций не
+  смотрит, разбор по типу молчит, и цена без числа доедет до витрины выручки
+  нулём ([`sql/dds/order_rebuild.sql`](../../sql/dds/order_rebuild.sql)).
+
+Общее у них одно, и оно же подсказывает лекарство: ответить может только
+сверка слоя с источником по дню — счёт заказов и позднейший `updated_at` с
+обеих сторон. Трансформации такую сверку не поручают намеренно: она собирает,
+а проверяет другой. Так две роли остаются различимы, и ни одна не
+притворяется другой.
+
 ## Не входит
 
 - Партиционная переобработка — переиграть произвольный прошлый день или

docs/architecture/storage.md

@@ -15,7 +15,9 @@ keeper, Kafka, каркас сервисов. Этап 2 собран: в `sql/d
 Тот же этап принёс первый справочник — словарь товаров из общего с генератором
 CSV-файла. Живёт он вне цепочки слоёв, в зоне `dic`, и читает не файл, а
 подложку над ним ([ADR 0012](../adr/0012-dictionary-home.md)). Этап 4 начат
-публичным договором события `dds.event_v` над точным состоянием ODS.
+публичным договором события `dds.event_v` над точным состоянием ODS, а следом
+получил первую физическую модель слоя — заказ: его пересобирает заменой
+партиций даг `dds_transform`.
 Дальше по тексту устройство описано так, как оно проектируется; построенное от
 заложенного отличает карта таблиц в конце.
 
@@ -129,6 +131,14 @@ FROM источник_dist`. Заводское значение `parallel_distr
 разложена по шардам по тому же ключу. Форму донора следует решать вместе с таким
 объектом, а не переносить на ODS заранее.
 
+Первым такой донор понадобился заказу DDS, и правило на нём видно целиком:
+`dds.order_stage_rep` объявлен через `AS dds.order_rep` и шардируется тем же
+`cityHash64(order_id)`. Совпадение структуры требует сама операция, а совпадение
+ключа — устройство замены: она идёт по локальным таблицам, шард за шардом, и
+строка, попавшая в доноре не на свой шард, при подмене пропала бы с одного шарда
+и задвоилась на другом. Доводы целиком — в
+[`sql/ddl/45-dds-tables.sql`](../../sql/ddl/45-dds-tables.sql).
+
 ## Служебные колонки
 
 Собственные колонки не повторяют имён виртуальных. Виртуальные даёт движок:
@@ -508,7 +518,8 @@ SQL-файлы исполняет штатный `SQLExecuteQueryOperator`. Пр
 | `20-ods-tables.sql` | типизированное событие, версии заказа и обе таблицы ошибок |
 | `30-ods-views.sql` | актуальные события, текущие заказы и матвью разбора в ODS |
 | `40-stg-views.sql` | матвью приёма: чтец в сырьё |
-| `50-dds-views.sql` | публичный договор события на языке модели DDS |
+| `45-dds-tables.sql` | физическая модель заказа и таблица-двойник, донор её партиций |
+| `50-dds-views.sql` | публичные договоры события и заказа на языке модели DDS |
 
 Справочники идут сразу за базами: зона `dic` не зависит ни от одного слоя, и
 правила порядка слоёв к ней не применяются.
@@ -567,6 +578,9 @@ SQL-файлы исполняет штатный `SQLExecuteQueryOperator`. Пр
 | `ods` | `ods.order_v` | текущая версия заказа на языке источника |
 | `ods` | `ods.order_errors_rep` / `_dist` | строки слепка, не прошедшие строгий приём |
 | `dds` | `dds.event_v` | событие на языке бизнес-модели без физической копии |
+| `dds` | `dds.order_rep` / `_dist` | заказ в текущем состоянии, партиция — день заказа |
+| `dds` | `dds.order_stage_rep` / `_dist` | двойник заказа: в нём день собирается перед заменой партиции |
+| `dds` | `dds.order_v` | публичный договор заказа |
 | `dic` | `dic.products_file` | чтец CSV-каталога, общего с генератором |
 | `dic` | `dic.products` | словарь товаров поверх подложки |
 
@@ -574,9 +588,10 @@ SQL-файлы исполняет штатный `SQLExecuteQueryOperator`. Пр
 `INSERT SELECT` одной задачи `parse_batch` из файла
 `sql/ods/order_parse.sql`.
 
-Слой DDS начат представлением `dds.event_v`; остальные его сущности и слой DM
-появятся в следующих тикетах этапа. Их состав задан разделом 7 мастер-спеки и
-переносится сюда по мере постройки.
+В слое DDS построены событие и заказ, и устроены они намеренно по-разному:
+событие остаётся представлением над ODS, заказ получил физическую модель.
+Сессии, карту идентичностей и слой DM принесут следующие тикеты этапа. Их
+состав задан разделом 7 мастер-спеки и переносится сюда по мере постройки.
 
 ## Что проверено
 
@@ -788,5 +803,32 @@ Kafka с пустым значением (ноль байт) и запись-н
 таблице. Это вычитано, а не измерено. На устройство приёма оговорка не влияет:
 служебные колонки мы заполняем выражением при любом ответе.
 
+**Проверка замены партиций 26 августа 2026 года (#151).** Context7 подтвердил
+договор `REPLACE PARTITION`: `ON CLUSTER` в синтаксисе есть, а источник и цель
+обязаны совпадать структурой, ключом партиции, ключом сортировки, первичным
+ключом и политикой хранения. На закреплённом ClickHouse 26.3 замена по локальным
+таблицам с `ON CLUSTER` прошла на обеих нодах, а сборка стартового мира в
+двойник разложила 1662 заказа строго по ключу: шард 1 получил только чётные
+`cityHash64(order_id)`, шард 2 — только нечётные.
+
+Тем же днём измерены ещё три поведения ClickHouse.
+
+- **Параметр в `ALTER … PARTITION` не подставляется.** Запрос с `{day:Date}` в
+  выражении партиции отвечает кодом 62, синтаксической ошибкой; литерал
+  проходит. Отсюда единственное место в репозитории, где значение попадает в
+  запрос подстановкой Jinja, а не параметром ClickHouse
+  ([`sql/dds/order_replace.sql`](../../sql/dds/order_replace.sql)).
+- **`TRUNCATE` — отдельное право, в `ALTER` оно не входит.** Роль `etl`, у
+  которой есть `ALTER ON dds.*`, получает на `TRUNCATE` код 497, а
+  `ALTER TABLE … DROP PARTITION ALL` на той же таблице проходит. Двойник
+  поэтому чистится снятием партиций: у операции нашлась форма, на которую право
+  уже выдано, и расширять доступ не понадобилось.
+- **`JSONExtract` читает `Decimal` прямо из строки JSON.** Цена на проводе —
+  строка с двумя знаками (`"1289.99"`), и разбор в
+  `Array(Tuple(sku String, qty UInt32, price Decimal(18, 2)))` даёт число без
+  промежуточного `toDecimal64`. Проверено на живом слепке; честность разбора
+  сторожит инвариант бэкенда — сумма «цена × количество» совпала с `items_total`
+  у всех 1662 заказов.
+
 **Сказано по памяти, проверки нет.** Группа пуста: оба утверждения, ждавшие
 матвью разбора, закрыты опытами при исполнении #43 и переехали выше.

sql/ddl/45-dds-tables.sql

@@ -0,0 +1,95 @@
+-- DDS: физическая модель заказа и донор её партиций.
+--
+-- Зерно — заказ в текущем состоянии, ключ order_id. Историю версий слой не
+-- хранит: у неё нет читателя ни в одной витрине, а ODS её и не обещает
+-- (docs/architecture/dds/order.md, «Зерно»).
+--
+-- Идемпотентность здесь устроена иначе, чем в приёме, и это главный урок
+-- сущности. В ODS повтор гасит колонка версии: ReplacingMergeTree(updated_at)
+-- оставляет позднюю строку. Здесь колонки версии нет вовсе — повтор гасит
+-- замена партиции целиком: день собирается заново и подменяется одним
+-- атомарным движением. Поэтому каждая строка таблицы — ровно один заказ, а
+-- чтению не нужен FINAL.
+
+-- Локальная таблица шарда.
+--
+-- Партиция — order_date, день заказа в поясе счётчика. Партиция дневная не
+-- ради размера куска, а потому что день — единица пересборки: пока заказ
+-- внутри окна изменяемости, его день «дышит» и собирается заново, за окном
+-- замерзает. Ключ партиции и есть контракт трансформации с хранилищем.
+--
+-- Ключ сортировки — order_id: единственный устойчивый ключ сущности, он же
+-- ключ шардирования ниже. Колонки версии в движке нет: см. шапку файла.
+CREATE TABLE IF NOT EXISTS dds.order_rep ON CLUSTER clickstream_cluster
+(
+    order_id String,
+    user_id UInt64,
+    status LowCardinality(String),
+    -- День заказа посчитан один раз при загрузке, по образцу EventDate у
+    -- события: суточные срезы группируются по готовой дате и пояса не
+    -- упоминают (docs/architecture/storage.md, «Часовые пояса»).
+    order_date Date,
+    -- Времена лежат местными: DDS обслуживает человека с дашбордом, и пояс
+    -- пересчитан один раз здесь, а не в каждом отчёте. Линза названа в типе,
+    -- поэтому пояс сервера на эти колонки не влияет.
+    created_at DateTime64(3, 'Europe/Samara'),
+    updated_at DateTime64(3, 'Europe/Samara'),
+    items_total Decimal(18, 2),
+    discount Decimal(18, 2),
+    delivery Decimal(18, 2),
+    total Decimal(18, 2),
+    -- Позиции — три массива одной длины: артикул, количество, цена. Родная
+    -- для колоночного хранилища денормализация: позиция не живёт без заказа и
+    -- своего ключа не имеет, а разворачивает массивы читатель — ARRAY JOIN в
+    -- витрине выручки. Форма записи выбрана по дому: широкое событие хранит
+    -- повторяющиеся группы Метрики такими же параллельными колонками
+    -- (productID, productPrice, productQuantity в ods.event_rep), а не Nested.
+    item_sku Array(String),
+    item_quantity Array(UInt32),
+    item_price Array(Decimal(18, 2)),
+    -- Координаты собственной загрузки слоя: каким запуском и когда собран
+    -- день. При замене партиции они одинаковы у всех строк дня. Ярлыки ODS
+    -- сюда не переезжают — почему, разобрано в доке модели.
+    _load_id String,
+    _load_ts DateTime64(3, 'UTC')
+)
+ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/{database}/{table}', '{replica}')
+PARTITION BY order_date
+ORDER BY order_id;
+
+-- Лицо загрузки и диагностики. Ключ шардирования — cityHash64(order_id), тот
+-- же, что у ods.order_dist: заказ остаётся на одном шарде на всём пути через
+-- слои, и соединения по зерну заказа обходятся без GLOBAL
+-- (docs/architecture/storage.md, «Раскладка по шардам»).
+CREATE TABLE IF NOT EXISTS dds.order_dist ON CLUSTER clickstream_cluster
+AS dds.order_rep
+ENGINE = Distributed('clickstream_cluster', 'dds', 'order_rep', cityHash64(order_id));
+
+-- Донор партиций: таблица-двойник, в которой день собирается перед подменой.
+--
+-- Двойник нужен потому, что REPLACE PARTITION берёт готовый кусок из другой
+-- таблицы, а не из результата запроса. Собрать день прямо в цель нельзя:
+-- вставка добавила бы строки к уже лежащим, и повтор прогона удвоил бы день.
+--
+-- Форма двойника задана правилом донора: структура, ключ партиции, ключ
+-- сортировки, первичный ключ и политика хранения обязаны совпадать с целью —
+-- это требование самой операции (проверено по документации ClickHouse,
+-- sql-reference/statements/alter/partition). Отсюда AS dds.order_rep вместо
+-- повторённого списка колонок: расхождение становится невозможным по
+-- построению и не держится на внимательности того, кто правит.
+--
+-- Своё имя в keeper двойник получает от макроса {table}, поэтому две таблицы
+-- не делят путь репликации.
+CREATE TABLE IF NOT EXISTS dds.order_stage_rep ON CLUSTER clickstream_cluster
+AS dds.order_rep
+ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/{database}/{table}', '{replica}')
+PARTITION BY order_date
+ORDER BY order_id;
+
+-- Двойник шардируется тем же ключом, что цель, и это не украшение: REPLACE
+-- PARTITION идёт по локальным таблицам, шард за шардом. Ляг строка заказа в
+-- доноре на другой шард, чем её место в цели, — подмена перенесла бы её не
+-- туда, и заказ пропал бы с одного шарда и задвоился на другом.
+CREATE TABLE IF NOT EXISTS dds.order_stage_dist ON CLUSTER clickstream_cluster
+AS dds.order_stage_rep
+ENGINE = Distributed('clickstream_cluster', 'dds', 'order_stage_rep', cityHash64(order_id));

sql/ddl/50-dds-views.sql

@@ -65,3 +65,37 @@ SELECT
     productQuantity AS product_quantity,
     productEventType AS product_event_type
 FROM ods.event_v;
+
+-- DDS: публичный договор заказа на языке бизнес-модели.
+--
+-- В отличие от события, у заказа под представлением лежит физическая модель:
+-- разбор позиций и пересчёт пояса стоят денег, и платить их при каждом чтении
+-- значило бы собирать модель заново на лету (docs/architecture/dds/order.md,
+-- «Отклонённые варианты»). Представление здесь ничего не вычисляет — оно
+-- только называет колонки, которые слой обещает читателю.
+--
+-- FINAL нет и не будет: строка на заказ одна по построению, повтор гасит
+-- замена партиции, а не колонка версии.
+--
+-- Служебные колонки перечислены наравне с деловыми — это координаты
+-- собственной загрузки слоя, и расследование происшествия начинается с них:
+-- по _load_id видно, какой запуск собрал день.
+CREATE VIEW IF NOT EXISTS dds.order_v ON CLUSTER clickstream_cluster
+AS
+SELECT
+    order_id,
+    user_id,
+    status,
+    order_date,
+    created_at,
+    updated_at,
+    items_total,
+    discount,
+    delivery,
+    total,
+    item_sku,
+    item_quantity,
+    item_price,
+    _load_id,
+    _load_ts
+FROM dds.order_dist;

sql/dds/order_rebuild.sql

@@ -0,0 +1,89 @@
+-- DDS: сборка дней заказа в таблицу-двойник перед заменой партиций.
+--
+-- Двойник очищается целиком в начале сборки, а не в конце: собранное остаётся
+-- лежать до следующего запуска, и разбор происшествия видит ровно то, чем
+-- подменили день. Очистка обязана быть первой ещё и по делу — вставка поверх
+-- прежних строк удвоила бы день в доноре, а REPLACE PARTITION перенёс бы
+-- удвоение в цель.
+--
+-- Дни собираются одной вставкой, а не по дню за раз: источник читается один
+-- раз на весь отрезок. Кто раскладывает строки по шардам — вставка через
+-- Distributed, ключ cityHash64(order_id) — разобрано в DDL двойника.
+--
+-- Чистит двойник DROP PARTITION ALL, а не TRUNCATE, и это не вкусовщина:
+-- TRUNCATE — отдельное право, роли etl оно не выдано, а снятие всех партиций
+-- входит в ALTER, который у неё есть (ADR 0007). Замерено: TRUNCATE отвечает
+-- ролью etl кодом 497, DROP PARTITION ALL проходит. Расширять права ради
+-- операции, для которой уже есть разрешённая форма, значило бы платить
+-- доступом за синоним.
+
+ALTER TABLE dds.order_stage_rep ON CLUSTER clickstream_cluster DROP PARTITION ALL;
+
+INSERT INTO dds.order_stage_dist
+(
+    order_id,
+    user_id,
+    status,
+    order_date,
+    created_at,
+    updated_at,
+    items_total,
+    discount,
+    delivery,
+    total,
+    item_sku,
+    item_quantity,
+    item_price,
+    _load_id,
+    _load_ts
+)
+-- Позиции разбираются здесь один раз, на границе ODS → DDS. Цена на проводе —
+-- строка с двумя знаками, и JSONExtract читает её прямо в Decimal: замерено на
+-- живом слепке, промежуточный toDecimal64 не нужен.
+--
+-- Разбор по типу молчалив, и знать об этом надо заранее. Строгий приём ODS
+-- проверил только, что items — массив, внутрь он не смотрел; здесь тоже никто
+-- не смотрит. Замерено: qty вида "oops" даёт 0, пропущенная цена даёт 0,
+-- скаляр вместо объекта выпадает из массива целиком. На каноническом слепке
+-- генератора такого не бывает, и сумма «цена × количество» сходится с
+-- items_total у всех заказов, но сходится она по свойству источника, а не
+-- потому, что её здесь кто-то проверяет. Постоянное место такой проверки —
+-- контур DQ этапа 4; пока она делается руками при приёмке.
+WITH JSONExtract(
+        items, 'Array(Tuple(sku String, qty UInt32, price Decimal(18, 2)))'
+    ) AS lines
+SELECT
+    order_id,
+    user_id,
+    status,
+    -- День заказа и оба времени переезжают в пояс счётчика: DDS обслуживает
+    -- человека с дашбордом, и пояс называется здесь в последний раз.
+    toDate(toTimeZone(created_at, 'Europe/Samara')),
+    toTimeZone(created_at, 'Europe/Samara'),
+    toTimeZone(updated_at, 'Europe/Samara'),
+    items_total,
+    discount,
+    delivery,
+    total,
+    arrayMap(line -> line.sku, lines),
+    arrayMap(line -> line.qty, lines),
+    arrayMap(line -> line.price, lines),
+    -- Координаты запуска, собравшего день: у всех строк дня они одни.
+    {load_id:String},
+    now64(3, 'UTC')
+FROM ods.order_v
+WHERE toDate(toTimeZone(created_at, 'Europe/Samara'))
+        BETWEEN {first_day:Date} AND {last_day:Date}
+    -- Вторая граница не меняет отбор, а открывает источнику отсечение
+    -- партиций: ODS нарезан по toDate(created_at) в UTC, и условие на
+    -- вычисленный день заказа для этого ключа непрозрачно — сервер прочёл бы
+    -- все партиции. Сутки пояса счётчика ложатся ровно на две даты UTC,
+    -- поэтому день слева берётся с запасом.
+    AND toDate(created_at) BETWEEN {first_day:Date} - 1 AND {last_day:Date}
+-- Тот же ноль, что в приёме заказов: заводская двойка исполнила бы вставку
+-- локально на каждом шарде, минуя ключ шардирования цели, и строки унаследовали
+-- бы раскладку источника. Ключи ODS и DDS здесь совпадают, и раскладка вышла бы
+-- верной — но по совпадению, а не по контракту (docs/architecture/storage.md,
+-- «Раскладка по шардам», капкан #137).
+SETTINGS distributed_foreground_insert = 1,
+    parallel_distributed_insert_select = 0;

sql/dds/order_replace.sql

@@ -0,0 +1,28 @@
+-- DDS: подмена одного дня заказов собранной партицией двойника.
+--
+-- Это и есть идемпотентность слоя. Повтор прогона безопасен, а «дыхание»
+-- статусов внутри окна выражается сменой всей партиции, а не правкой строк.
+--
+-- Замена идёт по локальным таблицам, шард за шардом: ON CLUSTER рассылает её
+-- на обе ноды, а по Distributed то же самое не работает — операции с
+-- партициями отвечают кодом 48 (docs/architecture/storage.md, «Что
+-- проверено»). Правильность пошардовой замены держится на том, что двойник
+-- шардирован тем же ключом, что цель, — см. sql/ddl/45-dds-tables.sql.
+--
+-- Отсюда граница обещания: атомарность здесь пошардовая. Внутри шарда подмена
+-- мгновенна — читатель видит либо прежний день целиком, либо новый; но две
+-- локальные замены независимы, и в зазоре между ними чтение через Distributed
+-- соберёт ответ из нового дня одного шарда и прежнего дня другого. Зазор
+-- короткий, а отказ одной из двух замен растягивает его до повторного прогона.
+--
+-- Пустая партиция двойника — законный случай, а не сбой: день, в котором
+-- заказов нет, снимает день из цели, и слой перестаёт показывать то, чего в
+-- источнике больше нет.
+--
+-- День подставляет Jinja, а не параметр ClickHouse: в ALTER … PARTITION
+-- сервер параметров не подставляет — замерено, отвечает синтаксической
+-- ошибкой. Общее правило репозитория «параметры передаются отдельно, а не
+-- строкой» здесь не нарушено по существу: значение пришло из ответа самого
+-- ClickHouse на запрос объёма, а не снаружи стенда.
+ALTER TABLE dds.order_rep ON CLUSTER clickstream_cluster
+REPLACE PARTITION '{{ params.day }}' FROM dds.order_stage_rep;

sql/dds/order_scope.sql

@@ -0,0 +1,65 @@
+-- DDS: объём пересборки заказов — с какого дня по какой собирает этот запуск.
+--
+-- Параметров у дага нет: объём он выводит сам из состояния слоёв
+-- (docs/architecture/etl.md, «Объём прогона»). Ответ — две даты, границы
+-- непрерывного отрезка; список дней между ними разворачивает даг. Пустой
+-- источник даёт пустой ответ, и даг на нём останавливается с внятной ошибкой
+-- вместо зелёного прогона, собравшего ничего.
+--
+-- Файл кончается без точки с запятой: даг читает его целиком и шлёт одним
+-- запросом, дописывая формат ответа, — точка с запятой сделала бы из этого
+-- мультизапрос, который ClickHouse не принимает.
+--
+-- Объём складывается из двух частей. Первая — дышащее окно: K последних дней
+-- источника, состояние которых слепок ещё может изменить. Вторая — хвост от
+-- первого дня, которого в слое нет: первая сборка застаёт слой пустым и
+-- собирает весь стартовый мир, а после разгона мира пачкой слой догоняет
+-- уехавший вперёд источник.
+--
+-- **Хвост считается от первого недостающего дня, а не от последнего
+-- собранного.** Разница не косметическая: замены партиций идут отдельными
+-- задачами и падают порознь, поэтому упавший день оставляет дыру, пока
+-- соседние дни собираются дальше. Считай мы прогресс по максимуму собранного
+-- дня, дыра позади окна не заросла бы никогда, а прогоны продолжали бы
+-- зеленеть над щербатым слоем. Цена честного ответа — проход по дню заказа в
+-- ODS; она того стоит.
+--
+-- Край окна берётся из делового поля, а не из ярлыков доставки: время
+-- создания заказа одинаково у всех его версий, а `snapshot_date` и `_load_id`
+-- выжившей строки достаются жребием (docs/architecture/etl.md, «Объём
+-- прогона», и docs/architecture/orders/ingestion.md, «Роль ODS»).
+--
+-- Известная неточность края: если в самый свежий день источника не родилось
+-- ни одного заказа, окно съезжает на день назад — лишний замёрзший день
+-- пересоберётся впустую, а пустой день не тронется. Данные от этого не
+-- страдают, меняется только ярлык загрузки лишнего дня. Точный ответ здесь
+-- дала бы дата слепка, но она и есть тот самый жребий.
+WITH
+    -- FINAL не нужен: время рождения заказа версии не различают.
+    (SELECT max(toDate(toTimeZone(created_at, 'Europe/Samara')))
+        FROM ods.order_dist) AS last_source_day,
+    (SELECT min(toDate(toTimeZone(created_at, 'Europe/Samara')))
+        FROM ods.order_dist) AS first_source_day,
+    -- GLOBAL обязателен: обе таблицы распределённые, и обычный NOT IN сверял
+    -- бы дни источника с локальной частью слоя на каждом шарде — день, чьи
+    -- заказы легли на соседний шард, показался бы недостающим.
+    -- minOrNull отличает «дыр нет» от даты по умолчанию.
+    (
+        SELECT minOrNull(day)
+        FROM (
+            SELECT DISTINCT toDate(toTimeZone(created_at, 'Europe/Samara')) AS day
+            FROM ods.order_dist
+        )
+        WHERE day GLOBAL NOT IN (SELECT order_date FROM dds.order_dist)
+    ) AS first_missing_day,
+    last_source_day - ({window_days:UInt16} - 1) AS window_start
+SELECT
+    -- Дыр нет — начинаем с окна; есть — с первой из них, если она старше окна.
+    -- greatest держит левый край не раньше первого дня, известного источнику:
+    -- у молодого мира окно шире самого мира.
+    greatest(
+        least(window_start, coalesce(first_missing_day, window_start)),
+        first_source_day
+    ) AS first_day,
+    last_source_day AS last_day
+WHERE (SELECT count() FROM ods.order_dist) > 0