M61 · Analytical SQL & DuckDB

How a query is evaluated

You write SQL in the order SELECT … FROM … WHERE … GROUP BY … HAVING … ORDER BY … LIMIT, but the database evaluates it (logically) in a different order. Knowing that order explains most "why can't I use that alias here?" errors:

FROM and JOIN: build the input rows.
WHERE: filter individual rows (no aggregates yet, no SELECT aliases).
GROUP BY: collapse rows into groups.
HAVING: filter groups (aggregates allowed).
Window functions are computed here, over the rows that survived.
SELECT: compute output columns and aliases.
DISTINCT, then ORDER BY (aliases allowed), then LIMIT/OFFSET.

Throughout this module we use a small shop dataset. Run the cell to create it and look at it.

import sqlite3

def make_shop():
    conn = sqlite3.connect(":memory:")
    conn.executescript("""
    CREATE TABLE customers (id INTEGER PRIMARY KEY, name TEXT, region TEXT);
    CREATE TABLE orders (id INTEGER PRIMARY KEY, customer_id INTEGER, order_date TEXT, amount REAL, status TEXT);
    INSERT INTO customers VALUES (1,'Ada','north'), (2,'Ben','south'), (3,'Cy','north'), (4,'Di','south');
    INSERT INTO orders VALUES
      (1,1,'2026-03-01',120,'paid'), (2,2,'2026-03-01', 80,'paid'), (3,1,'2026-03-02', 40,'refunded'),
      (4,3,'2026-03-02',200,'paid'), (5,2,'2026-03-03', 60,'paid'), (6,4,'2026-03-03', 90,'paid'),
      (7,1,'2026-03-04', 75,'paid'), (8,3,'2026-03-05', 30,'paid'), (9,4,'2026-03-05',110,'refunded'),
      (10,2,'2026-03-06',150,'paid');
    """)
    return conn

conn = make_shop()
print("sqlite", sqlite3.sqlite_version)
for row in conn.execute("SELECT * FROM orders LIMIT 4"):
    print(row)
for row in conn.execute("""
    SELECT c.region, COUNT(*) AS n, SUM(o.amount) AS revenue
    FROM orders o JOIN customers c ON c.id = o.customer_id
    WHERE o.status = 'paid'
    GROUP BY c.region
    HAVING SUM(o.amount) > 100
    ORDER BY revenue DESC"""):
    print(row)

Explanation

ORDER BY runs after SELECT, so it can use the alias. WHERE runs before grouping, so an aggregate there is an error ("misuse of aggregate"); that filter belongs in HAVING. (SQLite is lenient and also lets WHERE use a SELECT alias, but standard SQL and most engines do not, so don't rely on it.)

Remember: FROM → WHERE → GROUP BY → HAVING → windows → SELECT → ORDER BY → LIMIT.

Common table expressions

A CTE (WITH name AS (…)) names a subquery so later parts of the query can use it like a table. CTEs turn a nested, inside-out query into a top-to-bottom story of small steps, each with a name that says what it is. That is their real value: readability and testability, not speed.

import sqlite3
conn = sqlite3.connect(":memory:")
conn.executescript("""
CREATE TABLE customers (id INTEGER PRIMARY KEY, name TEXT, region TEXT);
CREATE TABLE orders (id INTEGER PRIMARY KEY, customer_id INTEGER, order_date TEXT, amount REAL, status TEXT);
INSERT INTO customers VALUES (1,'Ada','north'), (2,'Ben','south'), (3,'Cy','north'), (4,'Di','south');
INSERT INTO orders VALUES
  (1,1,'2026-03-01',120,'paid'), (2,2,'2026-03-01', 80,'paid'), (3,1,'2026-03-02', 40,'refunded'),
  (4,3,'2026-03-02',200,'paid'), (5,2,'2026-03-03', 60,'paid'), (6,4,'2026-03-03', 90,'paid'),
  (7,1,'2026-03-04', 75,'paid'), (8,3,'2026-03-05', 30,'paid'), (9,4,'2026-03-05',110,'refunded'),
  (10,2,'2026-03-06',150,'paid');
""")
query = """
WITH paid AS (                      -- step 1: only real revenue
    SELECT * FROM orders WHERE status = 'paid'
),
customer_revenue AS (               -- step 2: one row per customer
    SELECT c.id, c.name, c.region, SUM(p.amount) AS revenue
    FROM paid p JOIN customers c ON c.id = p.customer_id
    GROUP BY c.id
),
region_avg AS (                     -- step 3: one row per region
    SELECT region, AVG(revenue) AS avg_revenue FROM customer_revenue GROUP BY region
)
SELECT cr.name, cr.region, cr.revenue, ROUND(ra.avg_revenue, 1) AS region_avg
FROM customer_revenue cr JOIN region_avg ra USING (region)
WHERE cr.revenue > ra.avg_revenue
ORDER BY cr.revenue DESC
"""
for row in conn.execute(query):
    print(row)

Recursive CTEs: walking a hierarchy

A recursive CTE refers to itself. It has an anchor member (the starting rows) and a recursive member joined to the CTE's previous output, combined with UNION ALL. The database repeats the recursive step until it produces no new rows. Classic uses: org charts, bill-of-materials, category trees, and generating series of dates.

flowchart TD
  A[1 Grace, CEO] --> B[2 Alan, CTO]
  A --> C[3 Ada, CFO]
  B --> D[4 Linus, Eng Manager]
  B --> E[5 Barbara, Architect]
  D --> F[6 Guido, Engineer]
  D --> G[7 Margaret, Engineer]
  C --> H[8 Katherine, Analyst]

import sqlite3
conn = sqlite3.connect(":memory:")
conn.executescript("""
CREATE TABLE employees (id INTEGER PRIMARY KEY, name TEXT, title TEXT, manager_id INTEGER);
INSERT INTO employees VALUES
 (1,'Grace','CEO',NULL), (2,'Alan','CTO',1), (3,'Ada','CFO',1), (4,'Linus','Eng Manager',2),
 (5,'Barbara','Architect',2), (6,'Guido','Engineer',4), (7,'Margaret','Engineer',4), (8,'Katherine','Analyst',3);
""")
query = """
WITH RECURSIVE chain(id, name, depth, path) AS (
    SELECT id, name, 0, name FROM employees WHERE manager_id IS NULL     -- anchor: the root
    UNION ALL
    SELECT e.id, e.name, c.depth + 1, c.path || ' > ' || e.name          -- recursive step
    FROM employees e JOIN chain c ON e.manager_id = c.id
)
SELECT printf('%.*c', depth * 2, ' ') || name AS org, depth, path FROM chain ORDER BY path
"""
for org, depth, path in conn.execute(query):
    print(f"{org:<14} depth={depth}  {path}")

# Everyone under Alan (id 2), at any depth
under = conn.execute("""
WITH RECURSIVE reports(id) AS (
    SELECT id FROM employees WHERE manager_id = 2
    UNION ALL
    SELECT e.id FROM employees e JOIN reports r ON e.manager_id = r.id
)
SELECT name FROM employees WHERE id IN reports ORDER BY name""").fetchall()
print("Alan's org:", [n for (n,) in under])

Remember: CTEs name steps; recursive CTEs = anchor UNION ALL recursive step, with a termination condition.

Window functions

An aggregate with GROUP BY collapses rows: ten orders become two region totals. A window function computes across a set of related rows without collapsing them: every order keeps its row and gains a column such as "rank within its region" or "running total so far". The syntax is function(...) OVER (PARTITION BY … ORDER BY … frame):

PARTITION BY splits rows into independent groups (like GROUP BY, but without collapsing).
ORDER BY orders rows inside each partition (needed for ranking, LAG, running totals).
The frame (ROWS BETWEEN 2 PRECEDING AND CURRENT ROW) picks which rows around the current one an aggregate sees.

Ranking

Explanation

ROW_NUMBER always counts 1, 2, 3, 4 (ties broken by the extra player sort). RANK gives ties the same rank and then skips: 1, 1, 3, 4. DENSE_RANK gives ties the same rank without gaps: 1, 1, 2, 3.

The most common pattern in analytics is "top N per group": rank inside a partition, then filter in an outer query (you cannot filter on a window function in WHERE, because windows are computed after WHERE).

LAG and LEAD: comparing with neighbours

LAG(x, n, default) reads x from n rows earlier in the window order; LEAD reads later rows. They make period-over-period comparisons a one-liner.

The WINDOW w AS (…) clause names a window definition so you can reuse it.

Frames: running totals and moving averages

With ORDER BY inside OVER, an aggregate's default frame is "from the start of the partition up to the current row (and its ties)", which gives a running total. An explicit ROWS BETWEEN frame gives a moving window.

Remember: windows keep rows; PARTITION BY groups, ORDER BY orders, the frame chooses neighbours; filter on window results in an outer query.

CASE and conditional aggregation

CASE WHEN … THEN … ELSE … END is SQL's if/else expression. Inside an aggregate it becomes conditional aggregation, the standard way to pivot rows into columns: one pass over the data computes many filtered totals at once.

import sqlite3
conn = sqlite3.connect(":memory:")
conn.executescript("""
CREATE TABLE orders (id INTEGER PRIMARY KEY, customer_id INTEGER, order_date TEXT, amount REAL, status TEXT);
INSERT INTO orders VALUES
  (1,1,'2026-03-01',120,'paid'), (2,2,'2026-03-01', 80,'paid'), (3,1,'2026-03-02', 40,'refunded'),
  (4,3,'2026-03-02',200,'paid'), (5,2,'2026-03-03', 60,'paid'), (6,4,'2026-03-03', 90,'paid'),
  (7,1,'2026-03-04', 75,'paid'), (8,3,'2026-03-05', 30,'paid'), (9,4,'2026-03-05',110,'refunded'),
  (10,2,'2026-03-06',150,'paid');
""")
for row in conn.execute("""
    SELECT customer_id,
           SUM(CASE WHEN status = 'paid' THEN amount ELSE 0 END)     AS paid,
           SUM(CASE WHEN status = 'refunded' THEN amount ELSE 0 END) AS refunded,
           COUNT(*) FILTER (WHERE amount >= 100)                     AS big_orders,
           CASE WHEN SUM(amount) >= 250 THEN 'gold'
                WHEN SUM(amount) >= 150 THEN 'silver'
                ELSE 'bronze' END                                    AS tier
    FROM orders GROUP BY customer_id ORDER BY customer_id"""):
    print(row)

COUNT(*) FILTER (WHERE …) is the standard SQL shorthand for conditional aggregation (SQLite 3.30+, PostgreSQL, DuckDB). CASE works everywhere.

Remember: one GROUP BY plus several SUM(CASE …) columns is a pivot.

Upserts and idempotent loads

Pipelines re-run: after a crash, during a backfill, or because a file arrived twice. A load that uses plain INSERT duplicates rows on every re-run. An upsert ("update or insert") makes the load idempotent: running it twice leaves the same table as running it once. SQLite and PostgreSQL spell it INSERT … ON CONFLICT (key) DO UPDATE SET …; the special table excluded holds the row you tried to insert.

import sqlite3
conn = sqlite3.connect(":memory:")
conn.execute("CREATE TABLE log_append (product TEXT, qty INTEGER)")
conn.execute("CREATE TABLE stock (product TEXT PRIMARY KEY, qty INTEGER, updated TEXT)")
batch = [("apple", 10, "2026-03-01"), ("pear", 4, "2026-03-01")]

for run in range(2):                                   # the job runs twice
    conn.executemany("INSERT INTO log_append VALUES (?, ?)", [(p, q) for p, q, _ in batch])
    conn.executemany("""
        INSERT INTO stock (product, qty, updated) VALUES (?, ?, ?)
        ON CONFLICT (product) DO UPDATE SET qty = excluded.qty, updated = excluded.updated
    """, batch)

print("append rows:", conn.execute("SELECT COUNT(*) FROM log_append").fetchone()[0])
print("upsert rows:", conn.execute("SELECT * FROM stock ORDER BY product").fetchall())

# Only update when the incoming row is newer; ignore stale data
conn.execute("""
    INSERT INTO stock VALUES ('apple', 99, '2026-02-01')
    ON CONFLICT (product) DO UPDATE SET qty = excluded.qty, updated = excluded.updated
    WHERE excluded.updated > stock.updated
""")
print("after stale update:", conn.execute("SELECT * FROM stock WHERE product = 'apple'").fetchone())

# DO NOTHING keeps the first version; RETURNING shows what was written (SQLite 3.35+)
print(conn.execute("INSERT INTO stock VALUES ('kiwi', 7, '2026-03-02') ON CONFLICT DO NOTHING RETURNING product").fetchall())

The conflict target must be backed by a PRIMARY KEY or UNIQUE constraint. Other engines use MERGE INTO … WHEN MATCHED THEN UPDATE WHEN NOT MATCHED THEN INSERT (standard SQL, supported by PostgreSQL 15+, Snowflake, BigQuery, DuckDB 1.4+, Delta, Iceberg) for the same job.

Remember: idempotent loads = a natural key + upsert. You will rely on this in M62 and Project 9.

Working with dates

SQLite has no date type. Store timestamps as ISO-8601 text in UTC ('2026-03-05T13:00:00+00:00' or '2026-03-05 13:00:00'): ISO strings sort correctly as text, and the date functions understand them. The core functions are date(), time(), datetime(), julianday(), unixepoch() (3.38+), and strftime(format, value, modifiers…). Modifiers shift values: '+1 day', '-7 days', 'start of month', 'weekday 1' (next Monday).

import sqlite3
conn = sqlite3.connect(":memory:")
q = lambda sql: conn.execute(sql).fetchone()
print(q("SELECT date('2026-03-31', '+1 month')"))                 # SQLite normalizes overflow: 2026-05-01
print(q("SELECT date('2026-03-31', 'start of month', '+1 month', '-1 day')"))   # last day of March
print(q("SELECT strftime('%Y-%m', '2026-03-05'), strftime('%W', '2026-03-05'), strftime('%w', '2026-03-05')"))
print(q("SELECT julianday('2026-03-08 10:30') - julianday('2026-03-05 08:00')"))  # days between, fractional
print(q("SELECT CAST((unixepoch('2026-03-05 08:20:30') - unixepoch('2026-03-05 08:00:00')) AS INTEGER)"))  # seconds
print(q("SELECT date('2026-03-05', 'weekday 1')"))               # the following Monday

conn.execute("CREATE TABLE events (ts TEXT)")
conn.executemany("INSERT INTO events VALUES (?)", [("2026-03-02 09:00",), ("2026-03-04 18:00",), ("2026-03-10 07:00",)])
print(conn.execute("""SELECT date(ts, 'weekday 0', '-6 days') AS week_start, COUNT(*)
                      FROM events GROUP BY week_start ORDER BY week_start""").fetchall())

Remember: store UTC ISO-8601 text, bucket with strftime/date modifiers, and keep time-zone logic in one well-tested place.

Query plans and indexes

SQL is declarative: you say what, the query planner decides how. To see its decision, prefix a query with EXPLAIN QUERY PLAN. The two words to look for in SQLite are SCAN (read every row of the table, O(n)) and SEARCH … USING INDEX (jump to matching rows via a B-tree, O(log n) per lookup).

import random, sqlite3, time
conn = sqlite3.connect(":memory:")
conn.execute("CREATE TABLE orders (id INTEGER PRIMARY KEY, customer_id INTEGER, order_date TEXT, amount REAL)")
rng = random.Random(0)
conn.executemany("INSERT INTO orders (customer_id, order_date, amount) VALUES (?, ?, ?)",
                 [(rng.randrange(5000), f"2026-03-{rng.randint(1, 28):02d}", rng.random() * 100) for _ in range(200_000)])

sql = "SELECT SUM(amount) FROM orders WHERE customer_id = 42"
def plan(s):
    return [row[-1] for row in conn.execute("EXPLAIN QUERY PLAN " + s)]
def timed(s, n=50):
    t = time.perf_counter()
    for _ in range(n):
        conn.execute(s).fetchone()
    return (time.perf_counter() - t) / n * 1000

print(plan(sql), f"{timed(sql):.2f} ms")
conn.execute("CREATE INDEX ix_orders_customer ON orders (customer_id)")
print(plan(sql), f"{timed(sql):.3f} ms")
conn.execute("CREATE INDEX ix_orders_customer_amount ON orders (customer_id, amount)")
print(plan(sql), f"{timed(sql):.3f} ms")   # COVERING: answered from the index alone
print(plan("SELECT * FROM orders WHERE amount > 99"))   # no usable index -> SCAN

Index guidelines for analytical work:

Index columns used in selective WHERE filters and join keys. A filter matching half the table gains nothing from an index; a scan is cheaper.
A composite index (a, b) helps filters on a, or a and b, but not on b alone (leftmost-prefix rule).
A covering index includes every column the query needs, so the table is never touched.
Indexes cost write speed and disk space. Columnar OLAP engines like DuckDB rarely need them: they rely on fast scans plus min/max zone maps to skip data.
Run ANALYZE after bulk loads so the planner has statistics.

Remember: read the plan before guessing; SCAN on a big table behind a selective filter means an index is missing.

DuckDB: in-process OLAP

DuckDB is "SQLite for analytics": an embedded database that runs inside your Python process (no server), but with a columnar, vectorized, multi-threaded engine built for scans and aggregations. It reads CSV, JSON, and Parquet files directly (locally or from S3/HTTP), speaks a rich PostgreSQL-like SQL dialect, and exchanges data with pandas, Polars, and Arrow with little or no copying. For data up to hundreds of gigabytes on one machine, it often replaces a cluster.

DuckDB's SQL has friendly extensions worth knowing: GROUP BY ALL, SELECT * EXCLUDE (col), FROM tbl without SELECT, QUALIFY to filter on window functions directly (QUALIFY row_number() OVER (…) = 1), PIVOT/UNPIVOT, and list/struct types. Every window-function query you wrote for SQLite runs unchanged.

Remember: DuckDB = embedded, columnar, vectorized; query files in place; hand results to pandas or Polars with .df() / .pl().

SQL style and testing

Analytical SQL lives for years and is read far more than written. Professional conventions:

One clause per line; uppercase keywords (or consistently lowercase); snake_case names.
CTEs over nested subqueries, each named for what it holds (paid_orders, not t2).
Explicit JOIN … ON, never comma joins; qualify columns with short table aliases.
No SELECT * in production queries: name the columns, so upstream additions can't change your output.
Always bind values with parameters (? or :name), never f-strings: that prevents SQL injection and lets the database reuse plans.
Lint and format automatically: pixi add sqlfluff, then pixi run sqlfluff lint queries/.

Test SQL like code: build a tiny dataset where you know the right answer by hand, run the query, assert the result. Keep queries in .sql files or module constants so tests and production run the same text.

# test_reports.py  (run with: pixi run pytest)
import sqlite3
import pytest
from reports import TOP_CUSTOMER_PER_REGION   # the production query string

@pytest.fixture
def conn():
    c = sqlite3.connect(":memory:")
    c.executescript("""
        CREATE TABLE customers (id INTEGER PRIMARY KEY, name TEXT, region TEXT);
        CREATE TABLE orders (id INTEGER PRIMARY KEY, customer_id INTEGER, amount REAL, status TEXT);
        INSERT INTO customers VALUES (1, 'Ada', 'north'), (2, 'Cy', 'north'), (3, 'Ben', 'south');
        INSERT INTO orders VALUES (1, 1, 50, 'paid'), (2, 2, 70, 'paid'), (3, 2, 500, 'refunded'), (4, 3, 10, 'paid');
    """)
    yield c
    c.close()

def test_refunds_do_not_count(conn):
    rows = conn.execute(TOP_CUSTOMER_PER_REGION).fetchall()
    assert rows == [("north", "Cy", 70.0), ("south", "Ben", 10.0)]

def test_empty_tables_return_nothing(conn):
    conn.execute("DELETE FROM orders")
    assert conn.execute(TOP_CUSTOMER_PER_REGION).fetchall() == []

Include edge cases in the fixture on purpose: ties, NULLs, refunds, a region with no orders, a day with no sales. In dbt the same idea appears as data tests (unique, not_null, relationships) and unit tests with mocked inputs.

Remember: readable CTEs, named columns, bound parameters, and a hand-checked fixture for every query.

Practice

Each exercise asks you to assign a SQL string to a variable. The tests build the dataset shown in the prompt in an in-memory SQLite database, run your query, and compare the rows. Column names don't matter; column order and row order do.

Table orders(id, order_date TEXT, amount REAL, status TEXT) where status is 'paid' or 'refunded'. Write QUERY returning one row per order_date (ascending) with: the date, the number of paid orders, paid revenue, and refunded revenue (use 0, not NULL, when there is none).

Hint

SUM(CASE WHEN status = 'paid' THEN 1 ELSE 0 END) counts; SUM(CASE WHEN … THEN amount ELSE 0 END) totals. Integer 0 vs 0.0: SQLite sums a REAL column to a float.

Show solution

QUERY = """
SELECT order_date,
       SUM(CASE WHEN status = 'paid' THEN 1 ELSE 0 END)          AS paid_orders,
       SUM(CASE WHEN status = 'paid' THEN amount ELSE 0.0 END)     AS paid_revenue,
       SUM(CASE WHEN status = 'refunded' THEN amount ELSE 0.0 END) AS refunded_revenue
FROM orders
GROUP BY order_date
ORDER BY order_date
"""

Table orders(id, customer_id, order_date TEXT, amount REAL). Write QUERY returning each customer's most recent order as (customer_id, id, order_date, amount), ordered by customer_id. If a customer has two orders on the same latest date, pick the one with the larger id. Use ROW_NUMBER().

Hint

ROW_NUMBER() OVER (PARTITION BY customer_id ORDER BY order_date DESC, id DESC), then keep rn = 1 in the outer query.

Show solution

QUERY = """
WITH ranked AS (
    SELECT customer_id, id, order_date, amount,
           ROW_NUMBER() OVER (PARTITION BY customer_id ORDER BY order_date DESC, id DESC) AS rn
    FROM orders
)
SELECT customer_id, id, order_date, amount
FROM ranked
WHERE rn = 1
ORDER BY customer_id
"""

Table daily(day TEXT, revenue REAL), one row per day. Write QUERY returning (day, revenue, running_total, moving_avg) ordered by day, where moving_avg is the average of the current and up to two previous rows, rounded to 2 decimals with ROUND.

Hint

SUM(revenue) OVER (ORDER BY day) and AVG(revenue) OVER (ORDER BY day ROWS BETWEEN 2 PRECEDING AND CURRENT ROW).

Show solution

QUERY = """
SELECT day,
       revenue,
       SUM(revenue) OVER (ORDER BY day) AS running_total,
       ROUND(AVG(revenue) OVER (ORDER BY day ROWS BETWEEN 2 PRECEDING AND CURRENT ROW), 2) AS moving_avg
FROM daily
ORDER BY day
"""

Table sales(region TEXT, month TEXT, revenue REAL). Write QUERY returning (region, month, revenue, prev_revenue, growth_pct) ordered by region then month. prev_revenue is the previous month's revenue in the same region (NULL for the first month), and growth_pct is ROUND(100.0 * (revenue - prev) / prev, 1) (NULL when there is no previous month).

Hint

Compute LAG(revenue) OVER (PARTITION BY region ORDER BY month) in a CTE, then do the arithmetic in the outer query. Arithmetic with NULL gives NULL, which is what you want.

Show solution

QUERY = """
WITH with_prev AS (
    SELECT region, month, revenue,
           LAG(revenue) OVER (PARTITION BY region ORDER BY month) AS prev_revenue
    FROM sales
)
SELECT region, month, revenue, prev_revenue,
       ROUND(100.0 * (revenue - prev_revenue) / prev_revenue, 1) AS growth_pct
FROM with_prev
ORDER BY region, month
"""

Table employees(id, name, manager_id); the CEO has manager_id NULL. Write QUERY with a recursive CTE that returns (name, depth) for every employee who reports, directly or indirectly, to the employee with id 2, where depth is 1 for direct reports, 2 for their reports, and so on. Order by depth, then name. Employee 2 is not included.

Hint

Anchor: SELECT id, name, 1 FROM employees WHERE manager_id = 2. Recursive step: join employees e to reports r on e.manager_id = r.id with r.depth + 1.

Show solution

QUERY = """
WITH RECURSIVE reports(id, name, depth) AS (
    SELECT id, name, 1 FROM employees WHERE manager_id = 2
    UNION ALL
    SELECT e.id, e.name, r.depth + 1
    FROM employees e JOIN reports r ON e.manager_id = r.id
)
SELECT name, depth FROM reports ORDER BY depth, name
"""

Write load_inventory(conn, rows) where rows is a list of (sku, qty, as_of) tuples (as_of is an ISO date string). Create the table inventory(sku TEXT PRIMARY KEY, qty INTEGER, as_of TEXT) if it doesn't exist, then upsert every row so that an incoming row replaces the stored one only if its as_of is newer. Loading the same batch twice must leave the same table. Commit at the end.

Hint

ON CONFLICT (sku) DO UPDATE SET qty = excluded.qty, as_of = excluded.as_of WHERE excluded.as_of > inventory.as_of.

Show solution

def load_inventory(conn, rows):
    conn.execute("CREATE TABLE IF NOT EXISTS inventory (sku TEXT PRIMARY KEY, qty INTEGER, as_of TEXT)")
    conn.executemany("""
        INSERT INTO inventory (sku, qty, as_of) VALUES (?, ?, ?)
        ON CONFLICT (sku) DO UPDATE SET qty = excluded.qty, as_of = excluded.as_of
        WHERE excluded.as_of > inventory.as_of
    """, rows)
    conn.commit()

Mini-project: a sales analytics SQL report

Build a small reporting package whose every number comes from a tested SQL query. Generate a year of sales for a fictional store (products, categories, regions, salespeople) with a fixed seed, load it into SQLite, and produce a Markdown report.

pixi init sales-report && cd sales-report
pixi add python pytest
pixi add duckdb polars          # for the stretch goals
pixi run python generate.py     # writes sales.db
pixi run python report.py > report.md
pixi run pytest

Queries to write (each a constant in queries.py):

Monthly revenue with month-over-month growth (LAG).
Running year-to-date revenue and a 3-month moving average.
Top 3 products per category by revenue (DENSE_RANK, ties included).
Each salesperson's share of their region's revenue (window SUM over a partition).
A region × quarter pivot with conditional aggregation.
Customers whose order count dropped by half or more versus the previous quarter.

Queries are CTE-structured, use named columns, and bind every parameter
Each query has a pytest test against a small hand-checked fixture (including a tie and an empty month)
The report renders as a Markdown table per query
EXPLAIN QUERY PLAN output is checked for the two slowest queries, with an index added where it helps
Re-running generate.py produces identical data (fixed seed, upserts)

Stretch: export the tables to Parquet with DuckDB and run the same queries with DuckDB, replacing the top-N subquery with QUALIFY; compare timings on 10 million rows; return results as Polars frames.

Quiz

Recap

Logical order: FROM → WHERE → GROUP BY → HAVING → windows → SELECT → ORDER BY → LIMIT.
CTEs make queries readable; recursive CTEs walk hierarchies and generate series, with a termination condition.
Window functions keep rows: ranking, LAG/LEAD, running totals, and moving averages via PARTITION BY, ORDER BY, and frames.
Conditional aggregation pivots rows into columns; mind NULLs.
Upserts on a natural key make loads idempotent; prefer ON CONFLICT DO UPDATE over INSERT OR REPLACE.
Store UTC ISO text, bucket with date functions; read query plans and index selective filters.
DuckDB brings columnar OLAP into your process, queries files directly, and swaps data with pandas and Polars.
Treat SQL as code: style, named columns, parameters, and tests with hand-checked fixtures.

Going further

sqlite3 in the Python docs
SQLite: window functions, WITH clause, upsert, date functions, EXPLAIN QUERY PLAN
DuckDB documentation, including the Python client
modern-sql.com (Markus Winand) and Use The Index, Luke
Next: M62 · Pipelines, Orchestration & Data Quality