Полировка кода

2026-02-28 22:12:22 +03:00
parent 33110eed48
commit d6fed5c7b9
6 changed files with 105 additions and 30 deletions
@@ -25,5 +25,11 @@ CREATE TABLE IF NOT EXISTS stg.bookings (
    batch_id          TEXT NOT NULL
 )
 WITH (appendonly=true, orientation=row, compresstype=zlib, compresslevel=1)
+-- Ключ распределения: book_ref
+-- Обоснование: book_ref — это уникальный идентификатор бронирования.
+-- Использование book_ref обеспечивает:
+-- 1. Равномерное распределение данных по сегментам (book_ref имеет высокую кардинальность)
+-- 2. Co-location данных bookings и tickets при JOIN по book_ref
+-- 3. Оптимизацию запросов, которые фильтруют или группируют по book_ref
 DISTRIBUTED BY (book_ref);

@@ -9,6 +9,8 @@ DECLARE
    v_prev_ts     timestamp;
    v_src_count   bigint;
    v_stg_count   bigint;
+    v_dup_count   bigint;
+    v_null_amount_count bigint;
 BEGIN
    -- Опорная метка: максимум src_created_at_ts среди предыдущих батчей
    SELECT max(src_created_at_ts)
@@ -42,6 +44,33 @@ BEGIN
            v_stg_count;
    END IF;

+    -- Проверка на дубликаты book_ref
+    SELECT COUNT(*) - COUNT(DISTINCT book_ref)
+    INTO v_dup_count
+    FROM stg.bookings AS b
+    WHERE b.batch_id = v_batch_id;
+
+    IF v_dup_count <> 0 THEN
+        RAISE EXCEPTION
+            'DQ FAILED: найдены дубликаты book_ref (batch_id=%): %',
+            v_batch_id,
+            v_dup_count;
+    END IF;
+
+    -- Проверка на NULL или пустые total_amount
+    SELECT COUNT(*)
+    INTO v_null_amount_count
+    FROM stg.bookings AS b
+    WHERE b.batch_id = v_batch_id
+        AND (b.total_amount IS NULL OR b.total_amount = '');
+
+    IF v_null_amount_count <> 0 THEN
+        RAISE EXCEPTION
+            'DQ FAILED: найдены bookings с NULL или пустым total_amount (batch_id=%): %',
+            v_batch_id,
+            v_null_amount_count;
+    END IF;
+
    RAISE NOTICE
        'Проверка количества строк пройдена: источник=%, stg=%',
        v_src_count,
@@ -3,6 +3,13 @@
 -- берём строки, где book_date больше максимального src_created_at_ts
 -- среди "старых" батчей; верхняя граница по дате не используется.

+-- CTE для определения максимальной даты загрузки предыдущего батча
+WITH max_batch_ts AS (
+    SELECT COALESCE(MAX(src_created_at_ts), TIMESTAMP '1900-01-01 00:00:00') AS max_ts
+    FROM stg.bookings
+    WHERE batch_id <> '{{ run_id }}'::text
+        OR batch_id IS NULL
+)
 INSERT INTO stg.bookings (
    book_ref,
    book_date,
@@ -19,18 +26,15 @@ SELECT
    now(),
    '{{ run_id }}'::text
 FROM stg.bookings_ext AS ext
-WHERE ext.book_date > COALESCE(
-    (
-        SELECT max(src_created_at_ts)
-        FROM stg.bookings
-        WHERE batch_id <> '{{ run_id }}'::text
-            OR batch_id IS NULL
-    ),
-    TIMESTAMP '1900-01-01 00:00:00'
-)
+CROSS JOIN max_batch_ts AS mb
+WHERE ext.book_date > mb.max_ts
 AND NOT EXISTS (
    SELECT 1
    FROM stg.bookings AS b
    WHERE b.batch_id = '{{ run_id }}'::text
        AND b.book_ref = ext.book_ref::text
 );
+
+-- Обновляем статистику для оптимизатора Greenplum
+-- Это критично для корректной работы оптимизатора и выбора оптимального плана выполнения
+ANALYZE stg.bookings;
@@ -29,6 +29,11 @@ CREATE TABLE IF NOT EXISTS stg.tickets (
    batch_id           TEXT
 )
 WITH (appendonly=true, orientation=row, compresstype=zlib, compresslevel=1)
-- Распределяем по book_ref, чтобы джойны tickets → bookings по book_ref были без motion.
+-- Ключ распределения: book_ref
+-- Обоснование: book_ref — это основной бизнес-ключ для бронирований.
+-- Использование book_ref обеспечивает:
+-- 1. Co-location данных tickets и bookings при JOIN по book_ref
+-- 2. Равномерное распределение данных по сегментам (book_ref имеет высокую кардинальность)
+-- 3. Оптимизацию запросов, которые фильтруют или группируют по book_ref
 DISTRIBUTED BY (book_ref);

@@ -8,6 +8,8 @@ DECLARE
    v_stg_count     BIGINT;
    v_orphan_count  BIGINT;
    v_null_count    BIGINT;
+    v_dup_count     BIGINT;
+    v_empty_name_count BIGINT;
 BEGIN
    -- Опорная метка: максимум src_created_at_ts среди предыдущих батчей
    SELECT max(src_created_at_ts)
@@ -44,15 +46,13 @@ BEGIN
    END IF;

    -- Проверка ссылочной целостности: все tickets должны иметь соответствующие bookings в этом же STG
+    -- Используем LEFT JOIN вместо NOT EXISTS для лучшей производительности на больших объёмах
    SELECT COUNT(*)
    INTO v_orphan_count
    FROM stg.tickets AS t
+    LEFT JOIN stg.bookings AS b ON t.book_ref = b.book_ref
    WHERE t.batch_id = v_batch_id
-        AND NOT EXISTS (
-            SELECT 1
-            FROM stg.bookings AS b
-            WHERE b.book_ref = t.book_ref
-        );
+        AND b.book_ref IS NULL;

    IF v_orphan_count <> 0 THEN
        RAISE EXCEPTION
@@ -75,6 +75,33 @@ BEGIN
            v_null_count;
    END IF;

+    -- Проверка на дубликаты ticket_no
+    SELECT COUNT(*) - COUNT(DISTINCT ticket_no)
+    INTO v_dup_count
+    FROM stg.tickets AS t
+    WHERE t.batch_id = v_batch_id;
+
+    IF v_dup_count <> 0 THEN
+        RAISE EXCEPTION
+            'DQ FAILED: найдены дубликаты ticket_no (batch_id=%): %',
+            v_batch_id,
+            v_dup_count;
+    END IF;
+
+    -- Проверка на пустые passenger_name
+    SELECT COUNT(*)
+    INTO v_empty_name_count
+    FROM stg.tickets AS t
+    WHERE t.batch_id = v_batch_id
+        AND (t.passenger_name IS NULL OR t.passenger_name = '');
+
+    IF v_empty_name_count <> 0 THEN
+        RAISE EXCEPTION
+            'DQ FAILED: найдены tickets с пустым именем пассажира (batch_id=%): %',
+            v_batch_id,
+            v_empty_name_count;
+    END IF;
+
    RAISE NOTICE
        'DQ PASSED: tickets ок (batch_id=%): source=% stg=%',
        v_batch_id,
@@ -1,6 +1,13 @@
 -- Загрузка инкремента из stg.tickets_ext в stg.tickets
 -- Инкремент определяется по дате бронирования (book_date из bookings.bookings)

+-- CTE для определения максимальной даты загрузки предыдущего батча
+WITH max_batch_ts AS (
+    SELECT COALESCE(MAX(src_created_at_ts), TIMESTAMP '1900-01-01 00:00:00') AS max_ts
+    FROM stg.tickets
+    WHERE batch_id <> '{{ run_id }}'::text
+        OR batch_id IS NULL
+)
 INSERT INTO stg.tickets (
    ticket_no,
    book_ref,
@@ -22,18 +29,15 @@ SELECT
    '{{ run_id }}'::text
 FROM stg.tickets_ext AS ext
 JOIN stg.bookings_ext AS b ON ext.book_ref = b.book_ref
-WHERE b.book_date > COALESCE(
-    (
-        SELECT max(src_created_at_ts)
-        FROM stg.tickets
-        WHERE batch_id <> '{{ run_id }}'::text
-            OR batch_id IS NULL
-    ),
-    TIMESTAMP '1900-01-01 00:00:00'
-)
+CROSS JOIN max_batch_ts AS mb
+WHERE b.book_date > mb.max_ts
 AND NOT EXISTS (
    -- Защита от дублей: ticket_no в источнике уникален, и в stg его не дублируем.
    SELECT 1
    FROM stg.tickets AS t
    WHERE t.ticket_no = ext.ticket_no
 );
+
+-- Обновляем статистику для оптимизатора Greenplum
+-- Это критично для корректной работы оптимизатора и выбора оптимального плана выполнения
+ANALYZE stg.tickets;