Building a Data Stack Live with AI AgentsLivestream August 18

Skip to main content

Data loading patterns

Beyond basic COPY and INSERT statements, production data pipelines often need incremental loads, upserts, and idempotent operations. This guide covers common patterns you can use with MotherDuck.

Incremental loading​

Incremental loading adds only new or changed data to a target table, rather than reloading everything. This reduces processing time and resource usage for large datasets that receive frequent updates.

The core idea is to track a watermark: a column value that marks the boundary between already-loaded data and new data.

Using a timestamp watermark​

If your source data has an updated_at or created_at column, use it to filter for new records:

INSERT INTO analytics.events
SELECT * FROM read_parquet('s3://bucket/events/*.parquet')
WHERE updated_at > (SELECT MAX(updated_at) FROM analytics.events);

Using a monotonic ID watermark​

If your data has an auto-incrementing ID and records are never updated after creation, an ID-based watermark avoids timestamp precision issues:

INSERT INTO analytics.events
SELECT * FROM read_parquet('s3://bucket/events/*.parquet')
WHERE event_id > (SELECT COALESCE(MAX(event_id), 0) FROM analytics.events);
tip

Timestamp watermarks handle both new and updated records. ID-based watermarks only catch new records but avoid issues with clock skew and timestamp precision. Choose based on whether your source data gets updated in place.

Handling late-arriving data​

Data doesn't always arrive in order. Sensors go offline, mobile apps sync late, and distributed systems have clock skew. To account for this, subtract a safety buffer from your watermark:

INSERT INTO analytics.events
SELECT * FROM read_parquet('s3://bucket/events/*.parquet')
WHERE updated_at > (
    SELECT MAX(updated_at) - INTERVAL 2 HOURS
    FROM analytics.events
);

Combine this with deduplication (see below) to prevent duplicate rows from the overlapping window.

Upserts​

An upsert inserts new rows and updates existing ones in a single operation. DuckDB supports three syntaxes for this.

INSERT OR REPLACE​

The simplest approach: replaces the entire row when a conflict occurs on the primary key.

CREATE TABLE customers (
    id INTEGER PRIMARY KEY,
    name VARCHAR,
    email VARCHAR,
    updated_at TIMESTAMP
);

INSERT OR REPLACE INTO customers
SELECT * FROM read_csv('new_customers.csv');

INSERT ON CONFLICT​

For more control, use ON CONFLICT to update only specific columns:

INSERT INTO customers (id, name, email, updated_at)
SELECT * FROM read_csv('updates.csv')
ON CONFLICT (id) DO UPDATE SET
    name = EXCLUDED.name,
    email = EXCLUDED.email,
    updated_at = EXCLUDED.updated_at;

You can also use ON CONFLICT ... DO NOTHING to silently skip duplicates:

INSERT INTO customers (id, name, email, updated_at)
SELECT * FROM read_csv('updates.csv')
ON CONFLICT (id) DO NOTHING;
warning

ON CONFLICT requires a PRIMARY KEY or UNIQUE constraint on the conflict column(s). Without one, DuckDB raises an error. INSERT OR REPLACE also requires a PRIMARY KEY or UNIQUE constraint.

MERGE INTO​

MERGE INTO (DuckDB 1.4 and later) performs standard SQL upserts without requiring a primary key or unique constraint, which makes it the best fit for analytical tables that don't define keys:

MERGE INTO customers AS t
USING (SELECT * FROM read_csv('updates.csv')) AS s
ON t.id = s.id
WHEN MATCHED THEN UPDATE SET
    name = s.name,
    email = s.email,
    updated_at = s.updated_at
WHEN NOT MATCHED THEN INSERT (id, name, email, updated_at)
    VALUES (s.id, s.name, s.email, s.updated_at);

MERGE INTO also supports WHEN NOT MATCHED BY SOURCE clauses for handling rows that exist in the target but not in the source, such as deleting records that disappeared upstream. See the DuckDB MERGE INTO documentation for the full syntax.

Full refresh with swap​

When your dataset is small enough to reload entirely, or when incremental logic would be too complex, a full refresh with a table swap is the simplest reliable pattern:

-- Load into a staging table
CREATE OR REPLACE TABLE staging_products AS
SELECT * FROM read_parquet('s3://bucket/products/*.parquet');

-- Swap the tables
DROP TABLE IF EXISTS products;
ALTER TABLE staging_products RENAME TO products;
tip

This pattern is naturally idempotent: running it twice produces the same result. It also avoids issues with partial updates since the old table stays intact until the swap.

Deduplication on load​

Source data often contains duplicates, especially when replaying events or combining overlapping file batches. Use ROW_NUMBER() to keep only the most recent version of each record:

INSERT INTO events
SELECT * EXCLUDE (rn) FROM (
    SELECT *,
        ROW_NUMBER() OVER (
            PARTITION BY event_id
            ORDER BY received_at DESC
        ) AS rn
    FROM read_parquet('s3://bucket/events/*.parquet')
) WHERE rn = 1;

The EXCLUDE (rn) clause drops the helper row-number column so the inserted rows match the target schema.

For an incremental load with deduplication, combine the watermark filter with the dedup logic:

INSERT OR REPLACE INTO events
SELECT * EXCLUDE (rn) FROM (
    SELECT *,
        ROW_NUMBER() OVER (
            PARTITION BY event_id
            ORDER BY received_at DESC
        ) AS rn
    FROM read_parquet('s3://bucket/events/*.parquet')
    WHERE received_at > (
        SELECT MAX(received_at) - INTERVAL 2 HOURS
        FROM events
    )
) WHERE rn = 1;

Idempotent loads with transactions​

Wrap multi-step loads in a transaction so that either all steps succeed or none do. This prevents partial loads from leaving your data in an inconsistent state:

BEGIN TRANSACTION;

-- Step 1: Load new data into staging
CREATE OR REPLACE TABLE staging_orders AS
SELECT * FROM read_parquet('s3://bucket/daily/2026-03-10/*.parquet');

-- Step 2: Delete existing records for the same date range (idempotent reload)
DELETE FROM orders
WHERE order_date IN (SELECT DISTINCT order_date FROM staging_orders);

-- Step 3: Insert deduplicated staging data
INSERT INTO orders
SELECT * EXCLUDE (rn) FROM (
    SELECT *,
        ROW_NUMBER() OVER (PARTITION BY order_id ORDER BY updated_at DESC) AS rn
    FROM staging_orders
) WHERE rn = 1;

-- Step 4: Clean up
DROP TABLE staging_orders;

COMMIT;
tip

Using CREATE OR REPLACE for staging tables makes each step idempotent. If a load fails partway through and you re-run it, the staging table is recreated from scratch.

Scheduling and automating loads​

For production pipelines that run on a schedule, consider these options:

See also​