From Files to Tables, Building a Data Loader You Can Run Twice

In the last post, my project ended with a folder of CSV files which included a synthetic town of 1,000 people and about 60,000 bank transactions.

The thing about files is that they are useful, but they're like letters in a drawer. You can read them one at a time. You can't easily ask, "show me every salary paid in March to anyone who arrived in 2024."

A database can answer that in milliseconds. So this stage of the Newcomer Credit Data Platform is about moving data from files into PostgreSQL. It sounds like the most boring step in data engineering. It's actually where a lot of pipelines quietly go wrong. You end up with duplicated rows, half-loaded files, data nobody can trace back to its source, and failures that leave no record.

This post covers how I designed the loading step to avoid all of that, the transaction bug my tests caught that manual runs never showed, and the SQL I wrote once the data was in.

Series so far:

1. Before the First Pipeline: setting up the machine and project base
2. Fake Data, Real Discipline: generating a seeded synthetic population
3. From Files to Tables (this post)

The code is on GitHub.

What I ended up with

Loading the data:

$ make load
  customers          run   1:   1,000 inserted,       0 already loaded
  bank_transactions  run   2:  59,801 inserted,       0 already loaded
  applications       run   3:   1,000 inserted,       0 already loaded

Loading it again:

$ make load
  customers          run   4:       0 inserted,   1,000 already loaded
  bank_transactions  run   5:       0 inserted,  59,801 already loaded
  applications       run   6:       0 inserted,   1,000 already loaded

Nothing changed the second time, and both attempts are on record. A file that's been tampered with, reshaped or corrupted is refused loudly, logged as failed, and leaves zero partial rows behind. Twenty-two tests prove all of it, including ten that run against a real PostgreSQL database without ever touching the real data.

Part 1: Six ideas behind the loader

Before any code, these are the ideas the whole design rests on. Each one applies to loading anything. And that includes files, API responses, database extracts.

1. The loading dock

Warehouses don't carry deliveries straight onto the shelves. Trucks unload at a loading dock, goods get checked against the delivery note, and only then do they move to the right shelf.

My loader works the same way. It copies each CSV into a temporary table where every column is plain text, exactly as it arrived. Then it converts and checks the data on the way into the real table. The temporary table disappears afterwards.

Why land everything as text first? Because the file is still untrusted. Loading text never fails on a strange value, so all the converting and checking happens in one controlled place.

2. One truck, not sixty thousand trips

There are two ways to get 60,000 rows into a database. You can send 60,000 separate INSERT statements, which is like moving house by carrying each item to the new place one at a time. Or you can use PostgreSQL's COPY command, which streams the whole file in one go, like loading everything into a single truck.

COPY is often tens of times faster. On real volumes, that's the difference between a load that takes seconds and one that takes hours.

3. The guest list at the door

Every table has a primary key. This is a column whose value must be unique, like a national ID number. transaction_id is the primary key of the transactions table.

When the loader moves rows from the dock to the shelf, it says ON CONFLICT (transaction_id) DO NOTHING. Think of a bouncer with a guest list. If someone with that ID is already inside, they simply don't get in again. No error, no fuss.

That single line makes the loader idempotent. If you run it once or ten times, the result remains the same. Pipelines get retried constantly, after network blips, timeouts and crashes. If a retry can create duplicates, every failure becomes a manual clean-up job.

4. All or nothing

A bank transfer has two halves: money leaves one account and arrives in another. If a system crashed between them, money would vanish. So the database groups the steps into a transaction. Either every step happens, or none of them do.

My loader puts each file's rows and its "succeeded" status into one transaction. If row 45,000 has a broken value, the entire file rolls back. There's never half a file loaded with no record of what happened.

5. The flight recorder

Planes carry a flight recorder so investigators can reconstruct exactly what happened. My loader keeps one too. That's the table called raw.ingestion_runs, with one row per load attempt. It records which file was loaded, its fingerprint, how many rows went in, how many were skipped, and whether the attempt succeeded or failed, with the error.

It writes a running entry before loading, in its own committed transaction. So even a load that crashes leaves a record behind.

6. Check the seal and the shape before opening the box

Before loading a single row, the loader checks two things.

The seal. The generator writes a manifest.json containing a SHA-256 fingerprint of every file. The loader recomputes the fingerprint and compares. If it doesn't match, the file changed after it was produced. The classic cause is someone opening a CSV in Excel and saving it, which silently reformats dates, strips leading zeros and changes number formats.

The shape. The loader checks the header has exactly the expected columns, in order. If a source renames a column, the load stops instead of putting data in the wrong places.

Both checks fail loudly. A pipeline that stops and complains is annoying. A pipeline that silently loads wrong data is dangerous.

Part 2: Designing the landing tables

The raw schema is the delivery shelf from part one: data lands here as received and is never edited afterwards. Here's the full table definition, then each decision behind it.

sql/02_raw_tables.sql

-- 02_raw_tables.sql
-- The landing tables for generated data, plus an audit log of every load.
-- Run as ncp_dev inside newcomer_credit:  make tables
-- Safe to re-run: IF NOT EXISTS skips tables that already exist.

-- ---------------------------------------------------------------------------
-- The load log. One row per file, per load attempt, whether it worked or not.
-- ---------------------------------------------------------------------------
CREATE TABLE IF NOT EXISTS raw.ingestion_runs (
    run_id             BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    source_name        TEXT        NOT NULL,   -- e.g. 'bank_transactions'
    file_path          TEXT        NOT NULL,
    file_sha256        TEXT        NOT NULL,   -- fingerprint of the exact file loaded
    generator_version  TEXT,                   -- from manifest.json
    generator_seed     INTEGER,                -- from manifest.json
    rows_in_file       INTEGER,
    rows_inserted      INTEGER,
    rows_skipped       INTEGER,                -- already loaded before
    status             TEXT        NOT NULL
                       CHECK (status IN ('running', 'succeeded', 'failed')),
    error_message      TEXT,
    started_at         TIMESTAMPTZ NOT NULL DEFAULT now(),
    finished_at        TIMESTAMPTZ
);

-- ---------------------------------------------------------------------------
-- Landing tables. Columns match the CSV files, plus two lineage columns:
--   _run_id     which load brought this row in
--   _loaded_at  when it arrived
-- No foreign keys here on purpose: real sources arrive in any order.
-- Relationships get tested later, in dbt, instead of enforced at the door.
-- ---------------------------------------------------------------------------
CREATE TABLE IF NOT EXISTS raw.customers (
    customer_id   TEXT        PRIMARY KEY,
    arrival_date  DATE        NOT NULL,
    income_type   TEXT        NOT NULL,
    _run_id       BIGINT      NOT NULL,
    _loaded_at    TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE TABLE IF NOT EXISTS raw.bank_transactions (
    transaction_id  TEXT          PRIMARY KEY,
    customer_id     TEXT          NOT NULL,
    amount          NUMERIC(12,2) NOT NULL,   -- exact money: never FLOAT
    currency        CHAR(3)       NOT NULL,
    counterparty    TEXT          NOT NULL,
    description     TEXT,
    event_at        TIMESTAMPTZ   NOT NULL,   -- when it happened
    booked_at       TIMESTAMPTZ   NOT NULL,   -- when the bank recorded it
    available_at    TIMESTAMPTZ   NOT NULL,   -- when the lender could see it
    _run_id         BIGINT        NOT NULL,
    _loaded_at      TIMESTAMPTZ   NOT NULL DEFAULT now()
);

CREATE TABLE IF NOT EXISTS raw.applications (
    application_id  TEXT        PRIMARY KEY,
    customer_id     TEXT        NOT NULL,
    decided_at      TIMESTAMPTZ NOT NULL,
    _run_id         BIGINT      NOT NULL,
    _loaded_at      TIMESTAMPTZ NOT NULL DEFAULT now()
);

-- Speeds up the most common question later: "this customer's transactions, in time order".
CREATE INDEX IF NOT EXISTS ix_bank_transactions_customer_event
    ON raw.bank_transactions (customer_id, event_at);

The types

Each column's type is a promise about what it holds, and PostgreSQL enforces it on every row.

Type	What it holds	Where I use it
TEXT	Any text, any length	IDs, names, descriptions
DATE	A calendar date, no time	arrival_date
NUMERIC(12,2)	An exact decimal: up to 12 digits, 2 after the point	amount
CHAR(3)	Exactly three characters	currency
TIMESTAMPTZ	A moment in time, time zone aware	every timestamp
INTEGER / BIGINT	Whole numbers (BIGINT goes much higher)	counts and IDs

Money is never FLOAT. Floating-point numbers can't store most decimals exactly; in most languages, 0.1 + 0.2 comes out as 0.30000000000000004. With money, those tiny errors compound into totals that don't reconcile. NUMERIC is exact.

Timestamps are always TIMESTAMPTZ. It stores an exact moment, independent of where you're viewing it from. Plain TIMESTAMP stores a clock reading with no time zone, which is ambiguous the moment data crosses regions.

The constraints

Constraints are rules the database checks on every row, forever, for free.

transaction_id  TEXT  PRIMARY KEY,

Unique and never empty. It's what the ON CONFLICT bouncer checks against.

amount  NUMERIC(12,2)  NOT NULL,

Must always have a value.

status  TEXT  NOT NULL  CHECK (status IN ('running', 'succeeded', 'failed')),

Only these values are allowed. A typo like 'succeded' is rejected by the database itself, not by hoping the code is right.

run_id  BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,

PostgreSQL numbers the rows itself: 1, 2, 3 and onwards. Nobody supplies a run_id by hand.

_loaded_at  TIMESTAMPTZ  NOT NULL  DEFAULT now(),

Filled automatically with the current time if no value is given.

The lineage columns

Every landing table carries two columns that aren't in the source file:

• _run_id: which load brought this row in, pointing to a row in ingestion_runs.
• _loaded_at: when it arrived.

The leading underscore is a convention meaning "added by the pipeline, not from the source." With these two columns, any single transaction traces back to the exact file, fingerprint and moment it came from. That traceability is called lineage, and in regulated industries like lending, being able to answer "where did this number come from?" isn't optional.

What's deliberately missing is the foreign keys

A foreign key would force every customer_id in the transactions table to already exist in the customers table. That sounds sensible, but real sources don't arrive politely in order. A transactions file can land before the customers file it refers to. If the raw layer enforced foreign keys, loads would fail for reasons that aren't real errors.

So the raw layer accepts what arrives, and relationships are tested downstream instead of enforced at the door. The rule I'm keeping: raw is for recording what arrived. Judging it comes later.

The index

CREATE INDEX IF NOT EXISTS ix_bank_transactions_customer_event
    ON raw.bank_transactions (customer_id, event_at);

An index is the index at the back of a book. Without it, finding one customer's transactions means reading all 60,000 rows. With it, PostgreSQL jumps straight to the right page. I built it on the question this project asks most often: "this customer's transactions, in time order."

Creating the tables

make tables

make tables runs the file with psql. Because every statement uses IF NOT EXISTS, running it twice is harmless, the same idempotent setup principle from part one.

Part 3: The loader, step by step

The loader is one Python file, ingestion/load_generated.py, run with make load. The full file is at the end of this post. Here are the parts that matter.

A recipe for each source

SOURCES = {
    "customers": {
        "key": "customer_id",
        "columns": ["customer_id", "arrival_date", "income_type"],
        "casts": ["customer_id", "arrival_date::date", "income_type"],
    },
    "bank_transactions": {
        "key": "transaction_id",
        "columns": ["transaction_id", "customer_id", "amount", "currency", "counterparty",
                    "description", "event_at", "booked_at", "available_at"],
        "casts": ["transaction_id", "customer_id", "amount::numeric(12,2)", "currency",
                  "counterparty", "description", "event_at::timestamptz",
                  "booked_at::timestamptz", "available_at::timestamptz"],
    },
    ...
}

For each table: its primary key, the exact columns expected in the file, and how to convert each one from text to its real type. :: is PostgreSQL's conversion operator, so '1770.00'::numeric(12,2) turns text into an exact number.

The generator's hidden "answer key" file isn't in this list. The surest way to keep sensitive data out of a table is for the loader to have no idea it exists.

Step 1: log the attempt before anything else

def start_run(conn, schema, name, path, digest, manifest) -> int:
    """Log the attempt and COMMIT it, so even a failed load leaves a record."""
    with conn.transaction():
        row = conn.execute(
            f"""
            INSERT INTO {schema}.ingestion_runs
                (source_name, file_path, file_sha256, generator_version,
                 generator_seed, rows_in_file, status)
            VALUES (%s, %s, %s, %s, %s, %s, 'running')
            RETURNING run_id
            """,
            (name, str(path), digest, manifest.get("generator_version"),
             manifest.get("seed"), manifest["files"][name]["rows"]),
        ).fetchone()
    return row[0]

The running row gets its own transaction, so it's committed immediately. RETURNING run_id hands back the number PostgreSQL just assigned, which every loaded row will carry.

Steps 2 and 3: check the seal and the shape

if digest != expected["sha256"]:
    raise LoadError(
        f"{path.name}: checksum does not match manifest.json. "
        "The file changed after it was generated (was it opened and saved in Excel?)."
    )

header = read_header(path)
if header != spec["columns"]:
    raise LoadError(
        f"{path.name}: unexpected columns.\n  expected {spec['columns']}\n  found    {header}"
    )

raise stops everything immediately. LoadError is a custom error type, which lets the tests check for exactly this kind of failure rather than any failure at all. Notice the messages say what's wrong and suggest the likely cause. Error messages are written for the person reading them at 3am.

Step 4: the loading dock

with conn.transaction():
    conn.execute(
        f"CREATE TEMP TABLE tmp_load ({', '.join(c + ' TEXT' for c in spec['columns'])}) "
        "ON COMMIT DROP"
    )
    with conn.cursor().copy(
        f"COPY tmp_load ({cols}) FROM STDIN WITH (FORMAT csv, HEADER true)"
    ) as copy:
        with path.open("rb") as f:
            while chunk := f.read(1024 * 1024):
                copy.write(chunk)

A temporary table of plain text columns matching the file. ON COMMIT DROP means it vanishes when the transaction ends. COPY ... FROM STDIN streams the file into it in one-megabyte chunks, so even a huge file never needs to fit in memory.

Step 5: onto the shelf, past the bouncer

inserted = conn.execute(
    f"""
    INSERT INTO {schema}.{name} ({cols}, _run_id)
    SELECT {casts}, %s FROM tmp_load
    ON CONFLICT ({spec['key']}) DO NOTHING
    """,
    (run_id,),
).rowcount

INSERT ... SELECT copies rows from the dock into the real table, converting types as it goes. ON CONFLICT ... DO NOTHING skips rows already there. .rowcount reports how many actually went in, and everything else counts as skipped.

%s is a placeholder, and it matters. The run_id value is passed separately, and the database driver inserts it safely. Values should never be pasted into SQL text with string formatting, because that's how SQL injection happens: input crafted to rewrite your query. The table and column names in the f-string come from my own SOURCES dictionary, never from outside input, which is the only reason they're formatted in directly.

Step 6: record the outcome

conn.execute(
    f"""
    UPDATE {schema}.ingestion_runs
    SET status = 'succeeded', rows_inserted = %s, rows_skipped = %s,
        finished_at = now()
    WHERE run_id = %s
    """,
    (inserted, rows_in_file - inserted, run_id),
)

Still inside the same transaction as the insert. Rows and status land together, or neither does.

If anything goes wrong at any point:

except Exception as exc:
    finish_failed(conn, schema, run_id, str(exc))
    raise

The data transaction has already rolled back automatically. finish_failed marks the run as failed with the error message, in a fresh transaction. Then raise re-throws the error so the program stops loudly instead of carrying on as if nothing happened.

Part 4: Proving it works

First load, second load

make seed
make load
make load

The first load inserts everything. The second inserts nothing. Then the flight recorder, from inside the database:

make psql

SELECT run_id, source_name, status, rows_in_file, rows_inserted, rows_skipped
FROM raw.ingestion_runs
ORDER BY run_id;

 run_id |    source_name    |  status   | rows_in_file | rows_inserted | rows_skipped
--------+-------------------+-----------+--------------+---------------+--------------
      1 | customers         | succeeded |         1000 |          1000 |            0
      2 | bank_transactions | succeeded |        59801 |         59801 |            0
      3 | applications      | succeeded |         1000 |          1000 |            0
      4 | customers         | succeeded |         1000 |             0 |         1000
      5 | bank_transactions | succeeded |        59801 |             0 |        59801
      6 | applications      | succeeded |         1000 |             0 |         1000

make psql is a small convenience in the Makefile: it opens psql as the project user, in the project database, with the settings from .env already loaded.

"It doesn't create duplicates" is a claim. Two loads, identical counts and a log showing every row skipped the second time is proof.

Breaking it on purpose

A tampered file. I changed one word in customers.csv and saved it, the way an Excel edit would:

LoadError: customers.csv: checksum does not match manifest.json. The file changed after it was generated (was it opened and saved in Excel?).

SELECT run_id, source_name, status, error_message
FROM raw.ingestion_runs
WHERE status = 'failed';

The failed attempt is logged with its reason, and nothing in raw.customers changed. Regenerating with the same seed produced byte-identical files, and the next load went through cleanly with nothing new inserted.

A renamed column. With the header changed from decided_at to decision_time:

LoadError: applications.csv: unexpected columns.
  expected ['application_id', 'customer_id', 'decided_at']
  found    ['application_id', 'customer_id', 'decision_time']

A broken value halfway through a file. With one amount replaced by not-a-number, the conversion fails, the whole file rolls back, and the table holds zero rows from it. All or nothing, as designed.

Part 5: The bug my tests caught

The loader passed every run by hand. Twice. Then the test suite failed:

psycopg.errors.DuplicateTable: relation "tmp_load" already exists

The temporary table was supposed to vanish when each file's transaction committed. Somehow, it was still there for the next load.

The cause was a default in psycopg, the Python driver for PostgreSQL. Out of the box, psycopg opens a transaction the moment you run any query, and keeps it open until you explicitly commit. One test had run a simple SELECT count(*) before loading. That quietly opened a transaction. From then on, every "transaction" block the loader started was actually nested inside it, as a savepoint, and nothing truly committed. The temporary table never dropped, and the second load crashed into it.

The worse part is what the error was hiding. The succeeded statuses weren't really committed either. On a real system, a later crash would have rolled back loads that everyone believed had finished, and the log would have lied.

The manual runs never showed this, because the command-line path happened not to run any plain query before its first transaction.

The fix was one setting:

with psycopg.connect(autocommit=True) as conn:

With autocommit on, each plain query commits by itself, and each with conn.transaction() block is a real transaction that commits when it ends. Then I made sure it can't happen again:

def load_all(conn, data_dir: Path, schema: str = "raw") -> list[dict]:
    if not conn.autocommit:
        raise LoadError("Connect with psycopg.connect(autocommit=True) before loading.")

The loader now refuses to run on an unsafe connection, and there's a test proving it refuses.

What I took from it:

• A manual run proves one path works. Tests try paths you didn't think of.
• Know your database driver's transaction behaviour. "Implicit transactions" is one of those defaults that's harmless until it isn't.
• When a bug is fixed, add a guard and a test, so the fix can't silently disappear later.

Part 6: Testing a database without wrecking it

Tests that touch a real database raise a fair worry: will running them damage the real data? Mine use three techniques to make sure they can't.

A throwaway schema for every test. Each test creates a schema with a random name like test_raw_3f9c1a, builds the tables inside it from the same 02_raw_tables.sql file, runs, and drops the schema afterwards:

@pytest.fixture
def schema(conn):
    """A fresh, empty copy of the raw tables in a uniquely named schema."""
    name = f"test_raw_{uuid.uuid4().hex[:6]}"
    with conn.transaction():
        conn.execute(f"CREATE SCHEMA {name}")
        conn.execute(DDL.replace("raw.", f"{name}."))
    yield name
    with conn.transaction():
        conn.execute(f"DROP SCHEMA {name} CASCADE")

It's a rehearsal on a spare stage, never the main one. Because the test tables come from the same SQL file as the real ones, the tests always check the real design. A pytest fixture is setup code that runs before a test; everything after yield runs afterwards as cleanup.

Small, fresh data for every test. Another fixture generates a town of 20 customers into a temporary folder, so tests are fast and independent of whatever is in data/generated/.

A graceful skip. If PostgreSQL isn't reachable, the database tests are skipped with the reason instead of failing:

try:
    connection = psycopg.connect(connect_timeout=3, autocommit=True)
except psycopg.OperationalError as exc:
    pytest.skip(f"PostgreSQL not reachable: {exc}")

With the database off, make test reports 12 passed, 10 skipped: the generator tests still run, and the skip tells you exactly why the others didn't.

What the ten loader tests promise

Test	The promise
test_refuses_a_connection_without_autocommit	The loader won't run on an unsafe connection
test_loads_every_row_the_manifest_promises	Database row counts match the manifest exactly
test_loading_twice_creates_no_duplicates	Idempotency
test_every_load_attempt_is_logged	Two loads mean six logged runs, all succeeded
test_every_row_points_to_the_run_that_loaded_it	Lineage: no row without a successful run behind it
test_money_survives_the_load_exactly	The database total equals the file total, to the penny
test_timestamps_keep_their_utc_meaning	A timestamp means the same moment before and after loading
test_tampered_file_is_refused_and_logged	Checksum mismatch: refused, logged, nothing loaded
test_renamed_column_is_refused	A changed header stops the load
test_bad_value_rolls_back_the_whole_file	One bad value means zero rows from that file

The money test is worth showing, because it checks the whole chain from file to database with exact decimals:

def test_money_survives_the_load_exactly(conn, schema, data_dir):
    load_all(conn, data_dir, schema)
    lines = (data_dir / "bank_transactions.csv").read_text().splitlines()[1:]
    expected = sum(Decimal(line.split(",")[2]) for line in lines)
    loaded = conn.execute(f"SELECT sum(amount) FROM {schema}.bank_transactions").fetchone()[0]
    assert loaded == expected

Decimal is Python's exact decimal type, the counterpart of PostgreSQL's NUMERIC.

Fire drills

A test that has never failed hasn't proven anything yet, so I removed protections on purpose:

Planted bug	Tests that failed
Deleted the ON CONFLICT ... DO NOTHING line	3, led by test_loading_twice_creates_no_duplicates
Disabled the checksum check	1: test_tampered_file_is_refused_and_logged

Each time, the right test caught it and its name said exactly what had broken. I also confirmed the real raw.ingestion_runs table had the same number of rows before and after a full test run, so the isolation genuinely works.

Part 7: Querying the data

With the data in PostgreSQL, the questions get interesting. These are the queries I ran, each one teaching a different SQL idea.

First: what time zone am I looking at?

SET timezone = 'UTC';

My Mac is set to Nairobi time, so by default psql shows the first transaction like this:

 transaction_id | amount  |        event_at
----------------+---------+------------------------
 TXN-00000001   | 1770.00 | 2024-02-25 09:43:00+03

After SET timezone = 'UTC':

 TXN-00000001   | 1770.00 | 2024-02-25 06:43:00+00

Same moment, different clock on the wall. TIMESTAMPTZ stores the moment; the session's time zone only controls how it's displayed. It also controls functions like date_trunc, which decides where "a month" starts, so for analysis I set UTC first.

1. How big is everything? (UNION ALL)

SELECT 'customers' AS table_name, count(*) AS row_count FROM raw.customers
UNION ALL
SELECT 'bank_transactions', count(*) FROM raw.bank_transactions
UNION ALL
SELECT 'applications', count(*) FROM raw.applications;

    table_name     | row_count
-------------------+-----------
 customers         |      1000
 bank_transactions |     59801
 applications      |      1000

UNION ALL stacks results from several queries. Plain UNION also removes duplicate rows, which costs extra work and is rarely what you want.

2. The latest successful load per source (DISTINCT ON)

SELECT DISTINCT ON (source_name)
       source_name, run_id, status, rows_inserted, rows_skipped
FROM raw.ingestion_runs
WHERE status = 'succeeded'
ORDER BY source_name, run_id DESC;

DISTINCT ON (source_name) keeps only the first row for each source, and the ORDER BY decides which one is first: the newest. It's a PostgreSQL feature. Query 6 shows the version that works in any database.

3. Money in, money out (SUM ... FILTER)

SELECT customer_id,
       SUM(amount) FILTER (WHERE amount > 0) AS money_in,
       SUM(amount) FILTER (WHERE amount < 0) AS money_out,
       SUM(amount)                           AS net
FROM raw.bank_transactions
GROUP BY customer_id
ORDER BY net DESC
LIMIT 5;

 customer_id | money_in | money_out |   net
-------------+----------+-----------+----------
 CUST-00594  | 84000.00 | -28124.89 | 55875.11
 CUST-00992  | 84000.00 | -31816.74 | 52183.26
 CUST-00679  | 80880.00 | -30035.46 | 50844.54
 CUST-00440  | 84000.00 | -33325.46 | 50674.54
 CUST-00220  | 73500.00 | -26496.69 | 47003.31

GROUP BY makes one row per customer; FILTER limits which rows each sum includes. The portable equivalent is SUM(CASE WHEN amount > 0 THEN amount END).

4. One customer, month by month (date_trunc)

SELECT date_trunc('month', event_at) AS month,
       COUNT(*)                      AS transactions,
       SUM(amount)                   AS net
FROM raw.bank_transactions
WHERE customer_id = 'CUST-00001'
GROUP BY 1
ORDER BY 1
LIMIT 4;

         month          | transactions |   net
------------------------+--------------+---------
 2024-02-01 00:00:00+00 |            1 | 1770.00
 2024-03-01 00:00:00+00 |            4 |  783.04
 2024-04-01 00:00:00+00 |            4 |  801.39
 2024-05-01 00:00:00+00 |            4 |  793.25

date_trunc('month', ...) rounds each timestamp down to the start of its month. GROUP BY 1 means "group by the first column in the SELECT." The first month has a single transaction because this customer arrived after that month's rent date, so only their salary landed.

5. Days between salaries (LAG)

SELECT event_at::date AS paid_on,
       event_at::date - LAG(event_at::date) OVER (ORDER BY event_at) AS days_since_last
FROM raw.bank_transactions
WHERE customer_id = 'CUST-00001'
  AND description LIKE 'SALARY%'
ORDER BY event_at
LIMIT 6;

  paid_on   | days_since_last
------------+-----------------
 2024-02-25 |
 2024-03-28 |              32
 2024-04-27 |              30
 2024-05-27 |              30
 2024-06-27 |              31
 2024-07-26 |              29

LAG fetches a value from the previous row, and OVER (ORDER BY event_at) defines what "previous" means. This is a window function: it calculates across neighbouring rows without collapsing them the way GROUP BY does. Subtracting two dates gives a number of days.

This exact pattern becomes the project's income regularity feature: how reliably does money arrive?

6. Each customer's latest salary (ROW_NUMBER in a CTE)

WITH ranked AS (
    SELECT customer_id, amount, event_at,
           ROW_NUMBER() OVER (PARTITION BY customer_id ORDER BY event_at DESC) AS rn
    FROM raw.bank_transactions
    WHERE description LIKE 'SALARY%'
)
SELECT customer_id, amount, event_at::date AS last_paid
FROM ranked
WHERE rn = 1
ORDER BY customer_id
LIMIT 3;

 customer_id | amount  | last_paid
-------------+---------+------------
 CUST-00001  | 1770.00 | 2025-12-26
 CUST-00002  | 2130.00 | 2025-12-28
 CUST-00003  | 2590.00 | 2025-12-27

WITH ranked AS (...) is a common table expression (CTE): a named, temporary result that turns one tangled query into readable steps. ROW_NUMBER() numbers each customer's salaries from newest to oldest, and PARTITION BY customer_id restarts the count for every customer. Keeping rn = 1 gives the latest. The same "rank, then keep number one" pattern is the standard way to remove duplicates, which this project will need once deliberately messy data arrives.

7. Solving a puzzle from the last post

In the previous post I found about 860 more salary payments than rent payments, and explained it: customers who arrive after the 1st miss that month's rent but still get that month's salary. Here's the proof in SQL:

WITH per_customer AS (
    SELECT customer_id,
           COUNT(*) FILTER (WHERE description LIKE 'SALARY%')          AS salaries,
           COUNT(*) FILTER (WHERE description = 'RENT STANDING ORDER') AS rents
    FROM raw.bank_transactions
    GROUP BY customer_id
)
SELECT salaries - rents AS extra_salaries,
       COUNT(*)         AS customers
FROM per_customer
GROUP BY 1
ORDER BY 1;

 extra_salaries | customers
----------------+-----------
              0 |       141
              1 |       859

Exactly 859 customers have one extra salary and 141 have none. Nobody has two. The explanation from the last post holds, and now it's a query anyone can rerun rather than a claim.

8. The idea that drives the whole project (a join with a time condition)

SELECT a.application_id,
       a.decided_at::date,
       COUNT(*) FILTER (WHERE t.available_at <= a.decided_at) AS known_at_decision,
       COUNT(*) FILTER (WHERE t.available_at >  a.decided_at) AS arrived_later
FROM raw.applications a
JOIN raw.bank_transactions t USING (customer_id)
GROUP BY a.application_id, a.decided_at
ORDER BY a.application_id
LIMIT 4;

 application_id | decided_at | known_at_decision | arrived_later
----------------+------------+-------------------+---------------
 APP-00001      | 2024-10-04 |                30 |            59
 APP-00002      | 2025-12-27 |                40 |             1
 APP-00003      | 2025-01-25 |                22 |            45
 APP-00004      | 2025-11-13 |                44 |             6

JOIN ... USING (customer_id) matches each application with every transaction belonging to the same customer. a and t are aliases, short names for the tables.

Look at APP-00001. When that lending decision was made, 30 of the customer's transactions were known. Another 59 arrived afterwards. A feature that accidentally used all 89 would be using the future to explain the past. It would look brilliant in testing and fail in real life.

That available_at <= decided_at condition is called point-in-time correctness, and it's the most important line in this entire project. The next stage is built around it.

9. Did my index earn its place? (EXPLAIN ANALYZE)

EXPLAIN ANALYZE
SELECT * FROM raw.bank_transactions
WHERE customer_id = 'CUST-00500'
ORDER BY event_at;

 Index Scan using ix_bank_transactions_customer_event on bank_transactions
   (actual time=0.036..0.053 rows=90 loops=1)
 Execution Time: 0.413 ms

EXPLAIN ANALYZE runs the query and shows how PostgreSQL executed it. Index Scan means it used my index to jump straight to that customer's 90 rows instead of reading all 59,801. Under half a millisecond. When a query is slow, this is the first command to reach for.

Two psql tips. \timing on shows how long every query takes. And if a long result fills the screen with a : at the bottom, you're in the pager: press q to get back.

Prefer a GUI? In VS Code, the SQLTools extension with its PostgreSQL driver runs these queries in the editor, with results in a table. DBeaver and pgAdmin work too. The SQL is identical; only the window changes.

Part 8: The Makefile, updated

Four new targets joined the Makefile:

tables:           ## create the raw landing tables and the load log
	psql -v ON_ERROR_STOP=1 -f sql/02_raw_tables.sql

load:             ## load the generated CSVs into the raw schema
	$(PYTHON) -m ingestion.load_generated

psql:             ## open psql as the project user, in the project database
	psql

reset-raw:        ## DEV ONLY: empty the raw tables and the load log
	psql -v ON_ERROR_STOP=1 -c "TRUNCATE raw.customers, raw.bank_transactions, raw.applications, raw.ingestion_runs RESTART IDENTITY"

-v ON_ERROR_STOP=1 makes psql stop at the first error instead of carrying on through a broken script. make psql works with no arguments because the Makefile loads .env and exports it, so psql finds the standard PGHOST, PGUSER and related settings automatically.

reset-raw is labelled DEV ONLY for a reason. On a real system, you never wipe the raw layer; it's the permanent record of everything that ever arrived. Locally, while practising, being able to start clean is useful. RESTART IDENTITY resets run_id numbering to 1.

Part 9: Shipping it

Same workflow as part two: a branch, small commits that each tell one part of the story, a pull request, a review of my own diff, then a squash merge.

git switch -c feat/raw-loading

git add sql/02_raw_tables.sql
git commit -m "Add raw landing tables and ingestion run log"

git add ingestion/
git commit -m "Add idempotent CSV loader with checksum, header and transaction safety"

git add tests/ingestion/
git commit -m "Add database tests for loading, idempotency, lineage and failure handling"

git add -A
git commit -m "Add make tables, load, psql and reset-raw targets"

git push -u origin feat/raw-loading
gh pr create --title "Load generated data into the raw schema" --body "..."
gh pr merge --squash --delete-branch
git switch main
git pull

Before every commit, git status confirms that data/generated/ never appears. Generated data stays out of Git because anyone can recreate it exactly with make seed.

Notes for Windows and Linux

Linux (Ubuntu, Debian, WSL): install PostgreSQL from the official instructions and the client with sudo apt install postgresql-client. Everything in this post works unchanged once psql is on your PATH.

Windows: the smoothest path is WSL with the VS Code WSL extension, then the Linux notes above. make and the .env loading in the Makefile assume a Unix shell. If you use native Windows PostgreSQL, the SQL itself is identical; run it from pgAdmin or VS Code's SQLTools.

Time zones: whatever your machine's zone, SET timezone = 'UTC'; at the start of a psql session makes the output match the files.

Traps worth knowing

You see	What it means	Fix
relation "raw.bank_transactions" does not exist	Tables weren't created	make tables
LoadError: No manifest at ...	Nothing has been generated yet	make seed
LoadError: ... checksum does not match	A file changed after it was generated	Regenerate with make seed, then make load
LoadError: ... unexpected columns	A file's header changed	Regenerate, or update the loader's SOURCES if the change is intentional
relation "tmp_load" already exists	An implicit transaction is wrapping your loads	Connect with autocommit=True
connection refused	PostgreSQL isn't running	Start the service
Timestamps show +03 (or your local offset)	The session is in local time	SET timezone = 'UTC';
psql shows : and won't take input	You're in the pager	Press q
Leftover test_raw_... schemas	A test run was killed mid-way	DROP SCHEMA test_raw_xxxxxx CASCADE; for each
Tests report 10 skipped	The tests can't reach the database	Start PostgreSQL and check the connection settings

The lessons I'm keeping

• Land untrusted data as text in a temporary table, then convert on the way into the real table.
• Bulk load with COPY, never row by row.
• A primary key plus ON CONFLICT DO NOTHING makes a load idempotent.
• Put each file's rows and its status in one transaction: all or nothing.
• Log the attempt first, in its own committed transaction, so failures leave a record.
• Check the fingerprint and the header before loading anything. Fail loudly, with helpful messages.
• Add lineage columns so every row traces back to the load that brought it in.
• Record what arrives in raw; test relationships later rather than enforcing them at the door.
• NUMERIC for money, TIMESTAMPTZ for moments.
• A timestamp's display depends on the session time zone; its meaning doesn't.
• Use placeholders for values. Never paste values into SQL text.
• Index the columns you filter and sort by most, then confirm it with EXPLAIN ANALYZE.
• Know your database driver's transaction defaults. Use autocommit with explicit transactions.
• Database tests need isolation and a graceful skip.
• Never open source CSVs in Excel and save them.
• When tests catch what manual runs missed, believe the tests, then add a guard so it can't come back.

A checklist for loading any new source

At work or on the next project, this is the sequence I'll follow, whether the source is a file, an API or another database:

1. Decide the primary key. If the source doesn't have a stable one, that's the first problem to solve.
2. Write the landing table: real types, NOT NULL where it's genuinely required, lineage columns, no foreign keys.
3. Create a load log table before writing the loader.
4. Validate before loading: fingerprint or version, expected columns, row count.
5. Land raw data into a temporary table, then insert with conversion and ON CONFLICT.
6. Wrap each batch's data and status in one transaction.
7. Log failures with the real error message.
8. Prove idempotency by loading twice and comparing counts.
9. Break it on purpose: bad file, changed shape, bad value. Confirm each is refused and logged.
10. Test against an isolated schema, and skip cleanly when the database isn't available.

What's next

The data is in the database, but it's still raw. The next stage builds the first transformations with dbt: a staging layer that cleans and standardises, then the first real feature, income regularity, built on the LAG pattern from query 5 and the available_at <= decided_at rule from query 8. For the first time, one customer's transactions will flow all the way from a file to a feature served over an API.