Learn how SQL powers reliable data pipelines, historical tracking, event analytics, data quality, and performance tuning. This chapter focuses on practical patterns Data Engineers use to transform raw data into trusted, analytics-ready datasets.
1. Incremental Data Processing
Incremental processing processes only new or changed records rather than rebuilding an entire dataset on every run. It can reduce compute cost and processing time when the source provides a reliable change marker.
Example: Read changes since the last successful run
Assume source_orders has an updated_at timestamp.
SELECT order_id, customer_id, order_status, amount, updated_at
FROM source_orders
WHERE updated_at > :last_successful_watermark
AND updated_at <= :current_run_cutoff;The parameters represent the previous successful watermark and a fixed cutoff for this run. Persist the new watermark only after the target write succeeds.
Common incremental strategies
| Strategy | How it works | Best suited for |
|---|---|---|
| Timestamp watermark | Read rows changed after the last timestamp. | Sources with dependable update timestamps. |
| Increasing ID | Read IDs greater than the previous maximum. | Append-only sources with monotonic IDs. |
| CDC stream or log | Consume inserts, updates, and deletes from change events. | Low-latency pipelines and delete capture. |
| Partition-based | Reprocess affected date or key partitions. | Large partitioned analytical tables. |
Idempotent incremental loads with MERGE
An idempotent load can be safely rerun without creating duplicate target rows. MERGE syntax varies by warehouse; this illustrates the pattern.
MERGE INTO target_orders AS t
USING staged_orders AS s
ON t.order_id = s.order_id
WHEN MATCHED THEN UPDATE SET
customer_id = s.customer_id,
order_status = s.order_status,
amount = s.amount,
updated_at = s.updated_at
WHEN NOT MATCHED THEN INSERT
(order_id, customer_id, order_status, amount, updated_at)
VALUES
(s.order_id, s.customer_id, s.order_status, s.amount, s.updated_at);2. CDC and Change Detection
Change Data Capture (CDC) identifies source changes and makes them available downstream. A CDC event often includes a business key, operation type (insert, update, delete), source sequence, and event timestamp.
Select the latest change per key
WITH ranked_changes AS (
SELECT order_id, order_status, amount, operation, change_sequence,
ROW_NUMBER() OVER (
PARTITION BY order_id
ORDER BY change_sequence DESC
) AS rn
FROM order_changes
)
SELECT order_id, order_status, amount, operation
FROM ranked_changes
WHERE rn = 1;This selects the latest event per order. Apply inserts and updates as upserts; process delete events according to the target's data contract.
Detect changes by comparing snapshots
SELECT s.customer_id,
CASE
WHEN t.customer_id IS NULL THEN 'INSERT'
WHEN s.customer_name <> t.customer_name
OR s.city <> t.city THEN 'UPDATE'
ELSE 'UNCHANGED'
END AS change_type
FROM current_customers AS s
LEFT JOIN previous_customers AS t
ON s.customer_id = t.customer_id;NULL <> 'A' evaluates to unknown, not true. Use your database's NULL-safe comparison operator or explicit NULL logic. A left join from the current snapshot also won't detect rows deleted from the source; use a full comparison or anti-join for that.3. SCD Type 1 and Type 2
Slowly Changing Dimensions (SCD) describe how a warehouse tracks changes to descriptive attributes, such as a customer's city or a product's category.
| Type | Behavior | Use case |
|---|---|---|
| SCD Type 1 | Overwrite the old value; no attribute history retained. | Correcting a typo or keeping only the latest email. |
| SCD Type 2 | Insert a new version and retain the previous version. | Historical reporting, such as a customer's city at purchase time. |
SCD Type 1: Overwrite changed attributes
MERGE INTO dim_customer AS t
USING staged_customer AS s
ON t.customer_id = s.customer_id
WHEN MATCHED THEN UPDATE SET
customer_name = s.customer_name,
city = s.city
WHEN NOT MATCHED THEN INSERT (customer_id, customer_name, city)
VALUES (s.customer_id, s.customer_name, s.city);This keeps one current row per customer and replaces the tracked attributes.
SCD Type 2: Preserve history
A typical Type 2 dimension includes a surrogate key, business key, tracked attributes, effective_from, effective_to, and is_current. When an attribute changes, expire the old version and insert a new current version.
-- Step 1: Expire current rows with changed tracked attributes.
UPDATE dim_customer AS d
SET effective_to = :change_timestamp,
is_current = FALSE
FROM staged_customer AS s
WHERE d.customer_id = s.customer_id
AND d.is_current = TRUE
AND (d.customer_name <> s.customer_name
OR d.city <> s.city);
-- Step 2: Insert new customers and new versions.
INSERT INTO dim_customer (
customer_id, customer_name, city,
effective_from, effective_to, is_current
)
SELECT s.customer_id, s.customer_name, s.city,
:change_timestamp, NULL, TRUE
FROM staged_customer AS s
LEFT JOIN dim_customer AS d
ON d.customer_id = s.customer_id AND d.is_current = TRUE
WHERE d.customer_id IS NULL
OR (d.customer_name <> s.customer_name
OR d.city <> s.city);4. Data Quality Checks
Data quality checks detect invalid, incomplete, inconsistent, or unexpected data before it reaches downstream consumers.
Required fields must not be NULL
SELECT COUNT(*) AS invalid_rows
FROM orders
WHERE order_id IS NULL
OR customer_id IS NULL
OR order_date IS NULL;Check uniqueness of a business key
SELECT order_id, COUNT(*) AS row_count
FROM orders
GROUP BY order_id
HAVING COUNT(*) > 1;Check valid ranges and accepted values
SELECT *
FROM orders
WHERE amount < 0
OR order_status NOT IN ('NEW', 'PAID', 'SHIPPED', 'CANCELLED');If order_status can be NULL, validate it separately because NOT IN does not evaluate to true for NULL values.
Referential integrity
SELECT o.order_id, o.customer_id
FROM orders AS o
LEFT JOIN customers AS c ON o.customer_id = c.customer_id
WHERE c.customer_id IS NULL;Reconcile source and target counts
SELECT
(SELECT COUNT(*) FROM source_orders) AS source_count,
(SELECT COUNT(*) FROM target_orders) AS target_count;Matching counts do not guarantee matching data. Also compare counts by partition, business keys, aggregates, or hashes where appropriate.
5. Sessionization and Event Analytics
Sessionization groups events into sessions. A common rule starts a new session when the gap between consecutive events for the same user exceeds 30 minutes. The threshold is a business definition, not a universal constant.
Assume an events table with user_id, event_id, and event_ts.
WITH ordered_events AS (
SELECT user_id, event_id, event_ts,
LAG(event_ts) OVER (
PARTITION BY user_id ORDER BY event_ts, event_id
) AS previous_event_ts
FROM events
),
session_starts AS (
SELECT *,
CASE
WHEN previous_event_ts IS NULL THEN 1
WHEN event_ts > previous_event_ts + INTERVAL '30' MINUTE THEN 1
ELSE 0
END AS is_new_session
FROM ordered_events
),
session_numbers AS (
SELECT *,
SUM(is_new_session) OVER (
PARTITION BY user_id
ORDER BY event_ts, event_id
ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
) AS session_number
FROM session_starts
)
SELECT user_id, session_number,
MIN(event_ts) AS session_start,
MAX(event_ts) AS session_end,
COUNT(*) AS event_count
FROM session_numbers
GROUP BY user_id, session_number
ORDER BY user_id, session_start;Interval arithmetic differs by SQL engine. BigQuery commonly uses TIMESTAMP_DIFF; other engines may use different interval syntax. Use a deterministic tie-breaker when event timestamps can match.
Useful event analytics
- Daily active users: count distinct users by date.
- Conversion funnel: measure users progressing from one event type to another.
- Time to conversion: compare first-touch and conversion timestamps.
- Retention: measure whether users return in later time windows.
6. Identifying Missing or Duplicate Records
Find duplicate keys
SELECT customer_id, COUNT(*) AS row_count
FROM customer_events
GROUP BY customer_id
HAVING COUNT(*) > 1;Choose grouping columns based on the intended uniqueness rule. An event table may be unique by event_id, not by customer_id.
Keep one record per key with ROW_NUMBER
WITH ranked AS (
SELECT *,
ROW_NUMBER() OVER (
PARTITION BY event_id
ORDER BY ingested_at DESC, source_sequence DESC
) AS rn
FROM customer_events
)
SELECT *
FROM ranked
WHERE rn = 1;This keeps the latest record according to the chosen ordering. Make the ordering deterministic so reruns don't choose different survivors.
Find records missing from either table
-- Source records missing from the target
SELECT s.order_id
FROM source_orders AS s
LEFT JOIN target_orders AS t ON t.order_id = s.order_id
WHERE t.order_id IS NULL;
-- Target records absent from the source snapshot
SELECT t.order_id
FROM target_orders AS t
LEFT JOIN source_orders AS s ON s.order_id = t.order_id
WHERE s.order_id IS NULL;The second query is meaningful only when the source is a complete comparable snapshot. A record absent from today's incremental batch has not necessarily been deleted.
Find gaps in a numeric sequence
WITH numbered AS (
SELECT order_id,
LAG(order_id) OVER (ORDER BY order_id) AS previous_id
FROM orders
)
SELECT previous_id + 1 AS gap_start,
order_id - 1 AS gap_end
FROM numbered
WHERE order_id > previous_id + 1;This is useful only when gaps in the numeric sequence have business meaning. Gaps in generated IDs do not automatically indicate missing business records.
7. Query Optimization and Execution Plans
Query optimization aims to reduce scanned data, data movement, memory pressure, and unnecessary computation while preserving results.
Practical optimization checklist
- Read less data: Select only required columns rather than using
SELECT *in production transformations. - Filter early: Apply selective filters as close to the source as practical while preserving query semantics.
- Partition pruning: Filter on partition columns so the engine can skip irrelevant partitions.
- Use clustering or indexes appropriately: Follow platform guidance and access patterns.
- Check join cardinality: Unexpected many-to-many joins can multiply rows and inflate cost.
- Avoid repeated work: Reuse expensive intermediate results when useful, while accounting for storage and maintenance overhead.
- Keep statistics current: Optimizers may use statistics to estimate row counts and choose join strategies.
- Measure: Compare bytes scanned, elapsed time, shuffle/spill, rows read, and cost before and after.
Example: Filter a partitioned table
SELECT customer_id, SUM(amount) AS total_amount
FROM fact_orders
WHERE order_date >= DATE '2026-01-01'
AND order_date < DATE '2026-02-01'
GROUP BY customer_id;If order_date is the partition column, this bounded date filter may allow partition pruning. Behavior depends on the warehouse, table layout, and data types.
Inspect an execution plan
EXPLAIN
SELECT c.country, COUNT(*) AS order_count
FROM orders AS o
JOIN customers AS c ON o.customer_id = c.customer_id
WHERE o.order_date >= DATE '2026-01-01'
GROUP BY c.country;Execution-plan syntax and detail differ by database. Look for large scans, expensive sorts, repeated scans, skewed joins, poor row-count estimates, spills, and unnecessary redistribution. Some engines provide EXPLAIN ANALYZE, which executes the query and reports actual runtime statistics.
Interview Questions to Practice
1. What is the difference between full refresh and incremental processing?
A full refresh rebuilds the target from all relevant source data. Incremental processing handles only new or changed data, often reducing runtime and cost. It requires reliable change detection, safe retries, delete handling, and a strategy for late-arriving records.
2. How would you make an incremental pipeline idempotent?
Use a stable business key, deterministic deduplication, and an upsert or merge. Make the processing window repeatable, commit the watermark only after a successful target write, and ensure retries cannot create duplicate target rows.
3. When would you choose SCD Type 1 versus Type 2?
Use Type 1 when only the current value matters. Use Type 2 when historical versions must be retained for point-in-time analysis. Define tracked attributes, effective-time rules, and how to identify the current row.
4. How do you validate a data pipeline?
Combine schema and null checks, uniqueness rules, accepted-value checks, referential integrity, source-to-target reconciliation, freshness checks, and business-specific assertions. Define what happens when each check fails.
5. How would you identify duplicates without deleting valid records?
First define the business uniqueness key. Rank candidate rows with ROW_NUMBER and a deterministic ordering, inspect duplicate groups, and then apply a documented survivor rule or quarantine policy.
6. How do you optimize a slow warehouse query?
Start with the execution plan and workload metrics. Check scan volume, partition pruning, join cardinality, shuffles, sorts, spills, statistics, and repeated computation. Change one thing at a time and compare results and cost.
Chapter Summary
- Incremental processing reduces repeated work but needs reliable watermarks and retry-safe writes.
- CDC captures row-level changes; snapshot comparison can also detect changes.
- SCD Type 1 overwrites attributes, while Type 2 preserves historical versions.
- Data quality checks should be explicit, measurable, and connected to an action.
- Window functions such as LAG, ROW_NUMBER, and cumulative SUM are useful for event analytics and deduplication.
- Execution plans and runtime metrics should guide optimization decisions.
Note: SQL syntax differs across PostgreSQL, BigQuery, Snowflake, SQL Server, and other engines. Adapt interval arithmetic, NULL-safe comparisons, MERGE behavior, and execution-plan commands to your platform.
Comments
Post a Comment