Full-reload ETL jobs that process millions of unchanged rows are a solved problem — if you know the patterns. This lesson teaches production-grade incremental pipeline design using watermark tables and Change Data Capture, with complete SQL code for handling updates, deletes, and failure recovery.
Picture this: you have a nightly ETL job that refreshes your analytics warehouse by pulling every row from a 50-million-record orders table. It takes four hours. Your stakeholders want near-real-time dashboards. Your DBA is quietly building a case for why you shouldn't be allowed near production databases. There's a better way.
Incremental processing is the practice of identifying and moving only the records that have changed since the last pipeline run, rather than re-reading an entire source table from scratch. Done right, it can reduce your pipeline runtime from hours to minutes, slash I/O costs, and make near-real-time data ingestion actually achievable with plain SQL — no exotic streaming infrastructure required. The two foundational techniques are watermark tables (a lightweight bookmarking pattern) and Change Data Capture (reading from the database's own change log). This lesson covers both, from theory to production-grade implementation.
By the end of this lesson, you will be able to design and deploy incremental SQL pipelines that handle real-world edge cases: late-arriving records, soft deletes, schema drift, and multi-source fan-out. This isn't a 30-second blog post — we're going to build the real thing.
What you'll learn:
You should be comfortable with SQL joins, aggregations, and subqueries. Familiarity with MERGE statements and upsert patterns will help — if you need a refresher, the lesson on Bulk Data Loading and Upsert Patterns in SQL: MERGE, INSERT ON CONFLICT, and Incremental Load Strategies for Production Pipelines covers that ground well. You should also understand how transactions work, since watermark management depends on atomic commits — see SQL Transactions, Isolation Levels, and Locking: A Complete Guide to Concurrent Database Programming for background.
The examples below use PostgreSQL syntax except where noted. Equivalent patterns for SQL Server, BigQuery, and Snowflake are called out where they differ meaningfully.
Let's be precise about what makes full reloads painful, because understanding the pain tells you exactly what an incremental design needs to solve.
A full reload has three costs:
Incremental designs attack all three. But they introduce a new requirement: you need a reliable way to know which records changed and when. That's the fundamental problem every technique in this lesson is solving.
There are two broad strategies:
Each has different strengths. Watermarks are simpler to implement but can miss hard deletes. CDC captures everything but requires database-level configuration and appropriate permissions.
A watermark (sometimes called a "high-water mark") is a stored marker that represents how far through a source dataset your pipeline has successfully read. The simplest possible watermark is a single timestamp value stored somewhere durable.
Don't store your watermark in application config or environment variables. It needs to live in a database so it's durable, transactional, and queryable.
CREATE TABLE pipeline_watermarks (
pipeline_name VARCHAR(200) NOT NULL,
source_table VARCHAR(200) NOT NULL,
last_run_at TIMESTAMPTZ NOT NULL,
last_max_value TIMESTAMPTZ, -- the actual watermark value from source
records_processed BIGINT DEFAULT 0,
run_status VARCHAR(20) NOT NULL DEFAULT 'pending',
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
PRIMARY KEY (pipeline_name, source_table)
);
A few design choices worth calling out:
last_run_at vs last_max_value: These are different things. last_run_at is when your pipeline executed. last_max_value is the maximum updated_at timestamp your pipeline actually observed in the source data. You want last_max_value — because if your pipeline ran at 3:00am but the newest record in the source had a timestamp of 2:58am, your next watermark should be 2:58am, not 3:00am.run_status: Lets you detect and recover from partial failures. If a pipeline is stuck in 'running' status for longer than expected, something went wrong.Assume your source table looks like this:
CREATE TABLE orders (
order_id BIGINT PRIMARY KEY,
customer_id BIGINT NOT NULL,
order_total NUMERIC(12,2),
order_status VARCHAR(50),
created_at TIMESTAMPTZ NOT NULL,
updated_at TIMESTAMPTZ NOT NULL
);
CREATE INDEX idx_orders_updated_at ON orders(updated_at);
Warning: If your source table doesn't have an
updated_atcolumn (or equivalent), the watermark pattern becomes much harder to implement reliably. Before designing your pipeline, audit the source schema. Many legacy systems only havecreated_at— those tables can only capture new inserts, not updates.
The extraction query reads records that have changed since the last watermark:
-- Step 1: Read the current watermark
SELECT last_max_value
FROM pipeline_watermarks
WHERE pipeline_name = 'orders_to_warehouse'
AND source_table = 'orders';
-- Step 2: Extract changed records (using the watermark value from step 1)
-- Replace :last_watermark with the value retrieved above
SELECT
order_id,
customer_id,
order_total,
order_status,
created_at,
updated_at
FROM orders
WHERE updated_at > :last_watermark
AND updated_at <= NOW() - INTERVAL '5 seconds' -- safety buffer (explained below)
ORDER BY updated_at ASC;
That NOW() - INTERVAL '5 seconds' upper bound is not paranoia — it's essential. Without it, you can miss records that are being written to the source table in the same second your pipeline runs. If a transaction commits while you're reading, you might get an incomplete view of that batch. The safety buffer ensures you only read records that are safely committed and not actively being written.
There's a subtle bug in the simple > :last_watermark pattern. Suppose your last watermark was 2024-03-15 09:00:00.000000. You extract all records where updated_at > '2024-03-15 09:00:00.000000'. Next run, your new watermark is 2024-03-15 09:05:00.000000.
What if two records were updated at exactly 2024-03-15 09:05:00.000000? Your query captured them in this run. But your next run uses > '2024-03-15 09:05:00.000000', which excludes records at exactly that timestamp. If a third record was also updated at that exact microsecond and wasn't captured in this run (rare, but possible in high-write environments), it will be silently dropped.
The safer pattern uses >= on the lower bound with deduplication:
-- Use >= on the lower bound, accept that you'll re-process some records
SELECT
order_id,
customer_id,
order_total,
order_status,
created_at,
updated_at
FROM orders
WHERE updated_at >= :last_watermark -- >= not >
AND updated_at < :upper_bound
ORDER BY updated_at ASC;
Then in your destination, use an upsert (INSERT ... ON CONFLICT DO UPDATE or MERGE) rather than a plain insert. Duplicate rows just overwrite themselves. This "re-process at boundaries" approach trades a tiny amount of redundant work for correctness guarantees.
This is where most homegrown watermark systems fail. They update the watermark after writing to the destination, which seems logical but creates a race condition: if the destination write succeeds but the watermark update fails, the next pipeline run will re-process the same batch. If the destination write fails but the watermark was already updated, you'll skip records permanently.
The correct approach: update the watermark in the same transaction as your destination write, or treat the watermark as the last thing you update, and ensure your destination writes are idempotent.
BEGIN;
-- Write extracted records to destination (upsert so re-runs are safe)
INSERT INTO warehouse.orders (
order_id, customer_id, order_total, order_status, created_at, updated_at
)
SELECT
order_id, customer_id, order_total, order_status, created_at, updated_at
FROM staging.orders_batch -- loaded from extraction step
ON CONFLICT (order_id) DO UPDATE SET
customer_id = EXCLUDED.customer_id,
order_total = EXCLUDED.order_total,
order_status = EXCLUDED.order_status,
updated_at = EXCLUDED.updated_at;
-- Update the watermark only after the destination write succeeds
UPDATE pipeline_watermarks
SET
last_max_value = (SELECT MAX(updated_at) FROM staging.orders_batch),
records_processed = (SELECT COUNT(*) FROM staging.orders_batch),
run_status = 'success',
updated_at = NOW()
WHERE pipeline_name = 'orders_to_warehouse'
AND source_table = 'orders';
COMMIT;
If anything in this block fails, the ROLLBACK leaves both the destination and the watermark unchanged. The next run re-processes the same batch. Since the destination upsert is idempotent, re-processing is harmless.
Tip: If your source and destination live in different databases (common in real pipelines), you can't use a single SQL transaction. In that case, adopt the "write destination first, update watermark second" order, and make your destination writes idempotent. A failed watermark update means duplicate processing on retry — which your upsert handles — rather than skipped records, which would be data loss.
Timestamps aren't the only watermark mechanism. In many systems, auto-incrementing primary keys are a more reliable signal for new records (though not for updates). If your source table has a monotonically increasing order_id, you can watermark on that:
SELECT order_id, customer_id, order_total, order_status, created_at, updated_at
FROM orders
WHERE order_id > :last_max_order_id
ORDER BY order_id ASC
LIMIT 50000; -- process in chunks
Sequence-based watermarks are immune to clock skew and timezone confusion. Their limitation: they only capture inserts, not updates to existing rows. For tables where records are truly append-only (event logs, audit trails, clickstream data), sequence-based watermarks are often the better choice.
For tables that mix inserts and updates, you need a timestamp column — or CDC.
Watermarks are a polling pattern: your pipeline wakes up, asks "what changed since I last ran?", and reads from the source table. CDC is fundamentally different: the database writes every change to a log as it happens, and your pipeline reads from that log.
This distinction matters for several reasons:
Most enterprise databases implement CDC through their transaction log (also called write-ahead log or WAL):
pg_logical or tools like Debezium that decode the WALCHANGE_DATA_CAPTURE feature writes to system change tablesFor this lesson, we'll focus on two approaches you can work with directly in SQL: SQL Server CDC (since it exposes change data via queryable system tables) and Snowflake Streams (since they're entirely SQL-native).
Once CDC is enabled on a SQL Server table, the database automatically maintains a shadow change table. Here's how to enable it:
-- Enable CDC on the database
EXEC sys.sp_cdc_enable_db;
-- Enable CDC on a specific table
EXEC sys.sp_cdc_enable_table
@source_schema = N'dbo',
@source_name = N'orders',
@role_name = NULL,
@supports_net_changes = 1; -- enable net changes (latest state per row)
SQL Server then creates a table like cdc.dbo_orders_CT that looks like this:
| __$start_lsn | __$operation | __$update_mask | order_id | customer_id | order_total | order_status | updated_at |
|---|---|---|---|---|---|---|---|
| 0x0000A1... | 2 | NULL | 10001 | 5042 | 149.99 | 'completed' | 2024-03-15 09:02:11 |
| 0x0000A2... | 4 | 0x06 | 10001 | 5042 | 149.99 | 'refunded' | 2024-03-15 10:15:33 |
| 0x0000A3... | 1 | NULL | 10002 | 7821 | 89.00 | 'pending' | 2024-03-15 10:22:07 |
The __$operation values map to:
1 = Delete2 = Insert3 = Update (before image)4 = Update (after image)To read only the net latest changes (most useful for ETL), SQL Server provides CDC functions:
DECLARE @from_lsn BINARY(10) = sys.fn_cdc_get_min_lsn('dbo_orders');
DECLARE @to_lsn BINARY(10) = sys.fn_cdc_get_max_lsn();
-- Get all changes between two LSN positions
SELECT
__$operation,
order_id,
customer_id,
order_total,
order_status,
updated_at
FROM cdc.fn_cdc_get_all_changes_dbo_orders(
@from_lsn,
@to_lsn,
N'all'
)
ORDER BY __$start_lsn;
-- Get only the net (final) state per row — great for upsert pipelines
SELECT
__$operation,
order_id,
customer_id,
order_total,
order_status,
updated_at
FROM cdc.fn_cdc_get_net_changes_dbo_orders(
@from_lsn,
@to_lsn,
N'all with merge'
);
Store the @to_lsn value in your watermark table after a successful run. On the next run, your @from_lsn starts at the last @to_lsn.
Note: SQL Server CDC change tables have a retention period (default 3 days). If your pipeline doesn't run for longer than the retention window, you'll have a gap and need to fall back to a full reload. Always monitor pipeline lag in production.
Snowflake's Stream feature is arguably the most developer-friendly CDC implementation for data warehousing. A Stream is an object that sits on top of a table and records all DML changes since the last time you consumed it.
-- Create a stream on the orders table
CREATE STREAM orders_stream
ON TABLE raw.orders
SHOW_INITIAL_ROWS = TRUE; -- include existing rows on first read
Reading from a stream looks exactly like querying a table:
SELECT
METADATA$ACTION, -- 'INSERT' or 'DELETE'
METADATA$ISUPDATE, -- TRUE if this row is part of an update operation
METADATA$ROW_ID, -- unique identifier for the changed row
order_id,
customer_id,
order_total,
order_status,
updated_at
FROM orders_stream;
An "update" in Snowflake Streams appears as two rows: a DELETE of the old version (with METADATA$ISUPDATE = TRUE) and an INSERT of the new version (with METADATA$ISUPDATE = TRUE). To process only the final state:
-- Consume the stream inside a MERGE to apply changes to a target table
MERGE INTO analytics.orders AS target
USING (
-- Take only the final state: inserts and update-inserts, ignoring delete-pairs
SELECT *
FROM orders_stream
WHERE METADATA$ACTION = 'INSERT' -- includes both true inserts and update new-rows
) AS source
ON target.order_id = source.order_id
WHEN MATCHED THEN
UPDATE SET
customer_id = source.customer_id,
order_total = source.order_total,
order_status = source.order_status,
updated_at = source.updated_at
WHEN NOT MATCHED THEN
INSERT (order_id, customer_id, order_total, order_status, created_at, updated_at)
VALUES (source.order_id, source.customer_id, source.order_total,
source.order_status, source.created_at, source.updated_at);
Key insight: Snowflake automatically advances the stream's offset when you successfully consume it inside a DML statement. There's no manual watermark update needed — the database tracks it for you. This is one of the most elegant properties of Streams: it's impossible to accidentally advance the watermark before your write succeeds.
Deletes are where most incremental pipelines have a dirty secret: they silently ignore them.
If a record is deleted from the source table, a timestamp watermark query will never see it — the row is gone. You'll end up with ghost records in your destination that no longer exist in the source. Over time, your destination drifts.
The cleanest solution is to convince the upstream application team to implement soft deletes: instead of DELETE FROM orders WHERE order_id = 10001, they do:
UPDATE orders
SET deleted_at = NOW(), is_deleted = TRUE
WHERE order_id = 10001;
Your incremental query picks this up via updated_at, and your destination upsert propagates the is_deleted flag or copies the row to a deleted archive. Simple. Unfortunately, you often can't control upstream application behavior.
Have your source application (or a trigger) write deletes to a separate deleted_records table:
CREATE TABLE deleted_records (
source_table VARCHAR(200) NOT NULL,
record_id BIGINT NOT NULL,
deleted_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
-- Application or trigger writes here before deleting
CREATE OR REPLACE FUNCTION log_deleted_order()
RETURNS TRIGGER AS $$
BEGIN
INSERT INTO deleted_records (source_table, record_id, deleted_at)
VALUES ('orders', OLD.order_id, NOW());
RETURN OLD;
END;
$$ LANGUAGE plpgsql;
CREATE TRIGGER before_order_delete
BEFORE DELETE ON orders
FOR EACH ROW EXECUTE FUNCTION log_deleted_order();
Your pipeline then reads from both the main table (for inserts/updates) and the tombstone table (for deletes):
-- Process deletes in destination
DELETE FROM warehouse.orders
WHERE order_id IN (
SELECT record_id
FROM deleted_records
WHERE source_table = 'orders'
AND deleted_at > :last_watermark
);
For some use cases, the pragmatic answer is to accept that deletes won't propagate in real-time, but run a periodic full reconciliation (say, weekly) that diffs the source and destination and removes orphaned rows:
-- Find records in destination that no longer exist in source
DELETE FROM warehouse.orders
WHERE order_id NOT IN (
SELECT order_id FROM source_db.orders
);
This only works if the full reconciliation is fast enough (smaller tables, or when you can partition the reconciliation). It doesn't work for the 50-million-row case we opened with.
Warning: CDC-based approaches (SQL Server CDC, Snowflake Streams, Debezium) capture hard deletes natively. If your use case requires real-time delete propagation, this is the strongest argument for investing in a proper CDC setup over the watermark pattern.
Real pipelines don't process one table in isolation. You often need to process a cluster of related tables and maintain referential integrity in the destination. Here's a pattern for managing this cleanly.
Use your watermark table to track each source table separately, but process them in dependency order:
-- Watermarks for an e-commerce pipeline
INSERT INTO pipeline_watermarks (pipeline_name, source_table, last_run_at, last_max_value, run_status)
VALUES
('ecommerce_etl', 'customers', '2000-01-01', '2000-01-01', 'success'),
('ecommerce_etl', 'products', '2000-01-01', '2000-01-01', 'success'),
('ecommerce_etl', 'orders', '2000-01-01', '2000-01-01', 'success'),
('ecommerce_etl', 'order_items','2000-01-01','2000-01-01', 'success');
Processing order: customers → products → orders → order_items. Each table's watermark is updated independently. If order_items fails, customers, products, and orders keep their updated watermarks — you only re-process order_items on retry, not the whole pipeline.
This is especially useful when combined with Common Table Expressions (CTEs) for Cleaner SQL to structure complex multi-step extraction queries:
WITH watermarks AS (
SELECT source_table, last_max_value
FROM pipeline_watermarks
WHERE pipeline_name = 'ecommerce_etl'
),
new_orders AS (
SELECT o.*
FROM orders o
JOIN watermarks w ON w.source_table = 'orders'
WHERE o.updated_at > w.last_max_value
),
affected_customers AS (
-- Pull customer records touched by changed orders
SELECT DISTINCT c.*
FROM customers c
JOIN new_orders no ON no.customer_id = c.customer_id
)
SELECT * FROM affected_customers;
An incremental query can still be slow if it's designed carelessly. The key performance lever is indexing the columns you filter on.
Your extraction query filters on updated_at (for timestamp watermarks) or id (for sequence watermarks). Both must be indexed:
-- Essential indexes for watermark-based pipelines
CREATE INDEX CONCURRENTLY idx_orders_updated_at
ON orders(updated_at)
WHERE updated_at IS NOT NULL; -- partial index if NULL rows are common
-- For CDC-like approaches using sequence + timestamp composite
CREATE INDEX CONCURRENTLY idx_orders_seq_ts
ON orders(updated_at, order_id); -- composite for covering queries
If your source is a data warehouse with table partitioning by date, your incremental query should filter on the partition key too:
-- Partition-pruning-aware extraction (Snowflake/BigQuery pattern)
SELECT *
FROM orders
WHERE order_date >= DATE_TRUNC('day', :last_watermark::DATE) -- triggers partition pruning
AND updated_at > :last_watermark
AND updated_at < :upper_bound;
Without the order_date predicate, the query planner might scan all partitions even though updated_at would resolve to a small subset. The redundant partition filter is free to add and potentially very expensive to omit.
Tip: Always use EXPLAIN ANALYZE on your extraction queries against a production-sized dataset before deploying. An incremental query that does a sequential scan on a 50M-row table is worse than a full reload — you're paying the full scan cost but only getting a fraction of the rows.
Let's build a complete incremental pipeline for a realistic scenario: a user_events table that receives ~2 million new rows per day, with occasional updates to correct mislabeled events. You'll wire up the watermark table, write the extraction query, and implement the destination upsert.
-- Source table (imagine this lives in your transactional database)
CREATE TABLE user_events (
event_id BIGINT PRIMARY KEY GENERATED ALWAYS AS IDENTITY,
user_id BIGINT NOT NULL,
event_type VARCHAR(100) NOT NULL,
event_payload JSONB,
event_time TIMESTAMPTZ NOT NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX idx_user_events_updated_at ON user_events(updated_at);
-- Destination table (in your warehouse schema)
CREATE TABLE warehouse.user_events (
event_id BIGINT PRIMARY KEY,
user_id BIGINT NOT NULL,
event_type VARCHAR(100) NOT NULL,
event_payload JSONB,
event_time TIMESTAMPTZ NOT NULL,
created_at TIMESTAMPTZ NOT NULL,
updated_at TIMESTAMPTZ NOT NULL,
_loaded_at TIMESTAMPTZ NOT NULL DEFAULT NOW() -- pipeline metadata
);
-- Watermark record for this pipeline
INSERT INTO pipeline_watermarks (
pipeline_name, source_table, last_run_at, last_max_value, run_status
) VALUES (
'user_events_pipeline', 'user_events', NOW(), '2000-01-01 00:00:00+00', 'success'
);
Using the patterns from Stored Procedures and User-Defined Functions: Building Reusable SQL Logic to encapsulate the pipeline logic:
CREATE OR REPLACE PROCEDURE run_user_events_incremental()
LANGUAGE plpgsql AS $$
DECLARE
v_watermark TIMESTAMPTZ;
v_upper_bound TIMESTAMPTZ;
v_rows_inserted BIGINT;
BEGIN
-- 1. Mark the pipeline as running (detect crashes on next run)
UPDATE pipeline_watermarks
SET run_status = 'running', last_run_at = NOW()
WHERE pipeline_name = 'user_events_pipeline'
AND source_table = 'user_events'
AND run_status = 'success'; -- only start if previous run succeeded
-- If no row was updated, a run is already in progress — bail out
IF NOT FOUND THEN
RAISE NOTICE 'Pipeline already running or in error state. Exiting.';
RETURN;
END IF;
-- 2. Read current watermark
SELECT last_max_value
INTO v_watermark
FROM pipeline_watermarks
WHERE pipeline_name = 'user_events_pipeline'
AND source_table = 'user_events';
-- 3. Set upper bound with safety buffer
v_upper_bound := NOW() - INTERVAL '10 seconds';
RAISE NOTICE 'Extracting events from % to %', v_watermark, v_upper_bound;
-- 4. Extract and upsert in one statement
WITH extracted AS (
SELECT
event_id,
user_id,
event_type,
event_payload,
event_time,
created_at,
updated_at
FROM user_events
WHERE updated_at >= v_watermark -- >= to handle boundary overlaps
AND updated_at < v_upper_bound
)
INSERT INTO warehouse.user_events (
event_id, user_id, event_type, event_payload,
event_time, created_at, updated_at, _loaded_at
)
SELECT
event_id, user_id, event_type, event_payload,
event_time, created_at, updated_at, NOW()
FROM extracted
ON CONFLICT (event_id) DO UPDATE SET
user_id = EXCLUDED.user_id,
event_type = EXCLUDED.event_type,
event_payload = EXCLUDED.event_payload,
event_time = EXCLUDED.event_time,
updated_at = EXCLUDED.updated_at,
_loaded_at = NOW();
GET DIAGNOSTICS v_rows_inserted = ROW_COUNT;
-- 5. Update watermark atomically
UPDATE pipeline_watermarks
SET
last_max_value = v_upper_bound,
records_processed = v_rows_inserted,
run_status = 'success',
updated_at = NOW()
WHERE pipeline_name = 'user_events_pipeline'
AND source_table = 'user_events';
RAISE NOTICE 'Pipeline complete. Processed % rows.', v_rows_inserted;
EXCEPTION WHEN OTHERS THEN
-- Mark as failed so next run doesn't skip processing
UPDATE pipeline_watermarks
SET run_status = 'failed', updated_at = NOW()
WHERE pipeline_name = 'user_events_pipeline'
AND source_table = 'user_events';
RAISE;
END;
$$;
-- Execute the pipeline
CALL run_user_events_incremental();
-- Verify watermark was updated
SELECT pipeline_name, source_table, last_max_value, records_processed, run_status
FROM pipeline_watermarks
WHERE pipeline_name = 'user_events_pipeline';
-- Check destination row count vs source (for the incremental window)
SELECT
'source' AS location,
COUNT(*) AS row_count
FROM user_events
WHERE updated_at < NOW() - INTERVAL '10 seconds'
UNION ALL
SELECT
'destination' AS location,
COUNT(*) AS row_count
FROM warehouse.user_events;
Symptom: Records are missed or duplicated when pipeline run time and source timestamps diverge.
Fix: Always store MAX(updated_at) from the source batch as your watermark, not NOW().
Symptom: Records that were being written during pipeline execution appear partially, causing downstream integrity issues.
Fix: Always use updated_at < NOW() - INTERVAL 'N seconds' as an upper bound. The value of N depends on your database's transaction duration expectations — 5-30 seconds is usually sufficient.
Symptom: Records appear in multiple batches unexpectedly.
Fix: If your database stores timestamps at microsecond precision, ensure your watermark comparison uses the same precision. Type mismatches or implicit casting can cause fence-post errors.
Symptom: Records are consistently missed or duplicated depending on which server's clock is ahead.
Fix: Use the database server's NOW() function — not your application server's clock — for all watermark comparisons. Since both the updated_at column and the upper bound comparison happen on the same database server, clock skew is eliminated.
Symptom: A pipeline that started fast now runs as slowly as the full reload it replaced.
Fix: Check your index health. Over time, updated_at indexes on write-heavy tables can bloat significantly. Also check if you're accidentally disabling index usage through implicit type casts — a common anti-pattern described in Advanced SQL Anti-Patterns: Identifying and Refactoring Common Query Mistakes That Kill Performance at Scale.
Symptom: Destination row count grows monotonically but source row count is flat or declining.
Fix: Implement one of the delete-handling strategies from earlier: soft deletes, tombstone triggers, CDC, or scheduled reconciliation. Audit your destination vs source regularly:
-- Find ghost records in destination that no longer exist in source
SELECT d.event_id
FROM warehouse.user_events d
LEFT JOIN user_events s ON s.event_id = d.event_id
WHERE s.event_id IS NULL;
Symptom: Two pipeline instances launch simultaneously and process the same batch twice, or one reads a stale watermark.
Fix: The run_status = 'running' guard in the stored procedure above handles single-node cases. For distributed environments, use SELECT ... FOR UPDATE SKIP LOCKED to implement advisory locking on the watermark row:
BEGIN;
SELECT last_max_value
FROM pipeline_watermarks
WHERE pipeline_name = 'user_events_pipeline'
AND source_table = 'user_events'
AND run_status = 'success'
FOR UPDATE SKIP LOCKED; -- If another process has the lock, this returns 0 rows
-- If 0 rows returned, exit gracefully
An incremental pipeline has more failure modes than a simple SELECT query, and you need tests for each of them. The lesson on Writing Effective SQL Unit Tests: Validating Query Logic, Edge Cases, and Data Contracts in CI/CD Pipelines covers the testing framework in detail, but here are the essential scenarios specific to incremental pipelines:
-- Test 1: Verify idempotency — running twice produces same destination count
CALL run_user_events_incremental();
SELECT COUNT(*) AS count_after_first_run FROM warehouse.user_events;
CALL run_user_events_incremental(); -- second run should find nothing new
SELECT COUNT(*) AS count_after_second_run FROM warehouse.user_events;
-- counts should be equal
-- Test 2: Verify updates propagate
UPDATE user_events SET event_type = 'purchase_v2' WHERE event_id = 10001;
CALL run_user_events_incremental();
SELECT event_type FROM warehouse.user_events WHERE event_id = 10001;
-- should return 'purchase_v2'
-- Test 3: Verify boundary records are captured
-- Insert a record with updated_at just inside the window
INSERT INTO user_events (user_id, event_type, event_time, updated_at)
VALUES (9999, 'boundary_test', NOW() - INTERVAL '1 minute', NOW() - INTERVAL '30 seconds');
CALL run_user_events_incremental();
SELECT COUNT(*) FROM warehouse.user_events WHERE event_type = 'boundary_test';
-- should return 1
Incremental SQL pipelines are one of the highest-leverage skills in data engineering. Getting them right means the difference between pipelines that scale gracefully and ones that become liabilities the moment your data volume doubles.
Here's the mental model to carry forward:
updated_at or auto-increment columns, when you have write access to a watermark store, and when hard deletes aren't a concernFor next steps, explore Temporal Data Mastery: Writing Queries for Time-Series, Date Ranges, and Slowly Changing Dimensions to handle the SCD Type 2 patterns that often accompany incremental pipelines — tracking not just the current state of a record but its full history. If you're processing high volumes, Query Optimization with Materialized Views: Caching Complex Aggregations and Refreshing Strategies for High-Performance Analytics pairs naturally with incremental ingestion by keeping downstream aggregations fresh without full recomputation.
The patterns in this lesson are production-tested. Implement them with the care they deserve, and your pipelines will be something you're proud to put your name on.