Фикс DAG загрузки данных - построение дельты

README.md

@@ -342,7 +342,7 @@ load_bookings_to_stg = PostgresOperator(
     task_id="load_bookings_to_stg",
     postgres_conn_id="greenplum_conn",
     sql="stg/bookings_load.sql",
-    params={"batch_id": "{{ ds_nodash }}"},
+    params={"batch_id": "{{ run_id }}"},
 )
 ```
 

airflow/dags/bookings_to_gp_stage.py

@@ -12,7 +12,7 @@ from __future__ import annotations
 Каждый запуск DAG работает как «шаг по времени вперёд»:
 - генератор в демо-БД bookings добавляет следующий учебный день после max(book_date);
 - загрузка в Greenplum берёт все строки, появившиеся после предыдущих батчей;
-- логическая дата запуска (`ds`) используется как удобная метка запуска (через `ds_nodash` в `batch_id`, в логах и DQ).
+- `run_id` используется как метка запуска (в `batch_id`, в логах и DQ).
 """
 
 from datetime import datetime, timedelta

sql/stg/bookings_dq.sql

@@ -5,7 +5,7 @@
 
 DO $$
 DECLARE
-    v_batch_id    text    := '{{ ds_nodash }}'::text;
+    v_batch_id    text    := '{{ run_id }}'::text;
     v_prev_ts     timestamp;
     v_src_count   bigint;
     v_stg_count   bigint;

sql/stg/bookings_load.sql

@@ -12,19 +12,25 @@ INSERT INTO stg.bookings (
     batch_id
 )
 SELECT
-    book_ref::text,
-    book_date::text,
-    total_amount::text,
-    book_date::timestamp,
+    ext.book_ref::text,
+    ext.book_date::text,
+    ext.total_amount::text,
+    ext.book_date::timestamp,
     now(),
-    '{{ ds_nodash }}'::text
-FROM stg.bookings_ext
-WHERE book_date > COALESCE(
+    '{{ 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 <> '{{ ds_nodash }}'::text
+        WHERE batch_id <> '{{ run_id }}'::text
             OR batch_id IS NULL
     ),
     TIMESTAMP '1900-01-01 00:00:00'
+)
+    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
     );