SQL for Data Engineers: The Complete Guide to Building Data Pipelines (2026)

Quick Navigation

• What Makes Data Engineering SQL Different
• How Data Pipelines Use SQL
• The 7 SQL Skills Every Data Engineer Needs
• ETL vs ELT: Where SQL Fits
• SQL in the Modern Data Stack
• SQL Interview Questions for Data Engineers
• 8-Week Learning Roadmap
• Common Pipeline Mistakes to Avoid
• FAQ

Example: Multi-Stage Pipeline

WITH raw_cases AS (
  SELECT 
    case_id,
    detective_id,
    reported_date,
    case_status,
    priority_level
  FROM staging.incoming_cases
  WHERE reported_date >= CURRENT_DATE - INTERVAL '7 days'
),

validated_cases AS (
  SELECT *
  FROM raw_cases
  WHERE case_id IS NOT NULL
    AND detective_id IS NOT NULL
    AND case_status IN ('open', 'investigating', 'closed')
),

enriched_cases AS (
  SELECT 
    c.*,
    d.detective_name,
    d.department,
    COUNT(e.evidence_id) AS evidence_count
  FROM validated_cases c
  LEFT JOIN dim_detectives d ON c.detective_id = d.detective_id
  LEFT JOIN fact_evidence e ON c.case_id = e.case_id
  GROUP BY c.case_id, c.detective_id, c.reported_date, 
           c.case_status, c.priority_level,
           d.detective_name, d.department
)

SELECT * FROM enriched_cases;

Each CTE represents one transformation step: extract, validate, enrich. When the pipeline fails, query each CTE individually to find where the problem occurred.

Example: Tracking Case Status Changes

SELECT 
  case_id,
  status_timestamp,
  case_status,
  LAG(case_status) OVER (
    PARTITION BY case_id ORDER BY status_timestamp
  ) AS previous_status,
  LEAD(case_status) OVER (
    PARTITION BY case_id ORDER BY status_timestamp
  ) AS next_status,
  ROW_NUMBER() OVER (
    PARTITION BY case_id ORDER BY status_timestamp DESC
  ) AS recency_rank
FROM case_status_history
WHERE case_id IN (SELECT case_id FROM active_investigations);

LAG and LEAD access previous and next rows. ROW_NUMBER identifies the most recent record. All without GROUP BY.

Example: Removing Duplicate Evidence Records

WITH ranked_evidence AS (
  SELECT 
    *,
    ROW_NUMBER() OVER (
      PARTITION BY evidence_id 
      ORDER BY collected_timestamp DESC, 
               ingestion_timestamp DESC
    ) AS row_num
  FROM staging.evidence_raw
)

INSERT INTO warehouse.fact_evidence
SELECT 
  evidence_id,
  case_id,
  evidence_type,
  collected_timestamp,
  chain_of_custody
FROM ranked_evidence
WHERE row_num = 1;

ROW_NUMBER() with PARTITION BY identifies duplicates. We keep the most recently collected record. This pattern works for any deduplication scenario.

Example: Incremental Suspect Updates

MERGE INTO warehouse.dim_suspects target
USING (
  SELECT 
    suspect_id,
    suspect_name,
    known_aliases,
    last_seen_location,
    updated_at
  FROM staging.suspects_incremental
  WHERE updated_at > (
    SELECT MAX(last_updated) FROM warehouse.dim_suspects
  )
) source
ON target.suspect_id = source.suspect_id
WHEN MATCHED THEN 
  UPDATE SET
    suspect_name = source.suspect_name,
    known_aliases = source.known_aliases,
    last_seen_location = source.last_seen_location,
    last_updated = source.updated_at
WHEN NOT MATCHED THEN
  INSERT (suspect_id, suspect_name, known_aliases,
          last_seen_location, last_updated)
  VALUES (source.suspect_id, source.suspect_name,
          source.known_aliases, source.last_seen_location,
          source.updated_at);

MERGE handles both inserts and updates in one statement. The WHERE clause filters to only recent changes, processing a fraction of total rows.

Example: Quality Validation Before Loading

WITH quality_checks AS (
  SELECT 
    'null_case_ids' AS check_name,
    COUNT(*) AS failure_count
  FROM staging.new_cases
  WHERE case_id IS NULL
  
  UNION ALL
  
  SELECT 
    'future_dates' AS check_name,
    COUNT(*) AS failure_count
  FROM staging.new_cases
  WHERE reported_date > CURRENT_DATE
  
  UNION ALL
  
  SELECT 
    'invalid_status' AS check_name,
    COUNT(*) AS failure_count
  FROM staging.new_cases
  WHERE case_status NOT IN (
    'open', 'investigating', 'closed', 'cold'
  )
)

SELECT * FROM quality_checks WHERE failure_count > 0;

Run this before your main pipeline. If any check returns rows, halt the pipeline and alert. Prevention beats cleanup every time.

Example: Type 2 SCD Implementation

-- Step 1: Expire changed records
UPDATE warehouse.dim_detectives
SET 
  valid_to = CURRENT_DATE - INTERVAL '1 day',
  is_current = FALSE
WHERE detective_id IN (
  SELECT detective_id 
  FROM staging.detective_updates
)
AND is_current = TRUE;

-- Step 2: Insert new versions
INSERT INTO warehouse.dim_detectives (
  detective_id, detective_name, department,
  rank, valid_from, valid_to, is_current
)
SELECT 
  detective_id,
  detective_name,
  department,
  rank,
  CURRENT_DATE AS valid_from,
  '2099-12-31'::DATE AS valid_to,
  TRUE AS is_current
FROM staging.detective_updates;

Type 2 SCD maintains full history. Each change creates a new row with valid_from and valid_to dates. The is_current flag simplifies queries for the latest state.

Example: Analyzing Query Performance

-- Check execution plan before running
EXPLAIN ANALYZE
SELECT 
  c.case_id,
  c.case_status,
  COUNT(e.evidence_id) AS evidence_count
FROM cases c
LEFT JOIN evidence e ON c.case_id = e.case_id
WHERE c.reported_date >= '2024-01-01'
GROUP BY c.case_id, c.case_status;

-- Add index to improve join performance
CREATE INDEX idx_evidence_case_id
  ON evidence(case_id);

-- Partition large tables by date
CREATE TABLE cases_partitioned (
  case_id BIGINT,
  reported_date DATE,
  case_status VARCHAR(50)
) PARTITION BY RANGE (reported_date);

EXPLAIN ANALYZE shows where your query spends time. Add indexes on JOIN and WHERE columns. Partition tables by date for faster scans.

Example: dbt-Style SQL Model

-- models/mart_case_summary.sql
-- This runs inside your warehouse as an ELT transformation

WITH closed_cases AS (
  SELECT
    case_id,
    detective_id,
    reported_date,
    closed_date,
    closed_date - reported_date AS days_to_close
  FROM {{ ref('stg_cases') }}
  WHERE case_status = 'closed'
),

detective_stats AS (
  SELECT
    detective_id,
    COUNT(*) AS cases_closed,
    AVG(days_to_close) AS avg_days_to_close,
    MIN(days_to_close) AS fastest_close
  FROM closed_cases
  GROUP BY detective_id
)

SELECT
  d.detective_name,
  d.department,
  s.cases_closed,
  s.avg_days_to_close,
  s.fastest_close
FROM detective_stats s
JOIN {{ ref('dim_detectives') }} d
  ON s.detective_id = d.detective_id;

In ELT, this SQL runs inside the warehouse. dbt manages dependencies between models, runs tests, and generates documentation automatically.

Problem:

A table has duplicate rows based on user_id. Keep only the most recent record for each user.

WITH latest_events AS (
  SELECT 
    *,
    ROW_NUMBER() OVER (
      PARTITION BY user_id 
      ORDER BY event_timestamp DESC
    ) AS rn
  FROM user_events
)
DELETE FROM user_events
WHERE (user_id, event_timestamp) NOT IN (
  SELECT user_id, event_timestamp
  FROM latest_events
  WHERE rn = 1
);

Problem:

Find missing case_ids in a sequence (e.g., 1, 2, 4, 5, 8 has gaps at 3, 6, 7).

WITH case_sequence AS (
  SELECT 
    case_id,
    LEAD(case_id) OVER (ORDER BY case_id) AS next_case_id
  FROM cases
)
SELECT 
  case_id + 1 AS gap_start,
  next_case_id - 1 AS gap_end
FROM case_sequence
WHERE next_case_id - case_id > 1;

SELECT 
  case_date,
  cases_opened,
  SUM(cases_opened) OVER (
    ORDER BY case_date
  ) AS cumulative_cases,
  AVG(cases_opened) OVER (
    ORDER BY case_date 
    ROWS BETWEEN 6 PRECEDING AND CURRENT ROW
  ) AS seven_day_avg
FROM daily_case_stats
ORDER BY case_date;

Problem:

Write a MERGE that can run multiple times safely without creating duplicates.

MERGE INTO warehouse.suspects target
USING staging.suspect_updates source
ON target.suspect_id = source.suspect_id 
  AND target.is_current = TRUE
WHEN MATCHED 
  AND source.updated_at > target.last_updated THEN
  UPDATE SET
    suspect_name = source.suspect_name,
    last_updated = source.updated_at
WHEN NOT MATCHED THEN
  INSERT (suspect_id, suspect_name,
          last_updated, is_current)
  VALUES (source.suspect_id, source.suspect_name,
          source.updated_at, TRUE);

The additional condition checking updated_at prevents re-processing old data if the pipeline runs twice.

Problem:

A detective changes departments. Update the dimension table while preserving history.

-- Expire the old record
UPDATE dim_detectives
SET valid_to = CURRENT_DATE, is_current = FALSE
WHERE detective_id = 1042 AND is_current = TRUE;

-- Insert the new version
INSERT INTO dim_detectives (
  detective_id, detective_name, department, 
  valid_from, valid_to, is_current
)
VALUES (
  1042, 'Sarah Chen', 'Homicide',
  CURRENT_DATE, '2099-12-31', TRUE
);

1. Using SELECT * in Production

SELECT * makes your pipeline fragile. When someone adds a column to the source table, your pipeline breaks or pulls sensitive data you didn't intend to process.

Fix: Always specify exact columns.

2. Not Handling NULLs Properly

NULL != NULL in SQL. Comparisons with NULL require IS NULL or IS NOT NULL, not = or !=. Missing this breaks JOIN conditions and WHERE filters silently.

Fix: Use COALESCE() for defaults and IS NULL for comparisons.

3. Missing Indexes on Large Tables

Joining or filtering large tables without indexes forces full table scans. A query that should take seconds runs for hours.

Fix: Create indexes on JOIN keys and frequently filtered columns.

4. Writing Non-Idempotent Queries

If your pipeline runs twice (network retry, manual rerun, scheduler glitch), will it create duplicate data? Non-idempotent SQL is a production disaster waiting to happen.

Fix: Use MERGE instead of INSERT. Add timestamp checks. Design for reruns.

5. Ignoring EXPLAIN Plans

Running queries blind without checking execution plans means you have no idea if your query is doing full table scans, inefficient joins, or missing obvious optimizations.

Fix: Always run EXPLAIN before deploying expensive queries to production.

Is SQL enough for data engineering?

SQL is essential but not sufficient on its own. Data engineers also need Python (or Scala), orchestration tools like Airflow or Prefect, cloud platforms (AWS, GCP, Azure), and infrastructure skills (Docker, Kubernetes). That said, SQL is the foundation. You will write more SQL than any other language in a typical data engineering role.

Do data engineers need to know Python and SQL?

Yes. SQL handles data transformations inside the warehouse. Python handles orchestration, API calls, custom business logic, and anything SQL cannot express. Modern data engineers are fluent in both. Start with SQL, then add Python as you hit problems SQL cannot solve.

Is SQL still relevant for data engineering in 2026?

Absolutely. SQL has been the standard for 50 years and is not going anywhere. Every modern data warehouse (Snowflake, BigQuery, Redshift, Databricks) uses SQL as its primary interface. Tools like dbt are doubling down on SQL. The syntax evolves, but the core concepts remain fundamental.

What is the best way to practice SQL for data engineering?

Practice on realistic, large datasets using production patterns. Build an end-to-end pipeline: load data incrementally, deduplicate, track history with SCD, add quality checks, and optimize performance. Focus on pipeline scenarios rather than syntax puzzles.

Will AI replace SQL for data engineers?

AI can generate SQL queries, but it cannot design pipeline architectures, ensure idempotency across complex workflows, or handle the edge cases that break production systems. Think of it like calculators and math: calculators did not replace the need to understand mathematics. AI is a productivity tool, not a replacement for understanding how data pipelines work.