DWH-1179

--- Отформатированный код из задачи. Добавил только `lesson_id` в сортировку для того, чтобы создать детерминированную сортировку ---
CREATE TEMP TABLE original_report AS
WITH needed_lessons AS (
    SELECT
        tutor_id,
        lesson_id,
        lesson_state,
        start_time,
        pupil_id,
        -- Пометим флагом своеврем/несвоеврем отмены и прогулы. Всегда опираемся на статус урока первым делом.
        -- Если статус cancelled и первый отменивший - препод, ставим флаг своевременной отмены.
        COALESCE(lesson_state = 'cancelled' AND cancellation_made_by_role = 'tutor', FALSE)::INT AS is_ok_tutor_cancel,
        -- Если статус прогул П/обоих и есть попытка отмены до начала урока - поздняя отмена преподавателем.
        COALESCE(lesson_state IN ('tutor_no_show', 'both_no_show')
                 AND first_cancellation_datetime < start_time, FALSE)::INT AS is_late_tutor_cancel,
        -- Если статус прогул П/обоих и нет попытки отмены до начала урока - прогул преподавателя.
        COALESCE(lesson_state IN ('tutor_no_show', 'both_no_show')
                 AND (first_cancellation_datetime >= start_time OR first_cancellation_datetime IS NULL), FALSE)::INT AS is_no_tutor_cancel,
        -- Найдем последние 50 уроков преподавателя в любом статусе.
        ROW_NUMBER() OVER (PARTITION BY tutor_id ORDER BY start_time DESC, lesson_id) AS rn_tutor_desc
    FROM
        mart.events_with_lessons
    WHERE
        manager_of_intro_lesson IS FALSE
        AND is_regular_lesson + is_intro_lesson = 1
        AND lesson_state != 'new'
        AND lesson_state IS NOT NULL
        AND is_tetrika_lesson IS TRUE
        AND start_time < '2024-12-01'
),
-- Посчитаем количество отмен с одним учеником.
finding_cancels_with_same_pupil AS (
    SELECT
        tutor_id,
        lesson_id,
        pupil_id,
        start_time,
        lesson_state,
        -- Найдем количество "плохих" уроков с одним учеником.
        COUNT(lesson_id) OVER (PARTITION BY tutor_id, pupil_id ORDER BY start_time DESC, lesson_id) AS bad_lessons_with_pupil_count
    FROM
        needed_lessons
    WHERE
        (is_ok_tutor_cancel + is_late_tutor_cancel + is_no_tutor_cancel) > 0
),
all_cancels_together AS (
    SELECT
        needed_lessons.tutor_id,
        needed_lessons.lesson_id,
        needed_lessons.lesson_state,
        needed_lessons.start_time,
        needed_lessons.pupil_id,
        needed_lessons.is_ok_tutor_cancel,
        needed_lessons.is_late_tutor_cancel,
        needed_lessons.is_no_tutor_cancel,
        needed_lessons.rn_tutor_desc,
        -- Где есть информация о прогулах с одним учеником, возьмем её. Где нет - подставим 0.
        COALESCE(finding_cancels_with_same_pupil.bad_lessons_with_pupil_count, 0) AS bad_lessons_with_pupil_count,
        -- Итоговую стоимость каждого прогула рассчитываем так:
        -- своевременная отмена - вес 1, несвоевременная отмена - вес 1.5, прогул - вес 2.
        -- Если несколько плохих уроков с одним учеником, то каждый последующий плохой урок весит больше:
        -- 1-й как 1, второй - как 2, третий - как 3.
        is_ok_tutor_cancel * 1 * COALESCE(finding_cancels_with_same_pupil.bad_lessons_with_pupil_count, 0) +
        is_late_tutor_cancel * 1.5 * COALESCE(finding_cancels_with_same_pupil.bad_lessons_with_pupil_count, 0) +
        is_no_tutor_cancel * 2 * COALESCE(finding_cancels_with_same_pupil.bad_lessons_with_pupil_count, 0) AS final_score
    FROM
        needed_lessons
    LEFT JOIN
        finding_cancels_with_same_pupil USING (lesson_id)
)
SELECT *
FROM all_cancels_together;


--- Тестовая таблица, данные ограничены фильтром `clean.events.start_time < '2024-12-01'` ---
DROP TABLE IF EXISTS dwh_report;
CREATE TEMP TABLE dwh_report AS
SELECT
    dwh_1179_core_tutor_lesson_counter.tutor_id,
    dwh_1179_core_tutor_lesson_counter.lesson_id,
    dwh_1179_core_tutor_lesson_counter.lesson_started_at,
    dwh_1179_core_tutor_lesson_counter.tutor_lesson_number_desc,
    dwh_1179_core_tutor_lesson_counter.lesson_number_without_tutor_present,
    dwh_1179_core_tutor_lesson_counter.discipline_score
FROM sandbox.dwh_1179_core_tutor_lesson_counter
JOIN core.lessons
    ON lessons.lesson_id = dwh_1179_core_tutor_lesson_counter.lesson_id;

---  Проверяем есть ли лишние записи ---
SELECT *
FROM original_report
FULL JOIN dwh_report
    ON original_report.lesson_id = dwh_report.lesson_id
WHERE TRUE = TRUE
    AND original_report.lesson_id IS NULL
    AND dwh_report.tutor_lesson_number_desc IS NOT NULL;

--- Проверяем матчатся ли поля с кодом из задачи ---
SELECT
    original_report.lesson_id,
    dwh_report.lesson_id,

    original_report.start_time,
    dwh_report.lesson_started_at,

    original_report.rn_tutor_desc,
    dwh_report.tutor_lesson_number_desc,

    original_report.bad_lessons_with_pupil_count,
    dwh_report.lesson_number_without_tutor_present,

    original_report.final_score,
    dwh_report.discipline_score

FROM original_report
LEFT JOIN dwh_report
    ON original_report.lesson_id = dwh_report.lesson_id
WHERE
    original_report.rn_tutor_desc IS DISTINCT FROM dwh_report.tutor_lesson_number_desc
    OR original_report.bad_lessons_with_pupil_count IS DISTINCT FROM dwh_report.lesson_number_without_tutor_present
    OR original_report.final_score IS DISTINCT FROM dwh_report.discipline_score;