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

PostgreSQL 15 (tháng 10/2022) cuối cùng đã thêm MERGE — câu lệnh SQL chuẩn từ SQL:2003 mà PostgreSQL chậm hơn Oracle, SQL Server, và DB2 cả thập kỷ. Với ON CONFLICT (UPSERT) bạn đã biết từ bài 5 của module này, bạn có thể handle phần lớn trường hợp ghi idempotent. 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 (PG 16+): target row tồn tại nhưng không có source row nào match.

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    timestamptz 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. PG 17 — MERGE RETURNING

PostgreSQL 17 (tháng 9/2024) bổ sung RETURNING cho MERGE, kèm hàm đặc biệt merge_action() trả về action nào đã thực sự xảy ra cho mỗi row.

-- PG 17+: RETURNING + merge_action()
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 đủ).

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    timestamptz 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 (PG 15+) xử lý INSERT, UPDATE, DELETE trong một statement — phù hợp cho ETL multi-rule sync mà UPSERT không cover.
• UPSERT (ON CONFLICT, PG 9.5+) đơ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 với DISTINCT ON.
• Concurrent MERGE trên cùng target row có thể gây serialization failure (SQLSTATE 40001) — implement retry ở application layer.
• PG 17 thêm RETURNING và merge_action() — cho phép track từng row được INSERT/UPDATE/DELETE sau khi MERGE, hữu ích cho audit log và downstream processing.
• 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 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 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)