sql

name	sql
description	Reference for writing and tuning SQL on tabular sources: SELECT/INSERT/UPDATE/DELETE, INNER/LEFT/RIGHT/FULL/SEMI/ANTI joins, window functions (ROW_NUMBER, RANK, LAG/LEAD, running aggregates), recursive and non-recursive CTEs, EXPLAIN/EXPLAIN ANALYZE, index design, and the three Python access layers (DuckDB embedded analytics over Parquet/CSV, SQLAlchemy core/ORM for multi-dialect modeling, psycopg2 for direct PostgreSQL with cursor control). Includes worked feature-extraction and slow-query-diagnosis examples.
metadata	{"dependencies":["duckdb","sqlalchemy","psycopg2-binary"]}

SQL Reference

Three steps: identify the question shape, pick the SQL technique (Section 1), pick the access library (Section 3). Optimization is mechanical once you have a plan (Section 2).

1. Question to SQL technique

Question	Technique	Skeleton
Filter / top-N overall	WHERE + ORDER BY ... LIMIT N	WHERE col=? ORDER BY s DESC LIMIT 10
Top-N per group	ROW_NUMBER() window	ROW_NUMBER() OVER (PARTITION BY g ORDER BY s DESC)
Rank with ties	RANK() / DENSE_RANK()	RANK() OVER (ORDER BY s DESC)
Running / moving sum	SUM() OVER with frame	SUM(x) OVER (PARTITION BY g ORDER BY t ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)
Previous / next row	LAG / LEAD	LAG(x,1) OVER (PARTITION BY g ORDER BY t)
Group + filter aggregate	GROUP BY + HAVING	GROUP BY g HAVING SUM(x) > 0
Pivot rows to columns	conditional aggregation	SUM(CASE WHEN k='A' THEN v ELSE 0 END)
Composable subquery	non-recursive WITH	WITH a AS (...), b AS (...) SELECT ... FROM a JOIN b
Hierarchical / parent-child	recursive CTE	WITH RECURSIVE t AS (anchor UNION ALL recursive)
Rows in A not in B	LEFT JOIN ... WHERE b.k IS NULL / EXCEPT	EXCEPT deduplicates
Rows in A that have a B	EXISTS (semi-join)	WHERE EXISTS (SELECT 1 FROM b WHERE b.k=a.k)
Inner / left / full / cross join	[INNER|LEFT|FULL OUTER|CROSS] JOIN	strict / optional / diff / densify
Bulk insert / upsert	INSERT ... SELECT / INSERT ... ON CONFLICT	ON CONFLICT(k) DO UPDATE SET v=EXCLUDED.v
Update from another table	UPDATE ... FROM (PG) / UPDATE ... JOIN (MySQL)	UPDATE t SET v=s.v FROM s WHERE t.k=s.k

2. Optimization checklist

Step	Action
1	Capture the plan: EXPLAIN (shape) / EXPLAIN ANALYZE (timings)
2	Index filter and join columns; check for Seq Scan on filtered tables
3	Drop SELECT *; project only needed columns
4	Avoid functions / casts on indexed columns (WHERE LOWER(col)=? disables index)
5	Replace IN (subquery) with JOIN or EXISTS; batch large UPDATE/DELETE by key range
6	Add covering index for hot reads (CREATE INDEX ix ON t(a,b) INCLUDE (c)); ANALYZE table after bulk loads

Anti-patterns: OR across different indexes (use UNION ALL); implicit type casts; correlated scalar subqueries in SELECT; ORDER BY RAND() on large tables (use TABLESAMPLE).

3. Library dispatch

Use case	Library	Why
Embedded analytics over Parquet / CSV / Pandas	duckdb	Vectorised, zero-copy from arrow / pandas, no server
Light embedded transactional	sqlite3 (stdlib)	File-backed, no install
Multi-dialect modeling, ORM, migrations	sqlalchemy (+ alembic)	Dialect abstraction, schema as Python
Direct PostgreSQL, cursor control, COPY	psycopg2	Server-side cursors, fast bulk I/O

sqlalchemy wraps either backend via URL (postgresql+psycopg2://..., duckdb:///path.db, sqlite:///x.db).

4. Connection patterns

# DuckDB embedded — analytical query over Parquet, no schema needed
import duckdb
con = duckdb.connect()                              # in-memory; or "warehouse.duckdb"
df = con.execute("SELECT region, SUM(amount) AS total "
                 "FROM read_parquet('events/*.parquet') "
                 "WHERE ts >= '2026-01-01' GROUP BY region").fetch_df()

# sqlite3 stdlib — embedded transactional, schema introspection
import sqlite3
con = sqlite3.connect("app.db"); con.row_factory = sqlite3.Row
con.execute("SELECT name FROM sqlite_master WHERE type='table'").fetchall()
con.execute("PRAGMA table_info(users)").fetchall()  # (cid, name, type, notnull, dflt, pk)

# SQLAlchemy core + ORM — dialect-agnostic engine, parameterised query, declarative model
from sqlalchemy import create_engine, text, Column, Integer, String
from sqlalchemy.orm import declarative_base, Session
eng = create_engine("postgresql+psycopg2://user:pw@host/db", future=True)
with eng.connect() as cx:                                        # core
    rows = cx.execute(text("SELECT id, name FROM users WHERE status = :s"),
                      {"s": "active"}).all()
Base = declarative_base()
class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True); name = Column(String)
with Session(eng) as s:                                          # ORM
    users = s.query(User).filter(User.name.like("A%")).limit(10).all()

# psycopg2 — server-side named cursor for streaming large results
import psycopg2
with psycopg2.connect("dbname=db user=u password=pw host=h") as conn:
    with conn.cursor(name="bulk_export") as cur:                # server-side
        cur.itersize = 10_000
        cur.execute("SELECT id, payload FROM events WHERE ts >= %s", ("2026-01-01",))
        for row in cur: handle(row)                              # streamed

Always parameterise (%s for psycopg2, :name via text() for SQLAlchemy, ? for sqlite3 / DuckDB). Never f-string user input into SQL — that is the injection footgun.

5. Worked example: feature extraction over Parquet (DuckDB)

Per-user ML features — 30-day spend, rank by spend within country, days since last order. CTE + window + aggregate, materialised to Pandas.

import duckdb
q = """
WITH recent AS (
    SELECT user_id, country, ts, amount FROM read_parquet('orders/*.parquet')
    WHERE ts >= CURRENT_DATE - INTERVAL 30 DAY
),
agg AS (
    SELECT user_id, ANY_VALUE(country) AS country,
           SUM(amount) AS spend_30d, MAX(ts) AS last_ts, COUNT(*) AS n_orders_30d
    FROM recent GROUP BY user_id
)
SELECT a.user_id, a.country, a.spend_30d, a.n_orders_30d,
       DATE_DIFF('day', a.last_ts, CURRENT_DATE) AS days_since_last,
       RANK() OVER (PARTITION BY a.country ORDER BY a.spend_30d DESC) AS spend_rank_in_country
FROM agg a
"""
features = duckdb.connect().execute(q).fetch_df()

The CTE narrows the scan; the window runs over the small aggregate, not the raw stream. Same pattern against a PG warehouse via sqlalchemy + pd.read_sql(text(q), eng).

6. Worked example: diagnose a slow query (EXPLAIN ANALYZE)

import psycopg2
sql = ("SELECT u.id, u.name, COUNT(o.id) AS n FROM users u "
       "LEFT JOIN orders o ON o.user_id=u.id WHERE u.created_at >= %s "
       "GROUP BY u.id, u.name ORDER BY n DESC LIMIT 50")
with psycopg2.connect(DSN) as cx, cx.cursor() as cur:
    cur.execute("EXPLAIN (ANALYZE, BUFFERS, FORMAT TEXT) " + sql, ("2026-01-01",))
    for line in cur.fetchall(): print(line[0])

Read the plan top-down. Red flags and fixes:

Symptom	Fix
Seq Scan on a selectively filtered table	CREATE INDEX on the filter column
Hash Join builds huge hash on the large side	swap join order; inner = small filtered set
Sort spills to disk (external merge)	raise work_mem or push LIMIT below the sort
Rows Removed by Filter >> rows returned	predicate not sargable; rewrite to hit the index
loops=N large on inner of Nested Loop	force hash join via ANALYZE / index change

DuckDB has the same workflow (EXPLAIN ANALYZE SELECT ...;). Loop: capture, change one thing, re-capture, compare.

7. Efficient access patterns

Access shape	Schema move
Point lookup by k / range scan on ts per tenant	B-tree on k (PK) / composite (tenant_id, ts)
Heavy analytical scans	columnar store (DuckDB / Parquet), not row-store + indexes
Read-modify-write hotspots	SELECT ... FOR UPDATE in an explicit transaction
Append-only event log	partition by ts; query by partition key

Pitfalls

• SELECT * over wide tables; f-string user input into SQL (always parameterise).
• LEFT JOIN then WHERE on the right column — silently becomes inner join; filter in ON.
• ORDER BY ... LIMIT 1 per group via subquery instead of ROW_NUMBER() = 1.
• ORM N+1 — fix with eager loading or a single JOIN; forgetting ANALYZE after bulk load.