MERGE и UPSERT на собеседовании Data Engineer

Зачем спрашивают на собесе DE

UPSERT — основной инструмент идемпотентного ETL. Если пайплайн перезапустился — повторная вставка не должна сломать таблицу. На собесе DE обязательно: «как сделать idempotent insert», «зачем MERGE», «отличие ON CONFLICT от MERGE». Senior — нюансы concurrent write, deadlock-free MERGE.

Главная боль без понимания — DE сделал INSERT в DWH. Скрипт упал на середине, перезапустили — получили дубликаты. Аналитика показывает x2 заказов. Срочный фикс через DELETE-then-INSERT, потеря 2 часов.

Постановка задачи

Дано: таблица users(user_id PK, email, updated_at). Источник: батч пользователей из CDC.

Хочется:

• Если user_id уже есть → обновить.
• Если нет → вставить.
• Перезапуск батча — без дубликатов и без ломки.

Наивный путь — SELECT exists, IF then INSERT/UPDATE — race condition на concurrent batch. Правильный путь — атомарный UPSERT.

Postgres: INSERT ... ON CONFLICT

Postgres-расширение, очень удобное.

INSERT INTO users (user_id, email, updated_at)
VALUES (1, 'a@b.com', NOW())
ON CONFLICT (user_id) DO UPDATE
SET email = EXCLUDED.email,
    updated_at = EXCLUDED.updated_at;

EXCLUDED — псевдо-таблица с новыми значениями. ON CONFLICT (user_id) ссылается на UNIQUE-индекс / PRIMARY KEY.

Варианты:

-- Игнорировать дубликат
INSERT ... ON CONFLICT (user_id) DO NOTHING;

-- Условный update (обновить только если новее)
INSERT INTO users (user_id, email, updated_at) VALUES (...)
ON CONFLICT (user_id) DO UPDATE
SET email = EXCLUDED.email
WHERE users.updated_at < EXCLUDED.updated_at;

-- ON CONFLICT по нескольким колонкам / partial index
ON CONFLICT (user_id) WHERE deleted_at IS NULL DO UPDATE ...

Bulk вставка через VALUES:

INSERT INTO users VALUES
  (1, 'a@b.com', NOW()),
  (2, 'c@d.com', NOW())
ON CONFLICT (user_id) DO UPDATE SET email = EXCLUDED.email;

В Python через psycopg2.extras.execute_values или copy_expert для миллионов строк.

Стандартный MERGE (SQL:2003)

MERGE INTO users AS target
USING (SELECT * FROM staging_users) AS src
   ON target.user_id = src.user_id
WHEN MATCHED AND src.updated_at > target.updated_at THEN
  UPDATE SET email = src.email, updated_at = src.updated_at
WHEN NOT MATCHED THEN
  INSERT (user_id, email, updated_at)
  VALUES (src.user_id, src.email, src.updated_at)
WHEN MATCHED AND src.deleted = TRUE THEN
  DELETE;

Поддержка:

• Postgres ≥ 15 — да, ANSI MERGE.
• Oracle, SQL Server, DB2 — давно.
• ClickHouse — нет (есть ReplacingMergeTree как замена).
• Snowflake, BigQuery, Redshift — да.

MERGE мощнее ON CONFLICT — поддерживает DELETE и сложные WHEN. Но писать сложнее.

Greenplum, Snowflake, BigQuery

Greenplum — расширение Postgres. До GP 7 не было полной поддержки INSERT ON CONFLICT — приходилось делать DELETE + INSERT в одной транзакции. С 7+ — стандартный синтаксис.

Snowflake:

MERGE INTO users t
USING staging_users s ON t.user_id = s.user_id
WHEN MATCHED THEN UPDATE SET email = s.email
WHEN NOT MATCHED THEN INSERT VALUES (s.user_id, s.email);

BigQuery — почти такой же MERGE. Особенность: на больших таблицах MERGE дороже, чем INSERT + DELETE через partition. Часто пользуются INSERT ... SELECT ... NOT EXISTS или партиционным переписыванием.

ClickHouse: ReplacingMergeTree

В CH нет MERGE / UPDATE как в OLTP. Идиома — таблица с движком ReplacingMergeTree, где «дубликаты» (по PK) автоматически удаляются при merge.

CREATE TABLE users (
  user_id UInt64,
  email String,
  updated_at DateTime
) ENGINE = ReplacingMergeTree(updated_at)
ORDER BY user_id;

updated_at в скобках — version column. При merge'е остаётся строка с максимальным updated_at.

Особенности:

• Дедупликация eventual — происходит при background merge, не сразу. SELECT может видеть оба значения.
• OPTIMIZE TABLE ... FINAL форсит дедупликацию (дорого).
• SELECT ... FINAL принудительно объединяет на чтение (медленно, но честно).

Альтернатива — CollapsingMergeTree (для удалений) или VersionedCollapsingMergeTree.

MERGE в инкрементальном dbt

dbt-модель materialized='incremental' использует MERGE/UPSERT под капотом.

{{ config(
  materialized='incremental',
  unique_key='order_id',
  incremental_strategy='merge'
) }}

SELECT * FROM {{ source('raw', 'orders') }}
{% if is_incremental() %}
WHERE updated_at > (SELECT MAX(updated_at) FROM {{ this }})
{% endif %}

Стратегии:

• merge — стандартный MERGE.
• delete+insert — для адаптеров без MERGE.
• append — без дедупликации, чисто insert.
• insert_overwrite — переписать партицию целиком (хорошо для timeseries).

unique_key обязателен для merge. Без него — дубликаты.

Частые ошибки

Бить INSERT-ом без UPSERT в идемпотентном пайплайне. При ретрае — дубли. Всегда ON CONFLICT или MERGE для целевой таблицы с PK.

Использовать ON CONFLICT (col1, col2) без UNIQUE индекса. Postgres требует индекс. Иначе — ошибка there is no unique or exclusion constraint.

Забыть WHERE в условном update. ON CONFLICT DO UPDATE без WHERE обновит даже более старую версию. Если CDC несёт старые сообщения позже — данные будут затёрты.

Concurrent UPSERT в одну строку. Postgres сериализует на UNIQUE индексе, но при больших объёмах возможны deadlock'и. Сортировка батча по unique_key снижает вероятность.

MERGE в Postgres ≤ 14. Не поддерживается — только с 15. Раньше — INSERT ... ON CONFLICT.

Делать UPSERT построчно из приложения. 10 000 INSERT ... ON CONFLICT по одному = 10 000 round-trip. Bulk через execute_values или COPY + INSERT FROM SELECT — в 100× быстрее.

Не учитывать ClickHouse eventual dedup. SELECT после INSERT может видеть дубли. Всегда учитывай: либо FINAL, либо группируй на чтении (argMax).

MERGE без partition pruning. На партиционированной таблице MERGE может прочитать всё. Убедись, что ON-клоз содержит partition key.

Связанные темы

• Идемпотентность пайплайна для DE
• Партиционирование таблиц для DE
• SCD типы для DE
• dbt на собесе DE
• Подготовка к собесу Data Engineer

FAQ

MERGE атомарен?

Да, в рамках одной транзакции. Concurrent MERGE на пересекающиеся ключи может ловить deadlock — обрабатывается через retry.

Что лучше для bulk: ON CONFLICT или MERGE?

В Postgres — ON CONFLICT обычно проще и не медленнее. В Snowflake / BigQuery — MERGE стандарт, других вариантов нет.

Можно ли сделать UPSERT на partial unique index?

Да: ON CONFLICT (user_id) WHERE deleted_at IS NULL DO UPDATE .... Index должен совпадать с условием.

MERGE для миллиардов строк — это проблема?

Может быть. На больших фактовых таблицах часто эффективнее INSERT OVERWRITE PARTITION или копия таблицы. MERGE — для DWH-моделирования с разумными размерами.

Как обработать DELETE через CDC в UPSERT?

Источник несёт is_deleted=true. В MERGE — WHEN MATCHED AND src.is_deleted THEN DELETE. В ON CONFLICT — придётся отдельным DELETE-запросом или soft-delete через флаг.

Это официальная информация?

Нет. Статья основана на документации Postgres 15+, стандарте SQL:2003 и материалах по DWH.

Тренируйте Data Engineering — откройте тренажёр с 1500+ вопросами для собесов.