MERGE Statement — ETL multi-rule sync trong một câu lệnh

TL;DR: MERGE (chuẩn SQL:2003) là cách chuẩn hoá cho insert-hoặc-update, và mạnh hơn UPSERT đơn: nó so từng cặp source–target rồi áp dụng nhiều WHEN MATCHED / WHEN NOT MATCHED clause — gồm cả DELETE — trong một lần quét. Quy tắc then chốt: WHEN clause evaluate từ trên xuống, first match wins per row, nên clause đặc thù (DELETE, UPDATE có điều kiện) phải đứng trước catch-all. UPSERT (bài 05) đủ cho ~90% ingestion idempotent; chỉ chuyển sang MERGE khi cần DELETE hoặc nhiều rule (ETL multi-rule sync, slowly-changing dimension). MERGE là cú pháp ANSI có ở Oracle, SQL Server, Db2, và các engine mới — ghi chú dialect khi tính năng khác biệt (vd RETURNING/merge_action()).

MERGE là câu lệnh SQL chuẩn từ SQL:2003 cho bài toán "với mỗi cặp source–target, quyết định INSERT / UPDATE / DELETE / bỏ qua". Với UPSERT bạn đã biết từ bài 05 của module này, bạn handle được phần lớn trường hợp ghi idempotent theo một khoá. Nhưng khi ETL cần đồng thời INSERT, UPDATE, và DELETE trong một lần quét source table — đó là lúc MERGE thực sự cần thiết.

Bài này giải thích tại sao MERGE tồn tại, khi nào nó thắng UPSERT, cú pháp đầy đủ, demo thực tế TaskFlow bulk sync từ external PM tool, và những pitfall ẩn mà tài liệu ít nhắc.

1. Analogy — Đồng bộ danh sách contact

Hãy hình dung bạn có danh sách contact trên điện thoại (target) và danh sách contact trên server (source). Mỗi lần sync, có 3 trường hợp xảy ra cho mỗi contact:

MERGE là "nhìn vào 2 danh sách, so sánh từng entry, rồi quyết định INSERT / UPDATE / DELETE / DO NOTHING — tất cả trong một lần đi qua". Thay vì viết 3-4 câu lệnh riêng lẻ, bạn mô tả toàn bộ logic đồng bộ trong một statement.

2. UPSERT vs MERGE — khi nào MERGE thắng

Trước khi dùng MERGE, luôn tự hỏi: UPSERT có đủ không? UPSERT đơn giản hơn và cover 90% case.

3. Cú pháp MERGE — anatomy

MERGE INTO target_table t
USING source_table s          -- hoac subquery, CTE
ON t.key = s.key              -- join condition de xac dinh "match"

WHEN MATCHED AND s.deleted = true THEN
  DELETE

WHEN MATCHED AND s.updated_at > t.updated_at THEN
  UPDATE SET col1 = s.col1,
             col2 = s.col2,
             updated_at = s.updated_at

WHEN MATCHED THEN
  -- catch-all: da match nhung khong thoa dieu kien tren -> skip
  DO NOTHING

WHEN NOT MATCHED AND s.deleted = false THEN
  INSERT (col1, col2, col3)
  VALUES (s.col1, s.col2, s.col3);
-- Khong co semicolon sau WHEN cuoi; dat ; sau toan bo MERGE statement

Quy tắc quan trọng — Order matters:

• WHEN clauses được evaluate theo thứ tự từ trên xuống.
• First match wins per row — một source row chỉ trigger tối đa 1 WHEN clause.
• WHEN MATCHED = ON condition cho kết quả true (target row tồn tại).
• WHEN NOT MATCHED = ON condition cho kết quả false (target row chưa tồn tại).
• WHEN NOT MATCHED BY SOURCE (mở rộng chuẩn, một số engine hỗ trợ — vd SQL Server, PostgreSQL gần đây): target row tồn tại nhưng không có source row nào match — dùng để xoá/đánh dấu row không còn trong source.

flowchart TD
  ROW["Voi moi cap source-target"] --> ON{"Match dieu kien ON?<br/>(target row ton tai?)"}
  ON -->|"Co (MATCHED)"| W1{"WHEN MATCHED clause 1<br/>vd: s.deleted = true"}
  ON -->|"Khong (NOT MATCHED)"| W4{"WHEN NOT MATCHED clause<br/>vd: s.deleted = false"}
  W1 -->|"thoa"| DEL["DELETE -- dung, bo qua clause sau"]
  W1 -->|"khong thoa"| W2{"WHEN MATCHED clause 2<br/>vd: s.updated_at > t.updated_at"}
  W2 -->|"thoa"| UPD["UPDATE -- dung"]
  W2 -->|"khong thoa"| W3["WHEN MATCHED THEN DO NOTHING<br/>(catch-all)"]
  W4 -->|"thoa"| INS["INSERT"]
  W4 -->|"khong thoa"| SKIP["Bo qua row"]

Sơ đồ minh hoạ first match wins: mỗi row đi từ trên xuống, dừng ở clause đầu tiên thoả — nên clause đặc thù (DELETE, UPDATE có điều kiện) phải đứng trước catch-all.

4. Demo — TaskFlow bulk sync external PM tool

Bài toán: mỗi giờ, hệ thống pull task từ external project management tool (Jira, Linear, Asana) về staging table, rồi sync vào bảng tasks của TaskFlow. Source có thể chứa task mới, task cập nhật, và task bị xóa.

-- DDL tham khao TaskFlow (canonical)
-- tasks: id, external_id, project_id, title, status, assignee_id,
--        created_at, updated_at

-- Stage table tu external API response
CREATE TEMP TABLE staging_tasks (
  external_id   text        NOT NULL,
  title         text        NOT NULL,
  status        text        NOT NULL,
  assignee_id   int,
  is_deleted    boolean     NOT NULL DEFAULT false,
  updated_at    TIMESTAMP NOT NULL
);

-- Gia su da INSERT data tu API vao staging_tasks...

-- Sync trong 1 MERGE statement: 3 case, 1 lan quet
MERGE INTO tasks t
USING staging_tasks s
  ON t.external_id = s.external_id

WHEN MATCHED AND s.is_deleted = true THEN
  -- Case 1: task bi xoa o source -> hard delete (hoac soft delete neu co deleted_at)
  DELETE

WHEN MATCHED AND s.updated_at > t.updated_at THEN
  -- Case 2: task ton tai + source moi hon -> update
  UPDATE SET
    title       = s.title,
    status      = s.status,
    assignee_id = s.assignee_id,
    updated_at  = s.updated_at

WHEN MATCHED THEN
  -- Case 3: task ton tai + target da moi hon hoac bang -> skip
  DO NOTHING

WHEN NOT MATCHED AND s.is_deleted = false THEN
  -- Case 4: task chua co trong target + chua bi xoa -> insert
  INSERT (external_id, project_id, title, status, assignee_id, created_at, updated_at)
  VALUES (s.external_id, 1, s.title, s.status, s.assignee_id, now(), s.updated_at);

5. MERGE RETURNING — biết action nào đã chạy (dialect)

Một số engine cho lấy lại kết quả của MERGE ngay trong câu lệnh, kèm một hàm trả về action đã thực sự xảy ra cho mỗi row (PostgreSQL gọi là merge_action(), trả 'INSERT' / 'UPDATE' / 'DELETE'). Tính năng này không thuộc chuẩn SQL cốt lõi và mức hỗ trợ khác nhau — kiểm tra engine của bạn trước khi dùng. Cú pháp dưới minh hoạ kiểu PostgreSQL:

-- RETURNING + merge_action() (dialect: PostgreSQL)
MERGE INTO tasks t
USING staging_tasks s
  ON t.external_id = s.external_id

WHEN MATCHED AND s.is_deleted = true THEN
  DELETE

WHEN MATCHED AND s.updated_at > t.updated_at THEN
  UPDATE SET title = s.title, status = s.status, updated_at = s.updated_at

WHEN NOT MATCHED AND s.is_deleted = false THEN
  INSERT (external_id, project_id, title, status, assignee_id, created_at, updated_at)
  VALUES (s.external_id, 1, s.title, s.status, s.assignee_id, now(), s.updated_at)

RETURNING merge_action(), t.id, t.title, t.external_id;

-- merge_action() tra ve: 'INSERT' | 'UPDATE' | 'DELETE'
-- Row bi DO NOTHING khong xuat hien trong RETURNING

merge_action() trả về 'INSERT' / 'UPDATE' / 'DELETE' — row bị DO NOTHING không xuất hiện trong RETURNING. Dùng kết hợp với CTE để log audit trail hoặc downstream processing trong cùng một transaction (xem SelfCheck câu 4 cho pattern đầy đủ). Engine không có RETURNING/merge_action() thì lấy audit trail bằng trigger hoặc một câu SELECT riêng sau MERGE.

6. Pitfall — duplicate source, concurrent serialization, và khi nào không dùng MERGE

-- Neu staging_tasks co 2 row cung external_id:
-- external_id='EXT-001' xuat hien 2 lan trong staging_tasks
-- MERGE se match cung 1 target row 2 lan -> loi runtime:
-- ERROR: MERGE command cannot affect row a second time
-- (PG raise exception khi detect duplicate target match)

-- Fix: dedupe source truoc MERGE bang DISTINCT ON
MERGE INTO tasks t
USING (
  SELECT DISTINCT ON (external_id)
    *
  FROM staging_tasks
  ORDER BY external_id, updated_at DESC  -- giu ban moi nhat
) s
  ON t.external_id = s.external_id
-- ... WHEN clauses ...

ERROR: could not serialize access due to concurrent update
SQLSTATE: 40001

7. Applied — ETL Slowly-Changing Dimension Type 1

MERGE là pattern chuẩn cho data warehouse ETL — đặc biệt là Slowly-Changing Dimension (SCD). SCD Type 1 (overwrite — giữ giá trị mới nhất, bỏ history) là use case phổ biến nhất và fit hoàn hảo với MERGE.

-- Stage table tu ETL pipeline (raw extract tu source system)
CREATE TEMP TABLE staging_dim_customer (
  customer_id   int         NOT NULL,
  name          text        NOT NULL,
  email         text        NOT NULL,
  country       text        NOT NULL,
  valid_from    TIMESTAMP NOT NULL
);

-- SCD Type 1: overwrite old value — dim_customer luon reflect latest version
MERGE INTO dim_customer t
USING staging_dim_customer s
  ON t.customer_id = s.customer_id

WHEN MATCHED AND
     (t.name, t.email, t.country) IS DISTINCT FROM
     (s.name, s.email, s.country) THEN
  -- Chi update khi thuc su co thay doi (tranh write amplification)
  UPDATE SET
    name       = s.name,
    email      = s.email,
    country    = s.country,
    updated_at = now()

WHEN NOT MATCHED THEN
  -- Customer moi: insert
  INSERT (customer_id, name, email, country, created_at, updated_at)
  VALUES (s.customer_id, s.name, s.email, s.country, now(), now());

SCD Type 2 (giữ history qua valid_from / valid_to) phức tạp hơn — cần 2 MERGE chained hoặc PL/pgSQL function.

8. Deep Dive

9. Tóm tắt

• MERGE (chuẩn SQL:2003) xử lý INSERT, UPDATE, DELETE trong một statement — phù hợp cho ETL multi-rule sync mà UPSERT đơn không cover. Có ở Oracle/SQL Server/Db2 lâu, engine khác thêm gần đây.
• UPSERT đơn (ON CONFLICT ở PostgreSQL/SQLite, ON DUPLICATE KEY UPDATE ở MySQL) đơn giản hơn và đủ cho 90% case — chỉ chuyển sang MERGE khi thực sự cần DELETE hoặc nhiều hơn 1 condition rule.
• WHEN clause được evaluate theo thứ tự từ trên xuống, first match wins per row — thứ tự viết quan trọng: DELETE check phải đứng trước UPDATE catch-all.
• Source duplicate (hai source row match cùng target row) gây runtime error MERGE command cannot affect row a second time — luôn dedupe source trước MERGE (vd DISTINCT ON ở PostgreSQL, hoặc subquery với ROW_NUMBER() ở engine khác).
• Concurrent MERGE trên cùng target row có thể gây serialization failure (SQLSTATE 40001 — mã chuẩn SQL) — implement retry ở application layer.
• RETURNING + merge_action() (extension dialect, không thuộc chuẩn cốt lõi) cho phép track từng row được INSERT/UPDATE/DELETE sau khi MERGE — hữu ích cho audit log; engine không có thì dùng trigger hoặc SELECT riêng.
• SCD Type 1 (overwrite) là pattern chuẩn của MERGE trong data warehouse ETL: IS DISTINCT FROM để so sánh composite value an toàn với NULL.

10. Tự kiểm tra

MERGE INTO tasks t
USING (
SELECT * FROM (
  SELECT s.*,
         ROW_NUMBER() OVER (PARTITION BY external_id
                            ORDER BY updated_at DESC) AS rn
  FROM staging_tasks s
) d WHERE rn = 1            -- giu ban moi nhat per external_id
) s
ON t.external_id = s.external_id
WHEN MATCHED THEN UPDATE SET ...
WHEN NOT MATCHED THEN INSERT ...;

WHEN MATCHED THEN
UPDATE SET title = s.title  -- luon match truoc, DELETE ben duoi khong bao gio chay!
WHEN MATCHED AND s.is_deleted = true THEN
DELETE                      -- dead code: MATCHED da bi bat o tren

WHEN MATCHED AND s.is_deleted = true THEN
DELETE                      -- check specific condition truoc
WHEN MATCHED THEN
UPDATE SET title = s.title  -- catch-all chi chay neu DELETE khong match

-- <> voi NULL: bat ky phep so sanh voi NULL tra ve UNKNOWN (false trong WHERE)
-- (t.name, t.email) <> (s.name, s.email) -> UNKNOWN neu bat ky gia tri nao la NULL
-- => row bi bo qua du thuc ra co thay doi

-- IS DISTINCT FROM: xu ly NULL dung
-- (t.name, t.email) IS DISTINCT FROM (s.name, s.email)
-- -> TRUE neu bat ky column nao khac nhau, KE CA khi mot ben la NULL
-- -> FALSE chi khi tat ca column deu bang nhau (NULL = NULL la FALSE trong IS DISTINCT FROM)

WITH sync_result AS (
MERGE INTO tasks t
USING staging_tasks s ON t.external_id = s.external_id
WHEN MATCHED AND s.is_deleted = true THEN DELETE
WHEN MATCHED AND s.updated_at > t.updated_at THEN
  UPDATE SET title = s.title, status = s.status,
             updated_at = s.updated_at
WHEN NOT MATCHED AND s.is_deleted = false THEN
  INSERT (external_id, project_id, title, status, created_at, updated_at)
  VALUES (s.external_id, 1, s.title, s.status, now(), s.updated_at)
RETURNING merge_action() AS action, t.id, t.external_id
)
INSERT INTO sync_history (sync_run_id, action, task_id, external_id, synced_at)
SELECT $1, action, id, external_id, now()
FROM sync_result;

-- Pseudo-code: retry voi exponential backoff
for attempt in 1..5:
try:
  BEGIN;
  MERGE INTO tasks ... ;
  INSERT INTO sync_history ...;
  COMMIT;
  break  -- thanh cong
except SQLSTATE 40001:
  ROLLBACK;
  if attempt == 5: raise  -- het retry
  sleep(0.1 * 2^attempt)  -- exponential backoff: 0.2s, 0.4s, 0.8s...

Bài tiếp theo: Mini-challenge — sprint burndown chart (CTE + window + UPSERT)