Transforming Data with SQL in Your Pipeline: Filtering, Joining, and Aggregating Before It Reaches Your Warehouse

Introduction

Imagine you're building a pipeline that pulls order data from your e-commerce platform every night. The source system contains every event that ever touched an order: creation, update, cancellation, partial refund, re-shipment, the works. By morning, your analytics team needs a clean, consolidated view — one row per order, with the customer's name attached, only for orders that actually completed, broken down by region. If you dump the raw source data straight into your warehouse and leave all that cleanup for later, you've just handed your analysts a pile of raw materials instead of a finished product. They'll spend half their day massaging data instead of drawing insights from it.

SQL transformations inside your pipeline change this equation. Rather than loading everything and transforming it after the fact, you shape the data during transit — filtering out noise, joining related tables together, and collapsing granular events into meaningful summaries. By the time the data lands in your warehouse, it's ready to use. This approach is at the heart of the ETL (Extract, Transform, Load) pattern, and understanding how to apply it well is one of the most valuable skills a data engineer can develop.

By the end of this lesson, you'll be able to write SQL transformations that do real work inside a pipeline. You'll understand not just the syntax, but the reasoning behind each kind of transformation and when to reach for each one.

What you'll learn:

• Why transforming data before loading it into the warehouse matters
• How to use WHERE clauses to filter irrelevant or low-quality rows
• How to combine data from multiple source tables using JOIN
• How to collapse row-level detail into summaries using GROUP BY and aggregate functions
• How to chain these operations together into a single, production-ready transformation query

---

Prerequisites

You should be comfortable with what a data pipeline is conceptually — if you're not, start with What is a Data Pipeline? Architecture and Core Concepts for Data Engineers before continuing. You should also have a basic familiarity with SQL SELECT statements: selecting columns, the FROM clause, and simple conditions. You don't need to be an expert.

---

Why Transform in the Pipeline Instead of After?

Before we write a single line of SQL, let's establish the "why." There are two broad schools of thought when it comes to where transformation happens, and understanding the trade-off is important.

In an ETL approach, you Extract data from the source, Transform it in-flight (or in a staging area), and then Load only the clean, shaped data into your warehouse. In an ELT approach, you Load raw data into the warehouse first, then transform it there using the warehouse's own compute. Both have legitimate uses — the ETL vs ELT: Choosing the Right Approach article covers the trade-offs in depth.

For this lesson, we're focused on ETL-style transformations: SQL that runs before data reaches its final destination. Here's why this matters in practice:

Volume reduction. If your source system generates 10 million rows per day but only 200,000 of them represent completed orders, loading all 10 million wastes storage and slows down every downstream query. Filtering before load means your warehouse only ever sees what it needs.

Data quality. Joining and aggregating upstream gives you a natural checkpoint. If a join produces zero rows when you expected thousands, something is wrong with the source — and you catch it before bad data contaminates the warehouse.

Analyst experience. Analysts shouldn't need to know the quirks of your operational database schema to do their jobs. A well-designed transformation layer presents a clean, logical model.

---

Setting Up Our Scenario

Throughout this lesson we'll work with a realistic e-commerce scenario. Our source database (a PostgreSQL operational system) has three tables:

• orders — one row per order, with status, timestamps, and customer ID
• customers — customer profile data: name, email, and region
• order_items — one row per line item within an order, with product ID, quantity, and unit price

Our goal: produce a single transformed table called daily_order_summary that contains one row per completed order, with the customer's name and region attached, and the total order value calculated from the line items.

Here's what the raw tables look like:

-- orders table
-- order_id | customer_id | status      | created_at          | updated_at
-- 1001     | 42          | completed   | 2024-03-15 08:22:11 | 2024-03-15 14:05:33
-- 1002     | 17          | cancelled   | 2024-03-15 09:11:45 | 2024-03-15 09:45:22
-- 1003     | 42          | completed   | 2024-03-15 11:33:02 | 2024-03-15 16:20:44

-- customers table
-- customer_id | full_name       | email                   | region
-- 42          | Maria Santos    | maria@example.com       | LATAM
-- 17          | David Chen      | david@example.com       | APAC

-- order_items table
-- item_id | order_id | product_id | quantity | unit_price
-- 5001    | 1001     | 'SKU-A'    | 2        | 29.99
-- 5002    | 1001     | 'SKU-B'    | 1        | 14.99
-- 5003    | 1003     | 'SKU-C'    | 3        | 9.99

We want our output to look like this:

-- order_id | customer_name | region | order_total | order_date
-- 1001     | Maria Santos  | LATAM  | 74.97       | 2024-03-15
-- 1003     | Maria Santos  | LATAM  | 29.97       | 2024-03-15

Notice that order 1002 is gone (cancelled), David Chen's order is gone with it, and the totals are calculated from the line items. Let's build toward this step by step.

---

Step 1: Filtering with WHERE

The WHERE clause is your first line of defense against noise. It tells your query which rows to keep — everything else is discarded before it ever moves through the rest of the pipeline.

Basic filtering

In our scenario, we only want completed orders:

SELECT
    order_id,
    customer_id,
    created_at::DATE AS order_date
FROM orders
WHERE status = 'completed';

This is straightforward, but notice the ::DATE cast. The source system stores created_at as a full timestamp (2024-03-15 08:22:11). For daily reporting, we only need the date portion. Casting during extraction means you don't carry unnecessary precision into the warehouse.

Filtering on time windows

In production pipelines, you almost always want to process only new data rather than re-scanning the entire table every run. This is called incremental loading, and it's one of the most important patterns in data engineering — you can read about it in depth in Incremental Loading Patterns: Timestamps, CDC, and Watermarks.

For now, here's what a time-window filter looks like:

SELECT
    order_id,
    customer_id,
    created_at::DATE AS order_date
FROM orders
WHERE status = 'completed'
  AND created_at >= '2024-03-15 00:00:00'
  AND created_at <  '2024-03-16 00:00:00';

The >= / < pattern (rather than BETWEEN) is intentional. BETWEEN is inclusive on both ends, which can cause rows at exactly midnight to be counted twice across back-to-back pipeline runs. Using strict less-than on the upper bound prevents that.

Filtering out bad data

Filtering isn't only about business logic — it's also your first data quality gate. What if customer_id is NULL? What if created_at is somehow in the future?

SELECT
    order_id,
    customer_id,
    created_at::DATE AS order_date
FROM orders
WHERE status = 'completed'
  AND created_at >= '2024-03-15 00:00:00'
  AND created_at <  '2024-03-16 00:00:00'
  AND customer_id IS NOT NULL
  AND created_at <= NOW();

Each condition you add is a small assertion about data quality. For more systematic approaches to validating data as it flows through a pipeline, see Data Quality: Validation, Testing, and Monitoring Pipelines.

---

Step 2: Joining Tables Together

Now that we have a filtered set of orders, we need to attach the customer's name and region. This is what JOIN is for — combining rows from two or more tables based on a matching key.

Understanding JOIN types

A JOIN in SQL links rows from one table to rows in another where a condition is true — typically where a foreign key in one table matches a primary key in another.

The most common type is INNER JOIN, which returns only rows where a match exists on both sides. If an order has a customer_id that doesn't exist in the customers table, that order is silently dropped from the result.

SELECT
    o.order_id,
    c.full_name   AS customer_name,
    c.region,
    o.created_at::DATE AS order_date
FROM orders o
INNER JOIN customers c ON o.customer_id = c.customer_id
WHERE o.status = 'completed'
  AND o.created_at >= '2024-03-15 00:00:00'
  AND o.created_at <  '2024-03-16 00:00:00';

Notice we're aliasing the tables as o and c to keep things readable when referencing columns with the same name in multiple tables. The ON clause specifies the matching condition.

When to use LEFT JOIN instead

An INNER JOIN silently drops unmatched rows — which is often what you want, but sometimes it hides a data quality problem. If you want to keep orders even when no matching customer is found (so you can audit them), use a LEFT JOIN:

SELECT
    o.order_id,
    c.full_name   AS customer_name,
    c.region,
    o.created_at::DATE AS order_date
FROM orders o
LEFT JOIN customers c ON o.customer_id = c.customer_id
WHERE o.status = 'completed'
  AND o.created_at >= '2024-03-15 00:00:00'
  AND o.created_at <  '2024-03-16 00:00:00';

With a LEFT JOIN, orders without a matching customer will appear with NULL values for customer_name and region. You can then add a monitoring alert for rows where customer_name IS NULL — a downstream signal that your customer data is missing or mismatched.

Joining a third table

Now let's bring in order_items to calculate the total value of each order. We'll join it onto the order using order_id:

SELECT
    o.order_id,
    c.full_name   AS customer_name,
    c.region,
    o.created_at::DATE AS order_date,
    i.quantity,
    i.unit_price
FROM orders o
INNER JOIN customers c   ON o.customer_id = c.customer_id
INNER JOIN order_items i ON o.order_id    = i.order_id
WHERE o.status = 'completed'
  AND o.created_at >= '2024-03-15 00:00:00'
  AND o.created_at <  '2024-03-16 00:00:00';

If we ran this, we'd get one row per line item, not per order:

-- order_id | customer_name | region | order_date | quantity | unit_price
-- 1001     | Maria Santos  | LATAM  | 2024-03-15 | 2        | 29.99
-- 1001     | Maria Santos  | LATAM  | 2024-03-15 | 1        | 14.99
-- 1003     | Maria Santos  | LATAM  | 2024-03-15 | 3        | 9.99

Order 1001 appears twice because it has two line items. This is where aggregation comes in.

---

Step 3: Aggregating with GROUP BY

Aggregation is the process of collapsing multiple rows into a single summary row. In SQL, you do this with GROUP BY combined with aggregate functions like SUM(), COUNT(), AVG(), MIN(), and MAX().

Calculating order totals

Each line item's value is quantity * unit_price. To get the total for an entire order, we sum those products across all line items for that order:

SELECT
    o.order_id,
    c.full_name        AS customer_name,
    c.region,
    o.created_at::DATE AS order_date,
    SUM(i.quantity * i.unit_price) AS order_total
FROM orders o
INNER JOIN customers c   ON o.customer_id = c.customer_id
INNER JOIN order_items i ON o.order_id    = i.order_id
WHERE o.status = 'completed'
  AND o.created_at >= '2024-03-15 00:00:00'
  AND o.created_at <  '2024-03-16 00:00:00'
GROUP BY
    o.order_id,
    c.full_name,
    c.region,
    o.created_at::DATE;

The GROUP BY clause tells the database to collect all rows that have the same combination of order_id, customer_name, region, and order_date — and produce a single output row for each group, with SUM() applied across all the rows in that group.

This gives us exactly what we wanted:

-- order_id | customer_name | region | order_date | order_total
-- 1001     | Maria Santos  | LATAM  | 2024-03-15 | 74.97
-- 1003     | Maria Santos  | LATAM  | 2024-03-15 | 29.97

Filtering groups with HAVING

WHERE filters rows before grouping. Sometimes you want to filter groups after aggregation — for example, to drop orders with a suspiciously low total (which might indicate data quality issues):

SELECT
    o.order_id,
    c.full_name        AS customer_name,
    c.region,
    o.created_at::DATE AS order_date,
    SUM(i.quantity * i.unit_price) AS order_total
FROM orders o
INNER JOIN customers c   ON o.customer_id = c.customer_id
INNER JOIN order_items i ON o.order_id    = i.order_id
WHERE o.status = 'completed'
  AND o.created_at >= '2024-03-15 00:00:00'
  AND o.created_at <  '2024-03-16 00:00:00'
GROUP BY
    o.order_id,
    c.full_name,
    c.region,
    o.created_at::DATE
HAVING SUM(i.quantity * i.unit_price) > 0;

HAVING applies after the GROUP BY is computed, so you can reference aggregate expressions in it. Use WHERE for row-level filters and HAVING for group-level filters.

---

Step 4: Putting It All Together in a Pipeline

In a real pipeline, this SQL doesn't run in a vacuum — it runs as part of an orchestrated workflow. The pattern is:

1. Extract the raw data (or leave it in a staging schema accessible via SQL)
2. Run the transformation query
3. Write the result to the destination table

If you're using Python as your pipeline glue — which is covered in Building Your First Data Pipeline with Python — the structure might look like this:

import psycopg2
import os

# Source (operational DB) and destination (warehouse) connections
src_conn  = psycopg2.connect(os.environ["SOURCE_DB_URL"])
dest_conn = psycopg2.connect(os.environ["WAREHOUSE_DB_URL"])

TRANSFORM_QUERY = """
    SELECT
        o.order_id,
        c.full_name        AS customer_name,
        c.region,
        o.created_at::DATE AS order_date,
        SUM(i.quantity * i.unit_price) AS order_total
    FROM orders o
    INNER JOIN customers c   ON o.customer_id = c.customer_id
    INNER JOIN order_items i ON o.order_id    = i.order_id
    WHERE o.status = 'completed'
      AND o.created_at >= %(window_start)s
      AND o.created_at <  %(window_end)s
    GROUP BY
        o.order_id,
        c.full_name,
        c.region,
        o.created_at::DATE
    HAVING SUM(i.quantity * i.unit_price) > 0
"""

INSERT_QUERY = """
    INSERT INTO daily_order_summary
        (order_id, customer_name, region, order_date, order_total)
    VALUES (%s, %s, %s, %s, %s)
    ON CONFLICT (order_id) DO UPDATE
        SET order_total    = EXCLUDED.order_total,
            customer_name  = EXCLUDED.customer_name
"""

params = {
    "window_start": "2024-03-15 00:00:00",
    "window_end":   "2024-03-16 00:00:00"
}

with src_conn.cursor() as src_cur:
    src_cur.execute(TRANSFORM_QUERY, params)
    rows = src_cur.fetchall()

with dest_conn.cursor() as dest_cur:
    dest_cur.executemany(INSERT_QUERY, rows)
    dest_conn.commit()

src_conn.close()
dest_conn.close()

A few important things happening here:

Parameterized queries. The time window is passed as parameters (%(window_start)s), not string-interpolated. This protects against SQL injection and makes the pipeline easy to run for different date windows.

Upsert pattern. The ON CONFLICT ... DO UPDATE clause handles re-runs gracefully. If this pipeline runs twice for the same day (say, after a failure and retry), it updates existing rows rather than duplicating them. This is the essence of making a pipeline idempotent.

Separation of connections. The transformation query runs against the source, and the insert runs against the destination warehouse. They're independent connections, which matters when your source and destination are different database systems entirely.

---

Hands-On Exercise

Apply what you've learned with this exercise. Use any SQL environment you have available — a local PostgreSQL instance, an online SQL sandbox like db-fiddle.com, or SQLite.

Setup: Create and populate these three tables:

CREATE TABLE orders (
    order_id    INT PRIMARY KEY,
    customer_id INT,
    status      VARCHAR(20),
    created_at  TIMESTAMP
);

CREATE TABLE customers (
    customer_id INT PRIMARY KEY,
    full_name   VARCHAR(100),
    region      VARCHAR(20)
);

CREATE TABLE order_items (
    item_id    INT PRIMARY KEY,
    order_id   INT,
    product_id VARCHAR(20),
    quantity   INT,
    unit_price NUMERIC(10,2)
);

INSERT INTO orders VALUES
    (1001, 42, 'completed', '2024-03-15 08:22:11'),
    (1002, 17, 'cancelled', '2024-03-15 09:11:45'),
    (1003, 42, 'completed', '2024-03-15 11:33:02'),
    (1004, 99, 'completed', '2024-03-15 14:00:00');  -- customer 99 doesn't exist

INSERT INTO customers VALUES
    (42, 'Maria Santos', 'LATAM'),
    (17, 'David Chen',   'APAC');

INSERT INTO order_items VALUES
    (5001, 1001, 'SKU-A', 2, 29.99),
    (5002, 1001, 'SKU-B', 1, 14.99),
    (5003, 1003, 'SKU-C', 3, 9.99),
    (5004, 1004, 'SKU-D', 1, 49.99);

Tasks:

1. Write a query using INNER JOIN that produces the final daily_order_summary output for 2024-03-15. How many rows do you get? Why is order 1004 excluded?

2. Rewrite the query using LEFT JOIN on the customers table. What happens to order 1004 now?

3. Add a COUNT(i.item_id) AS item_count column to your aggregation. Which order has more than one line item?

4. Add a HAVING clause to exclude any orders with fewer than 2 line items. What is your final result set?

---

Common Mistakes & Troubleshooting

"My JOIN produced more rows than the source table." This is called a fan-out and it usually means there are duplicate keys in one of your joined tables. For example, if customers had two rows with customer_id = 42, every order for customer 42 would appear twice. Run SELECT customer_id, COUNT(*) FROM customers GROUP BY customer_id HAVING COUNT(*) > 1 to check for duplicates before trusting a join.

"My GROUP BY query throws an error about ungrouped columns." You've referenced a column in SELECT that's neither in GROUP BY nor inside an aggregate function. Add it to the GROUP BY, wrap it in MAX() or MIN() if the value is the same for all rows in a group, or remove it from the SELECT.

"My SUM is way too large — it looks like numbers are being multiplied." You have a many-to-many join producing duplicate rows before the aggregation happens. Try running the query without the GROUP BY first to inspect the raw joined rows — you'll likely see the same order_items row appearing multiple times.

"I filtered by date but the pipeline is running much slower than expected." Check whether the created_at column has an index. A full table scan on a 500-million-row orders table is a very different thing from an indexed range scan. If you can't add an index (read-only source), consider using a staging layer.

"The same order appears twice in the destination after re-running the pipeline." Your load step isn't idempotent — you're doing a plain INSERT rather than an upsert. Add ON CONFLICT DO UPDATE (Postgres), MERGE (SQL Server, BigQuery), or INSERT IGNORE / REPLACE (MySQL) depending on your destination system.

---

Summary & Next Steps

In this lesson you've learned how to do the three most important transformation operations in SQL — filter with WHERE, combine with JOIN, and summarize with GROUP BY — and how to wire them together into a single query that shapes data before it reaches your warehouse.

The key ideas to carry forward:

• Filter early. Every row you drop before loading is compute, storage, and latency saved.
• Choose your join type deliberately. INNER JOIN for strict matches; LEFT JOIN when you need to audit missing data.
• Aggregate to the right grain. Know what level of detail your consumers actually need and don't load more granularity than that.
• Make your load step idempotent. Use upserts so re-runs don't create duplicates.

From here, there are several natural directions to deepen your skills. If you're interested in how to orchestrate and schedule these pipeline runs reliably, Scheduling and Orchestrating Pipelines with Airflow is a natural next stop. If you want to handle the case where your source schema changes without warning — a very common real-world problem — read Schema Evolution Strategies for Production Data Pipelines. And if the data you're connecting to lives in cloud storage or external databases, Connecting to Databases, Flat Files, and Cloud Storage for Data Pipeline Ingestion will show you how to get it into the pipeline in the first place.