Merge pull request 'feat(dds): добавлена карта идентичностей' (#168) from 153-dds-identity-map into main

Reviewed-on: #168

dags/dds_transform.py

@@ -1,18 +1,18 @@
-"""Трансформации слоя DDS: сборка модели заменой дневных партиций.
+"""Трансформации слоя DDS: сборка моделей.
 
-Даг собирает модель поверх принятого ODS. Сущностей у него две — заказ и
-сессия, зависимости по данным между ними нет; карта идентичностей встанет
-соседней группой задач.
+Заказ и сессия собираются заменой дневных партиций. Карта идентичностей
+пересчитывается целиком после заказа, потому что читает его текущую модель.
 
 Расписания у дага нет и параметров тоже: объём каждого прогона он выводит сам
 из состояния слоёв и позиции мира, поэтому один и тот же запуск одинаково
 верен из триггера дневного конвейера, с кнопки менти и после падения
 (docs/architecture/etl.md, «Объём прогона»).
 
-Повтор прогона гасит не колонка версии, а замена партиции целиком: день
-собирается в таблицу-двойник и подменяет собой партицию цели. Почему слой
-устроен так — sql/ddl/45-dds-tables.sql; механика самой подмены —
-sql/dds/order_replace.sql.
+Заказ и сессия гасят повтор заменой партиции целиком: день собирается в
+таблицу-двойник и подменяет собой партицию цели. Карта пишет полный набор пар
+новой версией, а точное чтение даёт FINAL. Замена партиции разобрана в
+sql/dds/order_replace.sql; точное чтение карты описано в
+sql/ddl/50-dds-views.sql.
 """
 
 from __future__ import annotations
@@ -229,8 +229,18 @@ def dds_transform():
 
         rebuild >> replace
 
-    order()
+    orders = order()
     session()
 
+    identity_map = SQLExecuteQueryOperator(
+        task_id="identity_map",
+        conn_id=CLICKHOUSE_CONNECTION,
+        sql="dds/identity_map_rebuild.sql",
+        parameters={"load_id": "{{ run_id }}"},
+        do_xcom_push=False,
+    )
+
+    orders >> identity_map
+
 
 dds_transform()

docs/architecture/storage.md

@@ -587,6 +587,8 @@ SQL-файлы исполняет штатный `SQLExecuteQueryOperator`. Пр
 | `dds` | `dds.session_rep` / `_dist` | визит, нарезанный слоем; партиция — день сессии |
 | `dds` | `dds.session_stage_rep` / `_dist` | двойник сессии: в нём день собирается перед заменой партиции |
 | `dds` | `dds.session_v` | публичный договор сессии |
+| `dds` | `dds.identity_map_rep` / `_dist` | известная пара куки с пользователем; таблицы полного пересчёта |
+| `dds` | `dds.identity_map_v` | точный публичный договор карты идентичностей с `FINAL` |
 | `dic` | `dic.products_file` | чтец CSV-каталога, общего с генератором |
 | `dic` | `dic.products` | словарь товаров поверх подложки |
 
@@ -594,11 +596,11 @@ SQL-файлы исполняет штатный `SQLExecuteQueryOperator`. Пр
 `INSERT SELECT` одной задачи `parse_batch` из файла
 `sql/ods/order_parse.sql`.
 
-В слое DDS построены событие, заказ и сессия, и устроены они намеренно
-по-разному: событие остаётся представлением над ODS, у заказа и сессии —
-физические модели. Карту идентичностей и слой DM принесут следующие тикеты
-этапа. Их состав задан разделом 7 мастер-спеки и переносится сюда по мере
-постройки.
+В слое DDS построены событие, заказ, сессия и карта идентичностей. Событие
+остаётся представлением над ODS; заказ и сессия гасят повтор заменой партиции;
+карта пишет полный набор пар в версионную таблицу. Слой DM принесут следующие
+тикеты этапа. Его состав задан разделом 7 мастер-спеки и переносится сюда по
+мере постройки.
 
 ## Что проверено
 
@@ -818,6 +820,11 @@ Kafka с пустым значением (ноль байт) и запись-н
 двойник разложила 1662 заказа строго по ключу: шард 1 получил только чётные
 `cityHash64(order_id)`, шард 2 — только нечётные.
 
+**Проверка карты идентичностей 26 августа 2026 года (#153).** Context7
+подтвердил версию в `ReplacingMergeTree` и точное чтение через `FINAL`. Два
+полных прогона оставили 3180 физических строк и 1590 пар в `_v`; все пары
+подтверждены покупкой и заказом, а строк на чужом шарде нет.
+
 Тем же днём измерены ещё три поведения ClickHouse.
 
 - **Параметр в `ALTER … PARTITION` не подставляется.** Запрос с `{day:Date}` в

sql/ddl/45-dds-tables.sql

@@ -1,11 +1,7 @@
--- DDS: физические модели слоя и доноры их партиций.
+-- DDS: физические модели слоя и доноры партиций.
 --
--- Идемпотентность здесь устроена иначе, чем в приёме, и это главный урок обеих
--- сущностей файла. В ODS повтор гасит колонка версии: ReplacingMergeTree по
--- snapshot_date оставляет строку позднейшего слепка. Здесь колонки
--- версии нет вовсе — повтор гасит замена партиции целиком: день собирается
--- заново и подменяется одним атомарным движением. Поэтому строка таблицы
--- ровно одна на своё зерно, а чтению не нужен FINAL.
+-- Заказ и сессия гасят повтор заменой дневной партиции, без колонки версии.
+-- Карта идентичностей пересчитывается целиком и пишет новую версию пары.
 
 -- Локальная таблица заказа.
 --
@@ -170,3 +166,33 @@ ORDER BY (client_id, started_at);
 CREATE TABLE IF NOT EXISTS dds.session_stage_dist ON CLUSTER clickstream_cluster
 AS dds.session_stage_rep
 ENGINE = Distributed('clickstream_cluster', 'dds', 'session_stage_rep', cityHash64(client_id));
+
+-- DDS: известная связь куки трекера с пользователем магазина.
+--
+-- Полный запуск снова пишет все известные пары. Физические повторы до
+-- фонового слияния допустимы: _load_ts выбирает последнюю версию пары.
+-- Партиции нет, потому что карта пересчитывается целиком и её зерно не
+-- связано с днём.
+CREATE TABLE IF NOT EXISTS dds.identity_map_rep ON CLUSTER clickstream_cluster
+(
+    client_id UInt64,
+    user_id UInt64,
+    _load_id String,
+    _load_ts DateTime64(3, 'UTC')
+)
+ENGINE = ReplicatedReplacingMergeTree(
+    '/clickhouse/tables/{shard}/{database}/{table}',
+    '{replica}',
+    _load_ts
+)
+ORDER BY (client_id, user_id);
+
+-- Карта лежит по ключу будущего соединения с событиями и сессиями.
+CREATE TABLE IF NOT EXISTS dds.identity_map_dist ON CLUSTER clickstream_cluster
+AS dds.identity_map_rep
+ENGINE = Distributed(
+    'clickstream_cluster',
+    'dds',
+    'identity_map_rep',
+    cityHash64(client_id)
+);

sql/ddl/50-dds-views.sql

@@ -134,3 +134,17 @@ SELECT
     _load_id,
     _load_ts
 FROM dds.session_dist;
+
+-- DDS: точная известная связь куки с пользователем магазина.
+--
+-- Полный пересчёт оставляет физические версии одной пары до слияния частей.
+-- FINAL прячет это устройство от читателя; явный список колонок не даёт
+-- служебным изменениям таблицы случайно расширить договор.
+CREATE VIEW IF NOT EXISTS dds.identity_map_v ON CLUSTER clickstream_cluster
+AS
+SELECT
+    client_id,
+    user_id,
+    _load_id,
+    _load_ts
+FROM dds.identity_map_dist FINAL;

sql/dds/identity_map_rebuild.sql

@@ -0,0 +1,46 @@
+-- DDS: полный пересчёт известных пар куки с пользователем.
+--
+-- События разложены по client_id; заказы по order_id. Ключи соединения не
+-- ко-локированы, поэтому правило слоя требует явный GLOBAL. С подзапросами с
+-- обеих сторон ClickHouse 26.3 выполняет соединение один раз на инициаторе
+-- (docs/research/2026-08-22-distributed-join.md).
+
+INSERT INTO dds.identity_map_dist
+(
+    client_id,
+    user_id,
+    _load_id,
+    _load_ts
+)
+SELECT
+    client_id,
+    user_id,
+    {load_id:String},
+    now64(3, 'UTC')
+FROM
+(
+    SELECT DISTINCT
+        purchase.client_id,
+        orders.user_id
+    FROM
+    (
+        SELECT
+            client_id,
+            -- По договору события purchase_id содержит ровно один номер
+            -- заказа: одно подтверждение оформляет один заказ (спека, раздел 4).
+            purchase_id[1] AS order_id
+        FROM dds.event_v
+        WHERE event_type = 'purchase'
+    ) AS purchase
+    GLOBAL INNER JOIN
+    (
+        SELECT
+            order_id,
+            user_id
+        FROM dds.order_v
+    ) AS orders ON purchase.order_id = orders.order_id
+)
+-- В этой форме ClickHouse 26.3 уже пишет через инициатор. Явный ноль фиксирует
+-- раскладку по ключу цели, если форма запроса или план изменятся (капкан #137).
+SETTINGS distributed_foreground_insert = 1,
+    parallel_distributed_insert_select = 0;