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

- Зачем:
  - реализованы требования issue #153 к мосту между кукой и пользователем.
- Что:
  - добавлены физическая карта, точное представление и полный пересчёт через GLOBAL JOIN.
  - задача identity_map поставлена после сборки заказов в dds_transform.
  - устройство и живые проверки карты записаны в документации хранилища.
- Проверка:
  - make lint, make smoke, make check-clickhouse, make check-services.
  - три запуска dds_transform: 4770 физических строк и 1590 уникальных пар в _v.

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()