Implementing Pipeline-Level Idempotent Upserts with MERGE Statements Across Snowflake, BigQuery, and Redshift: Patterns, Pitfalls, and Performance Tuning

Introduction

Your nightly pipeline has been running cleanly for six weeks. Then, at 2:47 AM, your orchestrator drops a network connection mid-run, retries the job from the beginning, and by the time anyone checks the dashboard, your orders table has duplicated $4.2 million worth of transactions. The data exists twice — same order IDs, different inserted-at timestamps — and your downstream revenue report is now wrong in a way that's embarrassingly visible to the CFO.

This is the scenario that makes idempotent upserts worth understanding at a deep level. An idempotent operation is one you can run multiple times and always get the same result. In the context of data pipelines, it means that re-running a load — whether due to a retry, a backfill, or human error — never produces duplicates, phantom deletes, or corrupted aggregates. The MERGE statement is the SQL primitive that makes this possible at the warehouse level, but "use MERGE" is not the same as "you're idempotent." The details of how you structure that MERGE, what staging strategy you use, and how each cloud warehouse actually executes it determine whether your pipeline is genuinely safe or just optimistically fragile.

By the end of this lesson, you'll be able to implement production-grade idempotent upserts across Snowflake, BigQuery, and Redshift, understand the architectural trade-offs of each platform's MERGE semantics, tune query plans for large-scale merges, and design staging patterns that make your pipelines resilient to the full spectrum of failure modes.

What you'll learn:

• How MERGE semantics differ across Snowflake, BigQuery, and Redshift, and why those differences change your implementation strategy
• How to design staging tables and deduplication logic that guarantee idempotency at the pipeline level, not just the SQL level
• Performance tuning techniques specific to each platform: clustering, partitioning, and warehouse sizing for MERGE workloads
• How to handle schema evolution, late-arriving data, and soft-delete patterns inside MERGE operations
• Common failure modes — including non-obvious ones — and how to test your way to confidence

---

Prerequisites

This article assumes you're comfortable writing SQL including subqueries and window functions, have working knowledge of at least one cloud data warehouse, and understand the basics of pipeline orchestration. If you're newer to how pipelines are structured conceptually, What is a Data Pipeline? Architecture and Core Concepts for Data Engineers will give you the right foundation. You should also understand the difference between batch and incremental loading — if that's fuzzy, read Incremental Loading Patterns: Timestamps, CDC, and Watermarks before continuing.

---

What "Idempotent" Actually Means in a MERGE Context

Before writing a single line of SQL, let's be precise about what we're guaranteeing. A MERGE-based upsert is idempotent if re-running it against the same source data always produces the same target table state. Full stop. That guarantee has three practical components:

1. Insert idempotency: Rows that don't exist get inserted. Running the same insert twice doesn't create duplicates.
2. Update idempotency: Rows that exist get updated to the source value. Running the same update twice doesn't change the result.
3. Delete idempotency (if applicable): Rows marked for deletion get deleted. Running the delete twice doesn't error or cause phantom behavior.

The MERGE statement satisfies all three — in theory. In practice, you can break idempotency at the pipeline level in several ways that have nothing to do with the SQL syntax:

• Non-deterministic staging: If your staging table contains data from multiple pipeline runs mixed together, your MERGE will process different rows on each run.
• Multiple source matches: If your source data has duplicate primary keys and your warehouse resolves ambiguity differently across runs, your updates are non-deterministic.
• Transaction boundary mismatches: If your MERGE succeeds but a downstream step fails, and your retry re-loads additional source rows before re-merging, you've broken idempotency.

The pattern that solves all three is: isolated staging + deterministic deduplication + atomic MERGE. We'll build this up piece by piece.

---

The Canonical Pipeline-Level MERGE Pattern

Regardless of which warehouse you're targeting, the structure looks like this:

1. TRUNCATE or replace the staging table (isolate this run's data)
2. Load source rows into staging (COPY, INSERT, or external stage)
3. Deduplicate staging to one row per primary key (using window functions)
4. Execute MERGE from deduplicated staging into target
5. Log the run metadata (rows inserted, updated, deleted, run_id)
6. On retry: go back to step 1 — the truncate makes it safe

Step 1 is the most important safety valve. If you're appending to a staging table across runs, a retry will include rows from both the failed run and the retry, and your MERGE will behave unpredictably. Truncating or replacing the staging table at the start of each run ensures the data is scoped to exactly one pipeline execution.

Let's ground this in a realistic scenario. You're building a customer data pipeline that ingests rows from a CRM system's change data capture stream. The target table tracks current customer state:

-- Target table: customers
CREATE TABLE customers (
    customer_id       VARCHAR(36)   NOT NULL,
    email             VARCHAR(255),
    full_name         VARCHAR(255),
    plan_tier         VARCHAR(50),
    mrr_usd           NUMERIC(12,2),
    is_active         BOOLEAN,
    created_at        TIMESTAMP_NTZ,
    updated_at        TIMESTAMP_NTZ,
    deleted_at        TIMESTAMP_NTZ,
    _pipeline_run_id  VARCHAR(64),
    PRIMARY KEY (customer_id)
);

-- Staging table: customers_stage
CREATE TABLE customers_stage (
    customer_id       VARCHAR(36),
    email             VARCHAR(255),
    full_name         VARCHAR(255),
    plan_tier         VARCHAR(50),
    mrr_usd           NUMERIC(12,2),
    is_active         BOOLEAN,
    created_at        TIMESTAMP_NTZ,
    updated_at        TIMESTAMP_NTZ,
    deleted_at        TIMESTAMP_NTZ,
    _source_event_ts  TIMESTAMP_NTZ,  -- CDC event timestamp for dedup
    _pipeline_run_id  VARCHAR(64)
);

The _pipeline_run_id on the target table is your audit trail — it tells you which pipeline run last touched each row. The _source_event_ts on staging is the CDC event timestamp used for deduplication.

---

Snowflake: MERGE Semantics and Implementation

Snowflake's MERGE is ANSI SQL-compliant and executes as a single atomic transaction. The key behavioral detail to understand is how Snowflake handles multiple source rows matching a single target row.

The Multiple-Match Problem

By default, Snowflake raises an error if a MERGE statement contains multiple source rows matching the same target row:

SQL compilation error: Multiple rows in source table are matched to same row in target table. This could result in non-deterministic or erroneous results during MERGE.

This is actually a feature, not a bug — it's protecting you from a class of bugs where your staging table has duplicates and you'd get non-deterministic updates. Many engineers work around this by adding DISTINCT or restructuring their source subquery. The right fix is to deduplicate before the MERGE.

Deduplication with Window Functions

-- Step 3: Create a deduplicated view of staging
CREATE OR REPLACE TEMPORARY TABLE customers_stage_deduped AS
SELECT *
FROM (
    SELECT
        *,
        ROW_NUMBER() OVER (
            PARTITION BY customer_id
            ORDER BY _source_event_ts DESC, _pipeline_run_id DESC
        ) AS rn
    FROM customers_stage
)
WHERE rn = 1;

Using ROW_NUMBER() with PARTITION BY customer_id ORDER BY _source_event_ts DESC gives you the most recent event per customer. If your CDC source can emit multiple events per key within a single batch (common in Debezium-style streams), this deduplication is mandatory.

The Full Snowflake MERGE

MERGE INTO customers AS tgt
USING customers_stage_deduped AS src
    ON tgt.customer_id = src.customer_id
WHEN MATCHED AND src.deleted_at IS NOT NULL THEN
    -- Soft delete: mark the row as deleted rather than removing it
    UPDATE SET
        tgt.deleted_at       = src.deleted_at,
        tgt.is_active        = FALSE,
        tgt.updated_at       = src.updated_at,
        tgt._pipeline_run_id = src._pipeline_run_id
WHEN MATCHED AND src.deleted_at IS NULL THEN
    -- Standard update: only write if source is newer
    UPDATE SET
        tgt.email            = src.email,
        tgt.full_name        = src.full_name,
        tgt.plan_tier        = src.plan_tier,
        tgt.mrr_usd          = src.mrr_usd,
        tgt.is_active        = src.is_active,
        tgt.updated_at       = src.updated_at,
        tgt._pipeline_run_id = src._pipeline_run_id
WHEN NOT MATCHED THEN
    INSERT (
        customer_id, email, full_name, plan_tier, mrr_usd,
        is_active, created_at, updated_at, deleted_at, _pipeline_run_id
    )
    VALUES (
        src.customer_id, src.email, src.full_name, src.plan_tier, src.mrr_usd,
        src.is_active, src.created_at, src.updated_at, src.deleted_at,
        src._pipeline_run_id
    );

Notice the ordering of the WHEN MATCHED clauses. Snowflake evaluates them in order and applies the first matching clause, so putting the soft-delete check first ensures deleted rows don't accidentally trigger the standard update path.

Snowflake Performance Tuning

For large tables, MERGE performance in Snowflake depends primarily on two things: clustering and warehouse size.

Clustering keys: If your target table is large (say, 500M+ rows) and your primary key is a UUID or a high-cardinality string, Snowflake will scan more micro-partitions than necessary to find matches. Define a clustering key on the join column if it aligns with a common filter pattern:

ALTER TABLE customers CLUSTER BY (customer_id);

But be careful — automatic clustering has a cost. For tables where the primary key has no natural ordering correlation with ingestion time, clustering by a derived bucket can help:

-- For UUID primary keys, clustering on a hash bucket reduces scan waste
ALTER TABLE customers CLUSTER BY (HASH(customer_id, 100));

Warehouse sizing: MERGE in Snowflake is a memory-intensive operation because the query engine builds hash tables for the join. For large staging tables, a single X-Large warehouse often outperforms two Large warehouses running concurrently because the hash table fits in a single node's memory rather than being split across nodes. If you're seeing spill-to-disk warnings in your query profile, scale up before scaling out.

Transient staging tables: Using CREATE TRANSIENT TABLE for your staging table eliminates fail-safe storage costs (the extra 7-day historical data Snowflake keeps for disaster recovery). Since staging tables are ephemeral by nature, you don't need that protection:

CREATE OR REPLACE TRANSIENT TABLE customers_stage (...);
CREATE OR REPLACE TRANSIENT TABLE customers_stage_deduped (...);

On tables refreshed daily with tens of millions of rows, this can save meaningful storage costs.

---

BigQuery: MERGE Semantics and Implementation

BigQuery's MERGE follows the ANSI standard but has a few platform-specific behaviors and limitations that significantly affect your implementation strategy.

Partition Filters and Cost Control

BigQuery prices queries by bytes scanned. A MERGE on a 10TB table without a partition filter can be ruinously expensive. For large target tables, always partition and push a partition filter into your MERGE:

-- Target table with partition pruning
CREATE TABLE `project.dataset.customers`
PARTITION BY DATE(updated_at)
OPTIONS (require_partition_filter = FALSE)
AS SELECT ...;

Then in your MERGE, use a subquery that explicitly bounds the target partition being modified:

MERGE `project.dataset.customers` AS tgt
USING (
    SELECT * FROM `project.dataset.customers_stage`
    WHERE DATE(updated_at) >= DATE_SUB(CURRENT_DATE(), INTERVAL 7 DAY)
) AS src
ON tgt.customer_id = src.customer_id
   AND DATE(tgt.updated_at) >= DATE_SUB(CURRENT_DATE(), INTERVAL 7 DAY)
WHEN MATCHED AND src.deleted_at IS NOT NULL THEN
    UPDATE SET
        tgt.deleted_at       = src.deleted_at,
        tgt.is_active        = FALSE,
        tgt.updated_at       = src.updated_at,
        tgt._pipeline_run_id = src._pipeline_run_id
WHEN MATCHED AND src.deleted_at IS NULL THEN
    UPDATE SET
        tgt.email            = src.email,
        tgt.full_name        = src.full_name,
        tgt.plan_tier        = src.plan_tier,
        tgt.mrr_usd          = src.mrr_usd,
        tgt.is_active        = src.is_active,
        tgt.updated_at       = src.updated_at,
        tgt._pipeline_run_id = src._pipeline_run_id
WHEN NOT MATCHED THEN
    INSERT ROW;

The INSERT ROW shorthand inserts all columns from the source — convenient but use it only when source and target schemas are guaranteed to match.

BigQuery-Specific Limitation: No Multiple Matches Protection

Unlike Snowflake, BigQuery does not raise an error on multiple source rows matching the same target row — it silently picks one. This makes deduplication even more critical in BigQuery because you won't get a runtime error to warn you that something went wrong.

-- Always deduplicate staging before MERGE in BigQuery
CREATE OR REPLACE TEMP TABLE customers_stage_deduped AS
SELECT * EXCEPT(rn)
FROM (
    SELECT
        *,
        ROW_NUMBER() OVER (
            PARTITION BY customer_id
            ORDER BY _source_event_ts DESC
        ) AS rn
    FROM `project.dataset.customers_stage`
)
WHERE rn = 1;

The INSERT-OVERWRITE Alternative

For use cases where your pipeline loads a complete snapshot of a partition (rather than a CDC delta), BigQuery offers a compelling alternative to MERGE: partition overwrite. If you're loading all customer records updated in the last 24 hours and that set is small relative to your table, MERGE is the right tool. But if you're reprocessing an entire date partition from source, a partition overwrite is simpler and often faster:

INSERT OVERWRITE `project.dataset.customers`
PARTITION (DATE(updated_at) = '2024-01-15')
SELECT * FROM `project.dataset.customers_stage`
WHERE DATE(updated_at) = '2024-01-15';

This approach is inherently idempotent — you're replacing a partition atomically, so re-running it always produces the same result. It's not a MERGE, but it solves the idempotency problem elegantly for partition-scoped loads. The trade-off is that it doesn't handle cross-partition updates gracefully (a customer whose updated_at changed from yesterday to today spans two partitions).

BigQuery Performance Tuning

BigQuery's distributed execution model means that MERGE performance is primarily driven by data shuffle cost — how much data needs to move across slots during the join. Keep staging tables small and well-filtered.

Clustering on join keys: BigQuery clustering reduces bytes scanned during the MERGE by co-locating related rows in the same storage blocks:

CREATE TABLE `project.dataset.customers`
PARTITION BY DATE(updated_at)
CLUSTER BY customer_id
AS SELECT ...;

Staging as an external table or temp table: If your source data lives in Cloud Storage (GCS), you can reference it directly as an external table in your MERGE without a staging copy step. This reduces latency and storage costs for large batches.

---

Redshift: MERGE Semantics and Implementation

Redshift added native MERGE support in 2022, but many production Redshift environments still use the older DELETE+INSERT pattern for compatibility and performance reasons. We'll cover both, because understanding why the old pattern persisted teaches you a lot about Redshift's architecture.

Why Redshift Was Late to MERGE

Redshift is a columnar, massively parallel processing (MPP) database. Its storage model is optimized for large sequential scans, not random point lookups. Implementing MERGE efficiently on an MPP columnar store is genuinely hard — updates are particularly expensive because Redshift doesn't update in place; it marks old rows as deleted and writes new versions, which requires a VACUUM to reclaim space.

Native MERGE Syntax (Redshift 2022+)

MERGE INTO customers AS tgt
USING customers_stage_deduped AS src
ON tgt.customer_id = src.customer_id
REMOVE DUPLICATES;

The REMOVE DUPLICATES clause is Redshift-specific shorthand: when a source row matches a target row and all non-key columns are identical, skip the update. This avoids unnecessary write amplification and is critical for idempotent reruns where most rows haven't changed.

For our full upsert with soft deletes:

MERGE INTO customers AS tgt
USING customers_stage_deduped AS src
ON tgt.customer_id = src.customer_id
WHEN MATCHED AND src.deleted_at IS NOT NULL THEN
    UPDATE SET
        deleted_at       = src.deleted_at,
        is_active        = FALSE,
        updated_at       = src.updated_at,
        _pipeline_run_id = src._pipeline_run_id
WHEN MATCHED AND src.deleted_at IS NULL THEN
    UPDATE SET
        email            = src.email,
        full_name        = src.full_name,
        plan_tier        = src.plan_tier,
        mrr_usd          = src.mrr_usd,
        is_active        = src.is_active,
        updated_at       = src.updated_at,
        _pipeline_run_id = src._pipeline_run_id
WHEN NOT MATCHED THEN
    INSERT (customer_id, email, full_name, plan_tier, mrr_usd,
            is_active, created_at, updated_at, deleted_at, _pipeline_run_id)
    VALUES (src.customer_id, src.email, src.full_name, src.plan_tier, src.mrr_usd,
            src.is_active, src.created_at, src.updated_at, src.deleted_at,
            src._pipeline_run_id);

The Classic DELETE+INSERT Pattern (for Legacy Compatibility)

Before native MERGE, the idiomatic Redshift upsert looked like this:

BEGIN TRANSACTION;

-- Step 1: Delete target rows that exist in staging
DELETE FROM customers
WHERE customer_id IN (
    SELECT customer_id FROM customers_stage_deduped
);

-- Step 2: Insert all staging rows (both new and updated)
INSERT INTO customers
SELECT
    customer_id, email, full_name, plan_tier, mrr_usd,
    is_active, created_at, updated_at, deleted_at, _pipeline_run_id
FROM customers_stage_deduped;

COMMIT;

This pattern is idempotent when wrapped in a transaction: the DELETE removes old versions and the INSERT adds new ones. If the transaction fails, neither change is committed. The downside is that you lose all historical values for updated rows — which is fine if your target table is an SCD Type 1 (current state only) but wrong for SCD Type 2 scenarios.

For slowly changing dimensions in Redshift, you'll typically need a more complex transaction that handles the "close the old record, open a new one" logic explicitly.

Redshift Performance Tuning

Distribution style and sort keys are the two most important performance levers in Redshift, and they directly affect MERGE performance.

Distribution style: For MERGE workloads, you want the staging table and target table distributed on the same key (ideally the join key) so that matching rows are co-located on the same node and don't require a network shuffle:

-- Target table: distribute on customer_id
CREATE TABLE customers (
    customer_id VARCHAR(36),
    ...
)
DISTKEY (customer_id)
SORTKEY (updated_at);

-- Staging table: same distribution key
CREATE TABLE customers_stage (
    customer_id VARCHAR(36),
    ...
)
DISTKEY (customer_id);

When both tables use the same DISTKEY, the MERGE join happens node-local and avoids expensive redistribute broadcasts.

VACUUM and ANALYZE after MERGE: Redshift MERGEs generate table bloat (soft-deleted rows). Run VACUUM after large merges:

VACUUM DELETE ONLY customers;
ANALYZE customers;

Scheduling VACUUM DELETE ONLY (which only reclaims space, not re-sorts) is faster than a full VACUUM and is usually sufficient for merge-heavy tables.

Sort key alignment: If your staging table is loaded in sorted order (e.g., rows ordered by updated_at), Redshift can use zone maps to skip large portions of the target table during the DELETE phase of your MERGE. Load staging data in sort key order when possible.

---

Handling Schema Evolution Inside MERGE

One of the trickiest production scenarios is what happens when your source adds a new column. If you're using INSERT ROW in BigQuery or a wildcard insert in Redshift, a new source column will cause your MERGE to fail or silently drop data.

The defensive pattern is to always explicitly list columns in your INSERT clause — never use SELECT * or INSERT ROW in production MERGE statements unless you have a schema validation step immediately before:

-- Explicit column list: safe against source schema additions
WHEN NOT MATCHED THEN
    INSERT (customer_id, email, full_name, plan_tier, mrr_usd,
            is_active, created_at, updated_at, deleted_at, _pipeline_run_id)
    VALUES (src.customer_id, src.email, src.full_name, src.plan_tier,
            src.mrr_usd, src.is_active, src.created_at, src.updated_at,
            src.deleted_at, src._pipeline_run_id);

When you do need to evolve the schema, add the new column to the target table first, then update the MERGE statement, then deploy. This backward-compatible order prevents pipeline failures. Schema evolution strategies for pipelines are worth reading in depth — the MERGE context adds some specific complications around column ordering in SELECT lists.

---

Optimistic vs. Conditional Updates

A subtle but important optimization: don't update rows that haven't actually changed. Most MERGE workbooks update every matched row unconditionally. On large tables with a small delta, this generates unnecessary write I/O and bloat:

-- Conditional update: only write if something actually changed
WHEN MATCHED AND src.deleted_at IS NULL
    AND (
        tgt.email     IS DISTINCT FROM src.email
        OR tgt.full_name  IS DISTINCT FROM src.full_name
        OR tgt.plan_tier  IS DISTINCT FROM src.plan_tier
        OR tgt.mrr_usd    IS DISTINCT FROM src.mrr_usd
        OR tgt.is_active  IS DISTINCT FROM src.is_active
    )
THEN
    UPDATE SET ...

IS DISTINCT FROM is NULL-safe (unlike !=), so it correctly handles cases where either side is NULL. This pattern is supported in Snowflake, BigQuery, and Redshift.

On a typical SaaS customer table where only 2-3% of customers change per day, this optimization eliminates 97% of unnecessary row updates and can reduce MERGE runtime by 40-60% on large tables.

---

Pipeline Orchestration and Transaction Boundaries

The MERGE statement alone doesn't make your pipeline idempotent — the transaction boundary and retry logic around it matter just as much. Consider this Airflow DAG structure:

# Conceptual DAG structure
extract_to_gcs = ...        # Task 1: Extract CRM data to GCS
load_to_staging = ...       # Task 2: COPY from GCS to staging table
deduplicate_staging = ...   # Task 3: CREATE TABLE ... AS SELECT ... (dedup)
merge_to_target = ...       # Task 4: MERGE into target table
log_run_metadata = ...      # Task 5: INSERT into pipeline_runs audit table

Notice that Tasks 1-5 are separate DAG tasks. If Task 4 (the MERGE) succeeds but Task 5 (the audit log) fails, and Airflow retries from Task 4, you'll run the MERGE twice on the same data. Because the MERGE is idempotent, that's fine — it produces the same result. The audit log might get a duplicate entry, but that's a minor issue with a simple fix (use an INSERT ... ON CONFLICT DO NOTHING with a unique constraint on run_id).

Where it gets dangerous is if your retry logic starts at Task 1 and re-extracts data. If the source system has changed between the original run and the retry (new records added, records modified), your staging table will contain different data than the first attempt. The MERGE will then apply different changes than originally intended.

The solution is to make your extract either append-only with deterministic bounds, or to checkpoint your extracts so retries reload from the same snapshot. This connects to the broader problem of checkpointing and state management in long-running pipelines.

---

Soft Deletes vs. Hard Deletes in MERGE

Your MERGE strategy changes significantly depending on whether you need to hard-delete records from the target table. Hard deletes inside a MERGE require the WHEN NOT MATCHED BY SOURCE THEN DELETE clause, which is supported in Snowflake and Redshift but has some nuances in BigQuery.

For the CRM customer scenario, a common requirement is to process delete events from a CDC stream. The cleanest approach is soft deletes — mark the row as deleted but keep it in the table:

WHEN MATCHED AND src.operation_type = 'DELETE' THEN
    UPDATE SET
        tgt.deleted_at       = src.event_timestamp,
        tgt.is_active        = FALSE,
        tgt._pipeline_run_id = src._pipeline_run_id

Soft deletes are inherently idempotent — running the same delete event twice just sets the same deleted_at timestamp again. Hard deletes are trickier because WHEN NOT MATCHED BY SOURCE THEN DELETE deletes all target rows not present in the current staging batch, which on a partial load would incorrectly delete rows that simply weren't in the current increment.

If you need hard deletes, they should only be triggered by explicit delete events in your staging table, not by the absence of a row in the staging batch:

-- Explicit hard delete based on staged event, not absence in staging
WHEN MATCHED AND src.operation_type = 'DELETE' THEN DELETE

This is a much safer pattern than "delete anything not in source."

---

Testing Your MERGE for Idempotency

An idempotent pipeline that you haven't tested isn't actually idempotent — it's just untested. Here's a repeatable test protocol you can run against any MERGE implementation.

Test 1: Double-run test

def test_merge_idempotency(warehouse_conn, staging_rows, run_id):
    # Load staging and run merge once
    load_staging(warehouse_conn, staging_rows, run_id)
    execute_merge(warehouse_conn)
    snapshot_1 = query_target(warehouse_conn)

    # Reset staging to same data, run merge again
    truncate_staging(warehouse_conn)
    load_staging(warehouse_conn, staging_rows, run_id)
    execute_merge(warehouse_conn)
    snapshot_2 = query_target(warehouse_conn)

    # Both snapshots should be identical
    assert snapshot_1.equals(snapshot_2), "MERGE is not idempotent"

Test 2: Duplicate source test

def test_merge_with_duplicate_source(warehouse_conn, staging_rows, run_id):
    # Load duplicate rows (same customer_id, different event timestamps)
    duplicate_rows = staging_rows + staging_rows
    load_staging(warehouse_conn, duplicate_rows, run_id)
    execute_merge(warehouse_conn)

    # Verify no duplicate customer_ids in target
    duplicate_check = query(
        "SELECT customer_id, COUNT(*) FROM customers GROUP BY 1 HAVING COUNT(*) > 1"
    )
    assert duplicate_check.empty, "MERGE produced duplicate rows from duplicate source"

Test 3: Stale update test

def test_merge_does_not_apply_stale_updates(warehouse_conn):
    # Insert a row with timestamp T+2 into target
    insert_target(warehouse_conn, customer_id='abc', updated_at=T_plus_2)

    # Stage a row with timestamp T+1 (older)
    load_staging(warehouse_conn, customer_id='abc', updated_at=T_plus_1)
    execute_merge(warehouse_conn)

    # Verify target still has T+2 timestamp (stale update was rejected)
    result = query_target(warehouse_conn, customer_id='abc')
    assert result['updated_at'] == T_plus_2, "MERGE applied a stale update"

This third test will fail unless your MERGE includes a timestamp guard in the WHEN MATCHED condition:

WHEN MATCHED AND src.updated_at > tgt.updated_at AND src.deleted_at IS NULL THEN
    UPDATE SET ...

Adding src.updated_at > tgt.updated_at ensures you never overwrite a newer row with an older version — critical for CDC pipelines where out-of-order delivery is possible. This connects directly to the patterns discussed in handling late-arriving and out-of-order data in production pipelines.

---

Hands-On Exercise

Let's put this together in a complete, runnable exercise. You'll implement an idempotent MERGE pipeline for an e-commerce orders table in Snowflake. If you're using BigQuery or Redshift, the structure transfers directly with the platform-specific syntax adjustments we've covered.

Setup:

-- Create target table
CREATE OR REPLACE TABLE orders (
    order_id        VARCHAR(36)   NOT NULL PRIMARY KEY,
    customer_id     VARCHAR(36)   NOT NULL,
    order_status    VARCHAR(50),
    total_amount    NUMERIC(12,2),
    currency        CHAR(3),
    placed_at       TIMESTAMP_NTZ,
    fulfilled_at    TIMESTAMP_NTZ,
    cancelled_at    TIMESTAMP_NTZ,
    updated_at      TIMESTAMP_NTZ,
    _pipeline_run_id VARCHAR(64)
);

-- Create transient staging table
CREATE OR REPLACE TRANSIENT TABLE orders_stage (
    order_id        VARCHAR(36),
    customer_id     VARCHAR(36),
    order_status    VARCHAR(50),
    total_amount    NUMERIC(12,2),
    currency        CHAR(3),
    placed_at       TIMESTAMP_NTZ,
    fulfilled_at    TIMESTAMP_NTZ,
    cancelled_at    TIMESTAMP_NTZ,
    updated_at      TIMESTAMP_NTZ,
    _source_event_ts TIMESTAMP_NTZ,
    _pipeline_run_id VARCHAR(64)
);

Load test data simulating a CDC batch:

-- Simulate CDC events: order created, then updated, with a duplicate event
INSERT INTO orders_stage VALUES
-- New order created
('ord-001', 'cust-abc', 'PLACED', 199.99, 'USD',
 '2024-01-15 10:00:00', NULL, NULL, '2024-01-15 10:00:00',
 '2024-01-15 10:00:01', 'run-20240115-001'),
-- Order fulfilled (newer event, same order)
('ord-001', 'cust-abc', 'FULFILLED', 199.99, 'USD',
 '2024-01-15 10:00:00', '2024-01-15 14:30:00', NULL, '2024-01-15 14:30:00',
 '2024-01-15 14:30:01', 'run-20240115-001'),
-- Duplicate of the fulfilled event (should be deduped)
('ord-001', 'cust-abc', 'FULFILLED', 199.99, 'USD',
 '2024-01-15 10:00:00', '2024-01-15 14:30:00', NULL, '2024-01-15 14:30:00',
 '2024-01-15 14:30:01', 'run-20240115-001'),
-- Brand new order
('ord-002', 'cust-def', 'PLACED', 549.00, 'USD',
 '2024-01-15 11:00:00', NULL, NULL, '2024-01-15 11:00:00',
 '2024-01-15 11:00:05', 'run-20240115-001');

Run the deduplication:

CREATE OR REPLACE TEMPORARY TABLE orders_stage_deduped AS
SELECT * EXCLUDE(rn)
FROM (
    SELECT
        *,
        ROW_NUMBER() OVER (
            PARTITION BY order_id
            ORDER BY _source_event_ts DESC
        ) AS rn
    FROM orders_stage
)
WHERE rn = 1;

-- Verify: should be 2 rows (ord-001 and ord-002)
SELECT COUNT(*) FROM orders_stage_deduped;

Execute the MERGE:

MERGE INTO orders AS tgt
USING orders_stage_deduped AS src
    ON tgt.order_id = src.order_id
WHEN MATCHED AND src.cancelled_at IS NOT NULL THEN
    UPDATE SET
        tgt.order_status     = 'CANCELLED',
        tgt.cancelled_at     = src.cancelled_at,
        tgt.updated_at       = src.updated_at,
        tgt._pipeline_run_id = src._pipeline_run_id
WHEN MATCHED
    AND src.cancelled_at IS NULL
    AND src.updated_at > tgt.updated_at
THEN
    UPDATE SET
        tgt.order_status     = src.order_status,
        tgt.total_amount     = src.total_amount,
        tgt.fulfilled_at     = src.fulfilled_at,
        tgt.updated_at       = src.updated_at,
        tgt._pipeline_run_id = src._pipeline_run_id
WHEN NOT MATCHED THEN
    INSERT (order_id, customer_id, order_status, total_amount, currency,
            placed_at, fulfilled_at, cancelled_at, updated_at, _pipeline_run_id)
    VALUES (src.order_id, src.customer_id, src.order_status, src.total_amount,
            src.currency, src.placed_at, src.fulfilled_at, src.cancelled_at,
            src.updated_at, src._pipeline_run_id);

Verify idempotency: Run the staging load and MERGE a second time with the same data. Then:

-- Should return 0 rows if truly idempotent
SELECT order_id, COUNT(*) FROM orders GROUP BY 1 HAVING COUNT(*) > 1;

Extend the exercise: Add a cancellation event for ord-001 to the staging table and re-run the full pipeline. Verify that the CANCELLED status is applied correctly and that re-running it a third time doesn't change any values.

---

Common Mistakes & Troubleshooting

Mistake 1: Merging from an un-truncated staging table

Symptom: Rows get duplicated or updated to stale values after a retry.

Fix: Always TRUNCATE TABLE staging or CREATE OR REPLACE TABLE staging AS ... at the start of each pipeline run. The replace pattern is safer because it's atomic — if the load fails, the old staging data is still there from the previous run, which might allow a safe retry without re-extracting.

Mistake 2: Missing the multiple-source-match deduplication in BigQuery

Symptom: Updates are non-deterministic; different reruns produce different results.

Fix: Always use ROW_NUMBER() OVER (PARTITION BY pk ORDER BY ...) before MERGE in BigQuery. Add a data quality assertion that verifies no duplicates exist in your deduplicated staging table before executing the MERGE.

Mistake 3: Forgetting VACUUM after large Redshift MERGEs

Symptom: Queries on the target table get progressively slower over weeks; disk usage grows faster than expected.

Fix: Schedule VACUUM DELETE ONLY after MERGE-heavy pipeline runs. In Airflow, a RedshiftSQLOperator task at the end of your DAG works well. The data quality validation article covers monitoring patterns that will help you catch table bloat before it becomes a performance problem.

Mistake 4: Using WHEN NOT MATCHED BY SOURCE THEN DELETE on partial loads

Symptom: Large numbers of rows mysteriously disappear from the target table.

Fix: Only use WHEN NOT MATCHED BY SOURCE THEN DELETE when your staging table represents a complete snapshot of the source, not an incremental delta. For incremental loads, delete rows only based on explicit delete events in staging.

Mistake 5: Applying updates without a timestamp guard

Symptom: A late-arriving CDC event overwrites newer data in the target table.

Fix: Add AND src.updated_at > tgt.updated_at to every WHEN MATCHED ... UPDATE clause. This is NULL-unsafe, so also add AND tgt.updated_at IS NOT NULL or use IS DISTINCT FROM comparisons.

Mistake 6: Not accounting for NULL-safe comparisons in conditional updates

Symptom: Rows that should trigger conditional updates don't, because a NULL comparison with != evaluates to NULL (falsy).

Fix: Use IS DISTINCT FROM instead of != in your conditional update filters. NULL IS DISTINCT FROM 'value' returns TRUE, while NULL != 'value' returns NULL.

---

Summary & Next Steps

The MERGE statement is one of the most powerful primitives in your data engineering toolkit, but its power comes with significant responsibility for getting the details right. Let's recap the key principles:

1. Isolate staging per run: Truncate or replace staging at the start of each pipeline execution. This is the single most important safety guarantee.

2. Deduplicate before merging: Use ROW_NUMBER() OVER (PARTITION BY pk ORDER BY event_ts DESC) to reduce staging to one row per primary key. This is mandatory in BigQuery (no error on multiple matches) and best practice everywhere else.

3. Use timestamp guards: Add src.updated_at > tgt.updated_at to your update conditions to prevent late-arriving events from overwriting newer data.

4. Use conditional updates for performance: IS DISTINCT FROM comparisons in WHEN MATCHED conditions skip writes for unchanged rows, reducing write amplification on large tables with small deltas.

5. Tune for your platform: Snowflake needs clustering and warehouse sizing; BigQuery needs partition filters; Redshift needs matching distribution keys and post-MERGE VACUUM.

6. Test idempotency explicitly: Double-run tests, duplicate source tests, and stale update tests should be part of your CI pipeline for every MERGE implementation.

From here, there are several natural extensions to explore. If you're building pipelines that need to handle the full complexity of designing idempotent data pipelines with exactly-once semantics, that article extends these patterns to streaming contexts and multi-sink scenarios. For teams dealing with late data and reprocessing requirements, understanding data pipeline backfilling strategies covers how to safely replay historical loads through MERGE-based pipelines. And if your MERGE pipelines are growing expensive to run, cost attribution and pipeline-level resource optimization will help you profile and reduce your warehouse spend.

Idempotent upserts aren't glamorous engineering. But when your orchestrator retries at 3 AM and your revenue data is still exactly right by the time the CFO looks at the dashboard, you'll appreciate that you built it correctly.