airflow-greenplum/sql/dds/dim_routes_load.sql

-- Загрузка DDS dim_routes: SCD2 с hashdiff.

-- Учебный комментарий: Используем TEMP TABLE для подготовки дельты.
-- Это избавляет от дублирования логики hashdiff в UPDATE и INSERT блоках.
CREATE TEMP TABLE tmp_routes_src ON COMMIT DROP AS
SELECT
    route_no,
    departure_airport,
    arrival_airport,
    airplane_code,
    days_of_week,
    departure_time,
    duration,
    md5(
        COALESCE(departure_airport, '') || '|' ||
        COALESCE(arrival_airport, '') || '|' ||
        COALESCE(airplane_code, '') || '|' ||
        COALESCE(days_of_week::TEXT, '') || '|' ||
        COALESCE(departure_time::TEXT, '') || '|' ||
        COALESCE(duration::TEXT, '')
    ) AS hashdiff,
    ROW_NUMBER() OVER (PARTITION BY route_no ORDER BY validity DESC) AS rn
FROM ods.routes;

-- Statement 1: Закрыть устаревшие версии (valid_to = текущая дата).
UPDATE dds.dim_routes AS d
SET valid_to   = CURRENT_DATE,
    updated_at = now(),
    _load_id   = '{{ run_id }}',
    _load_ts   = now()
FROM tmp_routes_src AS s
WHERE s.rn = 1
    AND d.route_bk = s.route_no
    AND d.valid_to IS NULL
    AND d.hashdiff <> s.hashdiff;

-- Statement 1.1: Закрыть "исчезнувшие" маршруты.
UPDATE dds.dim_routes AS d
SET valid_to   = CURRENT_DATE,
    updated_at = now(),
    _load_id   = '{{ run_id }}',
    _load_ts   = now()
WHERE d.valid_to IS NULL
    AND NOT EXISTS (
        SELECT 1
        FROM tmp_routes_src AS s
        WHERE s.rn = 1
            AND s.route_no = d.route_bk
    );

-- Statement 2: Вставить новые версии (для изменённых и новых route_no).
WITH max_sk AS (
    -- Учебный комментарий: Генерация SK через MAX() + ROW_NUMBER()
    -- Этот подход работает безопасно только потому, что Airflow запускает
    -- джобы загрузки для одной таблицы строго последовательно (concurrency=1).
    SELECT COALESCE(MAX(route_sk), 0) AS v
    FROM dds.dim_routes
)
INSERT INTO dds.dim_routes (
    route_sk,
    route_bk,
    departure_airport,
    arrival_airport,
    airplane_code,
    departure_city,
    arrival_city,
    airplane_model,
    total_seats,
    days_of_week,
    departure_time,
    duration,
    hashdiff,
    valid_from,
    valid_to,
    created_at,
    updated_at,
    _load_id,
    _load_ts
)
SELECT
    (SELECT v FROM max_sk) + ROW_NUMBER() OVER (ORDER BY s.route_no)::INTEGER,
    s.route_no,
    s.departure_airport,
    s.arrival_airport,
    s.airplane_code,
    dep.city,
    arr.city,
    air.model,
    air.total_seats,
    s.days_of_week,
    s.departure_time,
    s.duration,
    s.hashdiff,
    CASE
        WHEN EXISTS (
            SELECT 1
            FROM dds.dim_routes AS d2
            WHERE d2.route_bk = s.route_no
        ) THEN CURRENT_DATE
        ELSE '1900-01-01'::DATE
    END AS valid_from,
    NULL,
    now(),
    now(),
    '{{ run_id }}',
    now()
FROM tmp_routes_src AS s
LEFT JOIN dds.dim_airports AS dep ON s.departure_airport = dep.airport_bk
LEFT JOIN dds.dim_airports AS arr ON s.arrival_airport = arr.airport_bk
LEFT JOIN dds.dim_airplanes AS air ON s.airplane_code = air.airplane_bk
WHERE s.rn = 1
    AND NOT EXISTS (
        SELECT 1
        FROM dds.dim_routes AS d
        WHERE d.route_bk = s.route_no
            AND d.valid_to IS NULL
            AND d.hashdiff = s.hashdiff
    );

-- Фаза 3: Обновление денормализованных атрибутов (refresh).
-- Нужна для SCD1-изменений в dim_airports/dim_airplanes (напр. переименование города).
-- Обновляем все версии (и текущие, и исторические), т.к. измерения SCD1.
-- При этом мы не перезаписываем _load_id и _load_ts, чтобы не размывать lineage версий.
UPDATE dds.dim_routes AS d
SET departure_city = dep.city,
    arrival_city   = arr.city,
    airplane_model = air.model,
    total_seats    = air.total_seats,
    updated_at     = now()
FROM dds.dim_airports AS dep,
     dds.dim_airports AS arr,
     dds.dim_airplanes AS air
WHERE dep.airport_bk = d.departure_airport
    AND arr.airport_bk = d.arrival_airport
    AND air.airplane_bk = d.airplane_code
    AND (
        d.departure_city IS DISTINCT FROM dep.city
        OR d.arrival_city IS DISTINCT FROM arr.city
        OR d.airplane_model IS DISTINCT FROM air.model
        OR d.total_seats IS DISTINCT FROM air.total_seats
    );

ANALYZE dds.dim_routes;