airflow-greenplum/docs/design/bookings_ods_design.md

ODS Layer: эталонный учебный план реализации (v3)

Контекст

STG-слой уже реализован как учебный эталон:

• данные из bookings-db читаются через PXF;
• в STG бизнес-колонки хранятся как TEXT;
• загрузка и DQ работают батчами (_load_id = {{ run_id }}).

Этот документ фиксирует простую и каноничную реализацию ODS для менти.

---

1) Что считаем эталоном для ODS

1.1. Роль ODS в этом стенде

ODS в учебном проекте — это:

• типизированные и очищенные данные;
• одна актуальная запись на бизнес-ключ;
• удобный слой для последующей сборки DDS/DM.

1.2. Что делаем, что не делаем

Делаем в ODS:

• приведение типов (TEXT -> TIMESTAMPTZ/NUMERIC/INT/BOOLEAN/...);
• дедупликацию внутри батча;
• UPSERT (SCD Type 1): обновляем текущую запись при изменении, вставляем новые.
• для snapshot-справочников (airports, airplanes, routes, seats) синхронизацию ключей: удаляем из ODS записи, которых нет в выбранном stg_batch_id.

Не делаем в ODS (в базовом эталоне):

• SCD Type 2 с периодами действия;
• сложную обработку late-arriving/backdated событий;
• отдельный DQ-слой с хранением результатов.

1.3. Где хранится история изменений

• История «как приходили данные» уже сохраняется в STG (append + _load_id).
• Историзацию измерений (SCD2) показываем позже в DDS (как в учебной статье dwh-modeling).

Итог: ODS = текущий слой (current state), простой и понятный.

---

2) Нейминг служебных полей (консистентно с de-roadmap)

Источник правил: docs/design/naming_conventions.md.

В ODS используем такие техполя:

• _load_id TEXT NOT NULL — идентификатор загрузки (берём stg_batch_id);
• _load_ts TIMESTAMP NOT NULL DEFAULT now() — время загрузки в ODS;
• event_ts TIMESTAMP — время события из источника (если у сущности оно есть).

2.1. Маппинг из текущего STG

• stg._load_id -> ods._load_id
• stg._load_ts не переносим 1:1; в ODS пишем собственный ods._load_ts = now()
• stg.event_ts -> ods.event_ts (для транзакционных таблиц)

2.2. Почему так

• нейминг совпадает с учебной статьёй (_load_id, _load_ts);
• студентам проще переносить паттерн между проектами;
• разделяем «когда событие произошло» (event_ts) и «когда загрузили в слой» (_load_ts).

---

3) Гранулярность и бизнес-ключи ODS

Таблица	Зерно	Бизнес-ключ
ods.airports	1 строка = аэропорт	airport_code
ods.airplanes	1 строка = самолёт	airplane_code
ods.routes	1 строка = версия маршрута	(route_no, validity)
ods.seats	1 строка = место в самолёте	(airplane_code, seat_no)
ods.bookings	1 строка = бронирование	book_ref
ods.tickets	1 строка = билет	ticket_no
ods.flights	1 строка = рейс	flight_id
ods.segments	1 строка = сегмент билета	(ticket_no, flight_id)
ods.boarding_passes	1 строка = посадочный на сегмент	(ticket_no, flight_id)

Критично для эталона:

• routes — составной ключ (route_no, validity): один route_no может иметь несколько версий с разными периодами действия. flights ссылается только на route_no (без validity), поэтому при join в DDS нужно будет выбирать подходящую версию маршрута;
• boarding_passes — составной ключ (ticket_no, flight_id).

---

4) Схема ODS-таблиц (v1, без SCD2)

Ниже — учебный минимум колонок. При необходимости можно добавлять бизнес-атрибуты без изменения паттерна загрузки.

4.1. Справочники

ods.airports

airport_code  TEXT NOT NULL
airport_name  TEXT NOT NULL  -- русское название, извлечено из JSON (поле 'ru')
city          TEXT NOT NULL  -- русское название, извлечено из JSON (поле 'ru')
country       TEXT NOT NULL  -- русское название, извлечено из JSON (поле 'ru')
coordinates   TEXT
timezone      TEXT NOT NULL
_load_id      TEXT NOT NULL
_load_ts      TIMESTAMP NOT NULL DEFAULT now()
DISTRIBUTED BY (airport_code)

Примечание о нормализации JSON: Источник (bookings.airports_data) хранит мультиязычные поля (airport_name, city, country) как JSON-объекты вида {"en": "...", "ru": "..."}. На уровне ODS выполняется нормализация: извлекается только русский вариант (->>'ru'). Это упрощает downstream-логику (DDS/DM не нужно знать о внутренней структуре JSON) и соответствует принципу "типизированные данные" на уровне ODS.

ods.airplanes

airplane_code TEXT NOT NULL
model         TEXT NOT NULL  -- русское название модели, извлечено из JSON (поле 'ru')
range_km      INTEGER
speed_kmh     INTEGER
_load_id      TEXT NOT NULL
_load_ts      TIMESTAMP NOT NULL DEFAULT now()
DISTRIBUTED BY (airplane_code)

Примечание о нормализации JSON: Источник (bookings.aircrafts_data) хранит поле model как JSON-объект {"en": "...", "ru": "..."}. На уровне ODS извлекается только русский вариант (->>'ru') для консистентности с airports.

ods.routes

route_no           TEXT NOT NULL
validity           TEXT NOT NULL
departure_airport  TEXT NOT NULL
arrival_airport    TEXT NOT NULL
airplane_code      TEXT NOT NULL
days_of_week       TEXT
departure_time     TIME              -- STG: scheduled_time (TEXT → TIME)
duration           INTERVAL          -- STG: duration (TEXT → INTERVAL)
_load_id           TEXT NOT NULL
_load_ts           TIMESTAMP NOT NULL DEFAULT now()
DISTRIBUTED BY (route_no)

ods.seats

airplane_code   TEXT NOT NULL
seat_no         TEXT NOT NULL
fare_conditions TEXT NOT NULL
_load_id        TEXT NOT NULL
_load_ts        TIMESTAMP NOT NULL DEFAULT now()
DISTRIBUTED BY (airplane_code)

4.2. Транзакционные

ods.bookings

book_ref      TEXT NOT NULL
book_date     TIMESTAMP WITH TIME ZONE NOT NULL
total_amount  NUMERIC(10,2) NOT NULL
event_ts      TIMESTAMP
_load_id      TEXT NOT NULL
_load_ts      TIMESTAMP NOT NULL DEFAULT now()
DISTRIBUTED BY (book_ref)

ods.tickets

ticket_no       TEXT NOT NULL
book_ref        TEXT NOT NULL
passenger_id    TEXT NOT NULL
passenger_name  TEXT NOT NULL
is_outbound     BOOLEAN
event_ts        TIMESTAMP
_load_id        TEXT NOT NULL
_load_ts        TIMESTAMP NOT NULL DEFAULT now()
DISTRIBUTED BY (ticket_no)

В STG tickets распределены по book_ref для co-location с bookings (append-only, lookup не нужен). В ODS нужен UPSERT по бизнес-ключу ticket_no, поэтому распределяем по нему — иначе каждый lookup потребует redistribute motion.

ods.flights

flight_id            INTEGER NOT NULL
route_no             TEXT NOT NULL
status               TEXT NOT NULL
scheduled_departure  TIMESTAMP WITH TIME ZONE
scheduled_arrival    TIMESTAMP WITH TIME ZONE
actual_departure     TIMESTAMP WITH TIME ZONE
actual_arrival       TIMESTAMP WITH TIME ZONE
event_ts             TIMESTAMP
_load_id             TEXT NOT NULL
_load_ts             TIMESTAMP NOT NULL DEFAULT now()
DISTRIBUTED BY (flight_id)

ods.segments

ticket_no        TEXT NOT NULL
flight_id        INTEGER NOT NULL
fare_conditions  TEXT NOT NULL
segment_amount   NUMERIC(10,2)
event_ts         TIMESTAMP
_load_id         TEXT NOT NULL
_load_ts         TIMESTAMP NOT NULL DEFAULT now()
DISTRIBUTED BY (ticket_no)

ods.boarding_passes

ticket_no      TEXT NOT NULL
flight_id      INTEGER NOT NULL
seat_no        TEXT NOT NULL
boarding_no    INTEGER
boarding_time  TIMESTAMP WITH TIME ZONE
event_ts       TIMESTAMP
_load_id       TEXT NOT NULL
_load_ts       TIMESTAMP NOT NULL DEFAULT now()
DISTRIBUTED BY (ticket_no)

Примечание: в учебном варианте не опираемся на физические PK/FK-constraint в Greenplum, а проверяем целостность через DQ-скрипты.

4.3. Маппинг STG → ODS (колонки с изменениями)

Большинство бизнес-колонок переносятся 1:1 с приведением типа (TEXT → ...). Ниже — только те, где происходит переименование или нетривиальное преобразование:

Таблица	STG колонка	ODS колонка	ODS тип	Комментарий
airplanes	range	range_km	INTEGER	Явное указание единиц (naming conv.)
airplanes	speed	speed_kmh	INTEGER	Явное указание единиц (naming conv.)
routes	scheduled_time	departure_time	TIME	Уточнение смысла
routes	duration	duration	INTERVAL	Только cast, без rename
tickets	outbound	is_outbound	BOOLEAN	Префикс is_ для boolean (naming conv.)
segments	price	segment_amount	NUMERIC(10,2)	Уточнение: сумма сегмента, не цена билета
flights	flight_id	flight_id	INTEGER	Только cast TEXT → INT
segments	flight_id	flight_id	INTEGER	Только cast TEXT → INT
boarding_passes	flight_id	flight_id	INTEGER	Только cast TEXT → INT
все транзакционные	event_ts	event_ts	TIMESTAMP	Прямой перенос (канон)
все	_load_id	_load_id	TEXT	Прямой перенос (канон)

Пример каста с переименованием в SQL (в CTE):

s.range::INTEGER       AS range_km,
s.speed::INTEGER       AS speed_kmh,
s.outbound::BOOLEAN    AS is_outbound,
s.price::NUMERIC(10,2) AS segment_amount

---

5) Контракт батча для ODS

Чтобы ODS был воспроизводимым, в каждом запуске используем один фиксированный stg_batch_id.

5.1. Зачем фиксировать stg_batch_id

На проде ETL-оркестратор всегда явно передаёт downstream-задачам идентификатор батча, который прошёл все проверки. Это гарантирует:

• воспроизводимость: повторный запуск обработает тот же снимок данных;
• изоляцию: ODS не подхватит «сырой» батч, который ещё не прошёл DQ в STG;
• отладку: по _load_id в ODS легко найти исходные данные в STG.

5.2. Как resolve stg_batch_id в DAG

В DAG bookings_to_gp_ods первым запускается PythonOperator, который определяет stg_batch_id и кладёт его в XCom:

import logging
from airflow.operators.python import PythonOperator
from airflow.providers.postgres.hooks.postgres import PostgresHook

log = logging.getLogger(__name__)

GREENPLUM_CONN_ID = "greenplum_conn"

def _resolve_stg_batch_id(**context):
    """Определяем stg_batch_id: из dag_run.conf или последний согласованный snapshot-батч."""
    conf = context["dag_run"].conf or {}
    stg_batch_id = conf.get("stg_batch_id")

    if not stg_batch_id:
        # Берём _load_id, который присутствует во всех snapshot-таблицах STG:
        # airports, airplanes, routes, seats. Это защищает от частично успешных запусков.
        hook = PostgresHook(postgres_conn_id=GREENPLUM_CONN_ID)
        result = hook.get_first(
            '''
            WITH candidate_batches AS (
                SELECT _load_id FROM stg.airports  WHERE _load_id IS NOT NULL GROUP BY _load_id
                INTERSECT
                SELECT _load_id FROM stg.airplanes WHERE _load_id IS NOT NULL GROUP BY _load_id
                INTERSECT
                SELECT _load_id FROM stg.routes    WHERE _load_id IS NOT NULL GROUP BY _load_id
                INTERSECT
                SELECT _load_id FROM stg.seats     WHERE _load_id IS NOT NULL GROUP BY _load_id
            ),
            batch_ready AS (
                SELECT
                    c._load_id,
                    GREATEST(
                        (SELECT MAX(_load_ts) FROM stg.airports  a WHERE a._load_id = c._load_id),
                        (SELECT MAX(_load_ts) FROM stg.airplanes a WHERE a._load_id = c._load_id),
                        (SELECT MAX(_load_ts) FROM stg.routes    r WHERE r._load_id = c._load_id),
                        (SELECT MAX(_load_ts) FROM stg.seats     s WHERE s._load_id = c._load_id)
                    ) AS ready_dttm
                FROM candidate_batches c
            )
            SELECT _load_id
            FROM batch_ready
            ORDER BY ready_dttm DESC
            LIMIT 1
            '''
        )
        stg_batch_id = result[0] if result and result[0] else None

    if not stg_batch_id:
        raise ValueError(
            "stg_batch_id не найден: передайте в conf или сначала выполните bookings_to_gp_stage"
        )

    log.info("Используем stg_batch_id = %s", stg_batch_id)
    return stg_batch_id  # автоматически попадёт в XCom как return_value

resolve_batch = PythonOperator(
    task_id="resolve_stg_batch_id",
    python_callable=_resolve_stg_batch_id,
)

5.3. Как используем в SQL

Во всех sql/ods/*_load.sql и sql/ods/*_dq.sql значение stg_batch_id подставляется через Jinja-шаблон:

WHERE _load_id = '{{ ti.xcom_pull(task_ids="resolve_stg_batch_id") }}'::text

Для читаемости в DAG можно вынести шаблон в константу:

STG_BATCH_ID = "{{ ti.xcom_pull(task_ids='resolve_stg_batch_id') }}"

В ODS записываем:

• _load_id = <stg_batch_id> — чтобы связать ODS-запись с STG-батчом;
• _load_ts = now() — фактическое время загрузки в ODS.

Это простой и понятный паттерн: один запуск ODS = один снимок STG-батча.

---

6) SQL-паттерны загрузки (SCD1 / UPSERT)

Стиль SQL: в ODS-скриптах используем CTE (Common Table Expressions) вместо вложенных подзапросов — CTE нагляднее, проще для чтения и отладки.

6.1. Шаблон для справочника (пример airports_load.sql)

Важно: CTE действует в рамках одного SQL-statement. UPDATE и INSERT — два отдельных statement, поэтому CTE src дублируется в каждом. Это не ошибка, а необходимость синтаксиса SQL.

-- Statement 1: UPDATE существующих записей (SCD1 — перезапись при изменении)
WITH src AS (
    SELECT
        airport_code,
        airport_name,
        city,
        country,
        coordinates,
        timezone
    FROM stg.airports
    WHERE _load_id = '{{ ti.xcom_pull(task_ids="resolve_stg_batch_id") }}'::text
)
UPDATE ods.airports AS o
SET airport_name = s.airport_name,
    city         = s.city,
    country      = s.country,
    coordinates  = s.coordinates,
    timezone     = s.timezone,
    _load_id     = '{{ ti.xcom_pull(task_ids="resolve_stg_batch_id") }}'::text,
    _load_ts     = now()
FROM src AS s
WHERE o.airport_code = s.airport_code
  AND (
      -- IS DISTINCT FROM — NULL-safe аналог <>:
      -- при NULL с одной стороны <> вернёт NULL (не обновит),
      -- а IS DISTINCT FROM вернёт TRUE (обновит корректно).
      o.airport_name IS DISTINCT FROM s.airport_name OR
      o.city         IS DISTINCT FROM s.city OR
      o.country      IS DISTINCT FROM s.country OR
      o.coordinates  IS DISTINCT FROM s.coordinates OR
      o.timezone     IS DISTINCT FROM s.timezone
  );

-- Statement 2: INSERT новых записей
WITH src AS (
    SELECT
        airport_code,
        airport_name,
        city,
        country,
        coordinates,
        timezone
    FROM stg.airports
    WHERE _load_id = '{{ ti.xcom_pull(task_ids="resolve_stg_batch_id") }}'::text
)
INSERT INTO ods.airports (
    airport_code, airport_name, city, country, coordinates, timezone,
    _load_id, _load_ts
)
SELECT
    s.airport_code, s.airport_name, s.city, s.country, s.coordinates, s.timezone,
    '{{ ti.xcom_pull(task_ids="resolve_stg_batch_id") }}'::text, now()
FROM src s
WHERE NOT EXISTS (
    SELECT 1
    FROM ods.airports o
    WHERE o.airport_code = s.airport_code
);

-- Statement 3: DELETE ключей, которых нет в snapshot текущего батча
WITH src_keys AS (
    SELECT DISTINCT airport_code
    FROM stg.airports
    WHERE _load_id = '{{ ti.xcom_pull(task_ids="resolve_stg_batch_id") }}'::text
)
DELETE FROM ods.airports o
WHERE NOT EXISTS (
    SELECT 1
    FROM src_keys s
    WHERE s.airport_code = o.airport_code
);

-- Обновляем статистику для оптимизатора запросов Greenplum
ANALYZE ods.airports;

6.2. Шаблон для транзакции (пример bookings_load.sql)

В транзакционных таблицах в одном STG-батче может быть несколько записей с одинаковым бизнес-ключом (например, обновления). Дедуплицируем через ROW_NUMBER() — стандартный и явный паттерн, часто встречающийся на собеседованиях и в DE-курсах.

-- Statement 1: UPDATE существующих записей
WITH src AS (
    SELECT
        book_ref,
        book_date::TIMESTAMP WITH TIME ZONE AS book_date,
        total_amount::NUMERIC(10,2)         AS total_amount,
        event_ts,
        ROW_NUMBER() OVER (
            PARTITION BY book_ref
            ORDER BY event_ts DESC NULLS LAST, _load_ts DESC
        ) AS rn
    FROM stg.bookings
    WHERE _load_id = '{{ ti.xcom_pull(task_ids="resolve_stg_batch_id") }}'::text
)
UPDATE ods.bookings AS o
SET book_date      = s.book_date,
    total_amount   = s.total_amount,
    event_ts       = s.event_ts,
    _load_id       = '{{ ti.xcom_pull(task_ids="resolve_stg_batch_id") }}'::text,
    _load_ts       = now()
FROM src AS s
WHERE s.rn = 1
  AND o.book_ref = s.book_ref
  AND (
      o.book_date    IS DISTINCT FROM s.book_date OR
      o.total_amount IS DISTINCT FROM s.total_amount
  );

-- Statement 2: INSERT новых записей
WITH src AS (
    SELECT
        book_ref,
        book_date::TIMESTAMP WITH TIME ZONE AS book_date,
        total_amount::NUMERIC(10,2)         AS total_amount,
        event_ts,
        ROW_NUMBER() OVER (
            PARTITION BY book_ref
            ORDER BY event_ts DESC NULLS LAST, _load_ts DESC
        ) AS rn
    FROM stg.bookings
    WHERE _load_id = '{{ ti.xcom_pull(task_ids="resolve_stg_batch_id") }}'::text
)
INSERT INTO ods.bookings (
    book_ref, book_date, total_amount, event_ts,
    _load_id, _load_ts
)
SELECT
    s.book_ref, s.book_date, s.total_amount, s.event_ts,
    '{{ ti.xcom_pull(task_ids="resolve_stg_batch_id") }}'::text, now()
FROM src s
WHERE s.rn = 1
  AND NOT EXISTS (
      SELECT 1
      FROM ods.bookings o
      WHERE o.book_ref = s.book_ref
  );

-- Обновляем статистику для оптимизатора запросов Greenplum
ANALYZE ods.bookings;

6.3. Идемпотентность паттерна

• Для инкрементальных таблиц паттерн UPDATE + INSERT WHERE NOT EXISTS — натурально идемпотентен: повторный запуск с тем же stg_batch_id не создаст дублей и не потеряет данные.
• Для snapshot-справочников идемпотентность сохраняется паттерном UPDATE + INSERT + DELETE not in snapshot: повторный запуск приводит ODS к тому же состоянию.

6.4. Поведение при пустом батче

• для инкрементальных таблиц (bookings, tickets, flights, segments, boarding_passes) пустой батч допустим;
• для snapshot-справочников (airports, routes) пустой батч считаем ошибкой. На main airplanes и seats — студенческие заглушки (load/dq = SELECT 1;), таблицы будут пустыми до реализации студентом.

---

7) DQ-проверки ODS (минимум, но строго)

Каждый DQ-скрипт должен:

• быть привязан к stg_batch_id (через XCom, как в load-скриптах);
• делать RAISE EXCEPTION при нарушении;
• давать понятную подсказку в тексте ошибки.

7.1. Обязательные проверки

1. Нет дублей по бизнес-ключу в ODS.

2. Покрытие батча: все ключи из STG текущего батча присутствуют в ODS.

3. Обязательные поля не NULL/не пустые.

4. Батч не пустой для эталонных snapshot-справочников (airports, routes): если STG-батч оказался пустым — это ошибка (источник недоступен или PXF не работает). На main airplanes_dq.sql и seats_dq.sql — студенческие заглушки.

5. Ссылочная целостность в ODS:

• tickets.book_ref -> bookings.book_ref
• flights.route_no -> routes.route_no (упрощённая проверка: наличие route_no в routes, без учёта конкретной версии validity. Это допустимо, потому что в источнике flights ссылается на route_no, а не на конкретную версию маршрута. При downstream join (DDS) нужно будет учитывать период validity)
• segments.ticket_no -> tickets.ticket_no
• segments.flight_id -> flights.flight_id
• boarding_passes (ticket_no, flight_id) -> segments (ticket_no, flight_id)

7.2. Пример проверки покрытия батча

SELECT COUNT(*)
FROM (
    SELECT DISTINCT book_ref
    FROM stg.bookings
    WHERE _load_id = '{{ ti.xcom_pull(task_ids="resolve_stg_batch_id") }}'::text
) s
WHERE NOT EXISTS (
    SELECT 1
    FROM ods.bookings o
    WHERE o.book_ref = s.book_ref
);

Ожидаемый результат: 0.

---

8) Структура файлов

Каждый *_ddl.sql начинается с CREATE SCHEMA IF NOT EXISTS ods; (по аналогии с STG DDL).

sql/ods/
├── airports_ddl.sql
├── airports_load.sql
├── airports_dq.sql
├── airplanes_ddl.sql
├── airplanes_load.sql
├── airplanes_dq.sql
├── routes_ddl.sql
├── routes_load.sql
├── routes_dq.sql
├── seats_ddl.sql
├── seats_load.sql
├── seats_dq.sql
├── bookings_ddl.sql
├── bookings_load.sql
├── bookings_dq.sql
├── tickets_ddl.sql
├── tickets_load.sql
├── tickets_dq.sql
├── flights_ddl.sql
├── flights_load.sql
├── flights_dq.sql
├── segments_ddl.sql
├── segments_load.sql
├── segments_dq.sql
├── boarding_passes_ddl.sql
├── boarding_passes_load.sql
└── boarding_passes_dq.sql

sql/ddl_gp.sql         (+ подключение sql/ods/*_ddl.sql)

airflow/dags/
├── bookings_ods_ddl.py
└── bookings_to_gp_ods.py

docs/bookings_to_gp_ods.md
Makefile                (ddl-gp включает ODS DDL)
tests/test_dags_smoke.py (+ smoke для 2 новых DAG)

---

9) DAG bookings_to_gp_ods: учебный граф зависимостей

Принцип: у каждой сущности строго load -> dq, и только после dq разрешаем downstream.

Корневой таск resolve_stg_batch_id определяет батч (см. секцию 5.2), после чего две параллельных ветки стартуют одновременно:

resolve_stg_batch_id
  ├-> load_ods_bookings  -> dq_ods_bookings -> load_ods_tickets -> dq_ods_tickets ──────────────────┐
  ├-> load_ods_airports  -> dq_ods_airports  ─┐                                                     │
  ├-> load_ods_airplanes -> dq_ods_airplanes ─┐
  │                                          ├-> load_ods_routes -> dq_ods_routes                  │
  └->                                        └-> load_ods_seats  -> dq_ods_seats                    │
                                                                                                     │
                                                  dq_ods_routes -> load_ods_flights -> dq_ods_flights │
                                                                                                     │
                                            dq_ods_flights + dq_ods_tickets -> load_ods_segments -> dq_ods_segments
                                                                                                     │
                                                              dq_ods_segments -> load_ods_boarding_passes -> dq_ods_boarding_passes
                                                                                                     │
                                                        [dq_ods_boarding_passes, dq_ods_seats] -> finish_ods_summary

Зависимости (по FK):

• tickets после bookings (FK: book_ref);
• routes после airports (FK: departure_airport, arrival_airport). На ветке solution также зависит от airplanes (FK: airplane_code);
• seats после airplanes (FK: airplane_code);
• flights после routes (FK: route_no);
• segments после flights и tickets (FK: flight_id, ticket_no);
• boarding_passes после segments (FK: ticket_no, flight_id).

Справочники (airports, airplanes) и транзакции (bookings) не зависят друг от друга в ODS — данные уже в STG. Поэтому они стартуют параллельно после resolve_stg_batch_id.

---

10) Порядок реализации

1. Подготовить DDL в sql/ods/*_ddl.sql.
2. Подключить sql/ods/*_ddl.sql в общий sql/ddl_gp.sql.
3. Использовать существующий Makefile-таргет ddl-gp для STG+ODS.
4. Создать DAG bookings_ods_ddl.py.
5. Реализовать sql/ods/*_load.sql (SCD1 UPSERT).
6. Реализовать sql/ods/*_dq.sql.
7. Создать DAG bookings_to_gp_ods.py (с параметром stg_batch_id).
8. Дописать smoke-тесты DAG в tests/test_dags_smoke.py.
9. Описать запуск и проверки в docs/bookings_to_gp_ods.md.

---

11) Критерии готовности (Definition of Done)

Готово, если:

1. Оба новых DAG парсятся и проходят smoke-тесты (make test).
2. make ddl-gp создаёт объекты STG+ODS без ошибок.
3. Для тестового stg_batch_id ODS-загрузка завершается успешно.
4. Все DQ-задачи зелёные и реально валят DAG при искусственной ошибке.
5. В ODS нет дублей по бизнес-ключам.
6. Нейминг техполей ODS консистентен с учебной статьёй: _load_id, _load_ts, event_ts.

---

12) Как проверять вручную

make up
make ddl-gp          # создать STG+ODS-объекты
# Trigger bookings_to_gp_stage (загрузить STG)
# Передать stg_batch_id в conf (рекомендуется) или дать ODS DAG выбрать
# последний согласованный batch автоматически.
# Trigger bookings_to_gp_ods с conf: {"stg_batch_id": "<значение>"}
make gp-psql

Проверочные SQL:

-- 1) Дубликаты в ODS (пример bookings)
SELECT book_ref, COUNT(*)
FROM ods.bookings
GROUP BY 1
HAVING COUNT(*) > 1;

-- 2) Покрытие текущего STG-батча в ODS
SELECT COUNT(*)
FROM (
    SELECT DISTINCT book_ref
    FROM stg.bookings
    WHERE _load_id = '<stg_batch_id>'
) s
WHERE NOT EXISTS (
    SELECT 1
    FROM ods.bookings o
    WHERE o.book_ref = s.book_ref
);

-- 3) Ссылочная целостность tickets -> bookings
SELECT COUNT(*)
FROM ods.tickets t
WHERE NOT EXISTS (
    SELECT 1
    FROM ods.bookings b
    WHERE b.book_ref = t.book_ref
);

Ожидаемо: все три запроса возвращают 0 проблемных строк.

---

13) Что будет следующим шагом

После стабилизации ODS:

• строим DDS;
• показываем SCD2 на измерениях DDS (valid_from/valid_to, created_at/updated_at) по тому же неймингу, который уже знаком студентам из dwh-modeling.