# Сценарий

Настроить инкрементальную загрузку таблиц products, customers, orders из источников pg-shop-1, pg-shop-2 ('New-York' и 'Istanbul' соответственно) в неисторические таблицы DWH. Не учитывать, что данные могут обновляться с некоторой глубиной в прошлом.

Ввести некоторый конфиг, в котором задаётся структура таблицы и параметры её загрузки. Код учитывает возможные изменения этого конфига, добавление новых таблиц, источников.

Считать, что у всех источников всех магазинов одинаковая структура таблиц для каждой из сущностей.

<details>
<summary>Подсказки</summary>
<br>

* Вначале можно в staging грузить очередной срез данных, появившихся позднее, чем последняя загрузка max_dttm на таргете. Потом -- подготавливать датасеты и вставлять строки с  created_at > максимальное время загрузки вставленных строк.
* Обеспечивай идемпотентность, где возможно. Очищай промежуточные таблицы перед вставкой. При этом следи, что другой даг прямо сейчас с ними не работает.
* Фильтруй данные, как можно сильнее снижая количество строк в датасете для очередной проверки и загрузки.
* Добавь технические поля для отслеживания источника, времени создания и/или обновления данных. 
* Учти, что ключ у таблиц в хранилище расширится - станет {table}_id и src_id.
</details>
<br>

Definition of done:
* 3 таблицы из двух источников грузятся в DWH.
* Решение должно легко расширяться под новые таблицы и источники.
* В таблицах нет дублей и пропусков, ID монотонно увеличиваются.
* Используй для самопроверки `airflow_data/dags/scripts/self-test.sql`.