Изменение данных и идемпотентность загрузки

Загрузка данных это INSERT, UPDATE и DELETE, только запускаемые по расписанию тысячи раз. Главное свойство хорошей загрузки называется идемпотентностью: повторный запуск за тот же период даёт тот же результат, что и первый. Без неё каждый перезапуск после сбоя создаёт дубли, а каждая попытка «догрузить вчера» превращается в ручную работу.

Наивная загрузка и почему она ломается

Витрина fact_orders обновляется из stg_orders: каждый день приезжают новые версии заказов. Самая простая загрузка:

INSERT INTO fact_orders
SELECT order_id, customer_id, status, total_amount, updated_at
FROM stg_orders
WHERE load_date = '2026-06-30';

Первый запуск работает. Второй запуск за тот же день, после падения Airflow на следующем шаге, вставляет те же 53 строки ещё раз. Если у витрины есть первичный ключ, второй запуск упадёт с duplicate key value violates unique constraint, и это хороший исход: вы узнали о проблеме. Если ключа нет, витрина молча получила дубли, и завтра выручка удвоится.

Второй дефект в этом запросе: заказ, изменённый дважды за день, приедет двумя версиями, и обе попадут в витрину. Даже единственный запуск нарушает зерно «одна строка на заказ».

Upsert: вставить или обновить

PostgreSQL решает обе проблемы одной конструкцией:

INSERT INTO fact_orders (order_id, customer_id, status, total_amount, updated_at)
SELECT DISTINCT ON (order_id) order_id, customer_id, status, total_amount, updated_at
FROM stg_orders
WHERE load_date = '2026-06-30'
ORDER BY order_id, updated_at DESC, loaded_at DESC
ON CONFLICT (order_id) DO UPDATE
SET status       = EXCLUDED.status,
    total_amount = EXCLUDED.total_amount,
    updated_at   = EXCLUDED.updated_at
WHERE EXCLUDED.updated_at > fact_orders.updated_at;

Три детали, каждая важна. DISTINCT ON оставляет одну версию на заказ из самой загрузки, иначе ON CONFLICT упадёт, столкнувшись с двумя строками одного ключа в одной команде. EXCLUDED это строка, которую пытались вставить. Условие WHERE EXCLUDED.updated_at > fact_orders.updated_at защищает от повторной загрузки старого файла поверх свежих данных: обновление происходит только если версия новее.

ON CONFLICT требует уникального индекса или первичного ключа по колонкам конфликта. Это ещё один аргумент за первичный ключ на витрине.

MERGE: стандартный способ

С PostgreSQL 15 доступен MERGE, он же есть в Greenplum 7, Spark, BigQuery и Snowflake:

MERGE INTO fact_orders AS f
USING (
  SELECT DISTINCT ON (order_id) order_id, customer_id, status, total_amount, updated_at
  FROM stg_orders
  WHERE load_date >= '2026-06-01'
  ORDER BY order_id, updated_at DESC, loaded_at DESC
) AS s ON f.order_id = s.order_id
WHEN MATCHED AND s.updated_at > f.updated_at THEN
  UPDATE SET status = s.status, total_amount = s.total_amount, updated_at = s.updated_at
WHEN NOT MATCHED THEN
  INSERT VALUES (s.order_id, s.customer_id, s.status, s.total_amount, s.updated_at);

Читается как правило: нашли совпадение и версия новее, обновляем; не нашли, вставляем. Можно добавить ветку WHEN MATCHED AND s.is_deleted THEN DELETE. В нашей базе такой MERGE за июнь трогает 674 строки, и после него витрина совпадает с orders по всем 8000 заказам. Повторный запуск не изменит ничего: все версии уже не новее. Это и есть идемпотентность.

Перезапись партиции: когда версий нет

Для событий и платежей, которые не меняются после появления, upsert избыточен. Проще перезаписать период целиком:

BEGIN;
DELETE FROM fact_payment
WHERE paid_at >= '2026-06-30' AND paid_at < '2026-07-01';
 
INSERT INTO fact_payment
SELECT payment_id, order_id, paid_at, amount, method
FROM payments
WHERE status = 'success'
  AND paid_at >= '2026-06-30' AND paid_at < '2026-07-01';
COMMIT;

Сколько бы раз ни запустить, в витрине ровно один набор строк за 30 июня. Обе команды в одной транзакции: читатели не увидят пустой день между удалением и вставкой. В Greenplum и ClickHouse то же делают через партиции: удалить партицию дня и загрузить заново, это дешевле построчного DELETE. В Spark это режим overwrite с динамическими партициями.

Границы периода, как и везде, полуинтервал. BETWEEN с датой здесь потерял бы последний день и оставил дубли на стыке.

Инкремент и опоздавшие данные

Полная перезагрузка невозможна, когда таблица большая. Тогда грузят инкремент: строки, изменённые после отметки, которую запомнили в прошлый раз. Отметку называют watermark.

INSERT INTO fact_event
SELECT event_id, user_id, device_id, event_type, event_ts
FROM events
WHERE received_at >= :last_watermark AND received_at < :new_watermark;

Ключевой вопрос: по какой колонке считать окно. Напрашивается event_ts, момент события. Но события опаздывают: в нашей базе 660 событий получены на следующий день или позже, максимальная задержка три дня. Событие, произошедшее 28 июня в 23:50 и полученное 29-го, при загрузке «за 28 июня» по event_ts уже не попадёт: окно 28-го закрыто. Из 164 событий 28 июня к утру 29-го сервер видел только 161.

Окно инкремента считается по времени поступления, received_at, а event_ts остаётся полем данных. Тогда опоздавшее событие приедет в следующем окне и ляжет в партицию своего дня. Из этого следует, что дневная партиция события за 28 июня остаётся «живой» ещё несколько дней, и отчёт за вчера может немного измениться. Об этом стоит договориться с потребителями данных, иначе они будут считать это ошибкой.

Если опоздавшее событие может прийти повторно, инкремент дополняют дедупликацией: upsert по event_id или DELETE из витрины тех ключей, которые есть в порции, перед INSERT.

UPDATE и DELETE с соединением

Массовые исправления пишутся через FROM:

UPDATE orders o
SET delivery_city = c.city
FROM customers c
WHERE c.customer_id = o.customer_id
  AND o.delivery_city IS NULL;   -- 625 строк

DELETE ... USING работает так же. Полезное дополнение RETURNING: команда возвращает затронутые строки, и их можно записать в журнал изменений или сразу проверить. DELETE FROM payments WHERE status = 'failed' RETURNING payment_id вернёт 619 идентификаторов.

Перед массовым UPDATE или DELETE на проде запускайте тот же WHERE в SELECT count(*). Пять секунд проверки против восстановления из резервной копии.

Практика

Окружение задач содержит витрину fact_orders, заполненную заказами до 1 июня 2026 года, с первичным ключом по order_id. Проверяется состояние витрины после ваших команд.